lq/nl-im-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

微信小程序

Browse Source
This commit is contained in:
liqi
2025-12-15 09:04:14 +08:00
parent 4f2b0bb213
commit 2485cd3ac9
12 changed files with 2622 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

419
internal/api/call_handler.go Normal file
View File

@@ -0,0 +1,419 @@
/**
* package api
*
* 通话相关 API 处理器
* 功能：
* 1. 创建/加入/离开通话房间
* 2. 处理 WebRTC 信令（offer/answer/ice）
* 3. 生成 RTMP 推拉流地址（小程序）
*/
package api
import (
"fmt"
"xk-websocket-v2/internal/mediaserver"
"xk-websocket-v2/internal/utils"
"github.com/gin-gonic/gin"
"github.com/spf13/viper"
)
// ========== 请求/响应结构体 ==========
// CreateCallRoomRequest 创建通话房间请求
type CreateCallRoomRequest struct {
RoomID string `json:"room_id" binding:"required"`
CallType string `json:"call_type" binding:"required,oneof=audio video"` // audio/video
IsGroupCall bool `json:"is_group_call"`
}
// JoinCallRoomRequest 加入通话房间请求
type JoinCallRoomRequest struct {
RoomID string `json:"room_id" binding:"required"`
UserID string `json:"user_id" binding:"required"`
Platform string `json:"platform" binding:"required,oneof=h5 app miniprogram"` // h5/app/miniprogram
}
// LeaveCallRoomRequest 离开通话房间请求
type LeaveCallRoomRequest struct {
RoomID string `json:"room_id" binding:"required"`
UserID string `json:"user_id" binding:"required"`
}
// WebRTCOfferRequest WebRTC Offer 请求
type WebRTCOfferRequest struct {
RoomID string `json:"room_id" binding:"required"`
UserID string `json:"user_id" binding:"required"`
SDP string `json:"sdp" binding:"required"`
}
// WebRTCICERequest WebRTC ICE 候选请求
type WebRTCICERequest struct {
RoomID string `json:"room_id" binding:"required"`
UserID string `json:"user_id" binding:"required"`
Candidate string `json:"candidate" binding:"required"`
}
// JoinCallRoomResponse 加入通话房间响应
type JoinCallRoomResponse struct {
RoomID string `json:"room_id"`
Platform string `json:"platform"`
ICEServers []ICEServerConfig `json:"ice_servers,omitempty"` // H5/App 用
PushURL string `json:"push_url,omitempty"` // 小程序用
PullURLs []PullURLInfo `json:"pull_urls,omitempty"` // 小程序用
Participants []ParticipantInfo `json:"participants"`
}
// ICEServerConfig ICE 服务器配置
type ICEServerConfig struct {
URLs []string `json:"urls"`
Username string `json:"username,omitempty"`
Credential string `json:"credential,omitempty"`
}
// PullURLInfo 拉流地址信息
type PullURLInfo struct {
UserID string `json:"user_id"`
URL string `json:"url"`
FLVURL string `json:"flv_url,omitempty"`
}
// ParticipantInfo 参与者信息
type ParticipantInfo struct {
UserID string `json:"user_id"`
Platform string `json:"platform"`
HasAudio bool `json:"has_audio"`
HasVideo bool `json:"has_video"`
}
// RoomInfoResponse 房间信息响应
type RoomInfoResponse struct {
RoomID string `json:"room_id"`
CallType string `json:"call_type"`
IsGroupCall bool `json:"is_group_call"`
ParticipantCount int `json:"participant_count"`
Participants []ParticipantInfo `json:"participants"`
}
// ========== API 处理函数 ==========
// CreateCallRoomHandler 创建通话房间
// POST /api/call/room
func CreateCallRoomHandler(c *gin.Context) {
var req CreateCallRoomRequest
if err := c.ShouldBindJSON(&req); err != nil {
utils.BadRequest(c, "参数错误: "+err.Error())
return
}
ms := mediaserver.GetServer()
if ms.GetConfig() == nil || !ms.GetConfig().Enabled {
utils.Error(c, 503, "媒体服务未启用")
return
}
// 创建房间
room := ms.GetOrCreateRoom(req.RoomID)
room.SetCallType(req.CallType, req.IsGroupCall)
utils.SuccessWithData(c, gin.H{
"room_id": req.RoomID,
"call_type": req.CallType,
"is_group": req.IsGroupCall,
"created": true,
}, "创建成功")
}
// JoinCallRoomHandler 加入通话房间
// POST /api/call/join
func JoinCallRoomHandler(c *gin.Context) {
var req JoinCallRoomRequest
if err := c.ShouldBindJSON(&req); err != nil {
utils.BadRequest(c, "参数错误: "+err.Error())
return
}
ms := mediaserver.GetServer()
if ms.GetConfig() == nil || !ms.GetConfig().Enabled {
utils.Error(c, 503, "媒体服务未启用")
return
}
room := ms.GetRoom(req.RoomID)
if room == nil {
// 自动创建房间
room = ms.GetOrCreateRoom(req.RoomID)
}
response := JoinCallRoomResponse{
RoomID: req.RoomID,
Platform: req.Platform,
}
switch req.Platform {
case "h5", "app":
// H5/App 使用 WebRTC
pType := mediaserver.ParticipantTypeWebRTC
_, err := room.AddParticipant(req.UserID, pType)
if err != nil {
utils.BadRequest(c, err.Error())
return
}
// 返回 ICE 服务器配置
response.ICEServers = getICEServers(req.UserID)
case "miniprogram":
// 小程序使用 RTMP
pType := mediaserver.ParticipantTypeRTMP
participant, err := room.AddParticipant(req.UserID, pType)
if err != nil {
utils.BadRequest(c, err.Error())
return
}
// 生成推拉流地址
rtmpServer := ms.GetRTMP()
if rtmpServer == nil {
utils.Error(c, 503, "RTMP服务未启用")
return
}
stream, err := rtmpServer.GenerateStreamURLs(req.RoomID, req.UserID)
if err != nil {
utils.InternalError(c, "生成推流地址失败")
return
}
// 更新参与者的流地址
room.SetParticipantRTMPURLs(req.UserID, stream.PushURL, stream.PullURL, stream.ID)
participant.PushURL = stream.PushURL
participant.PullURL = stream.PullURL
participant.StreamID = stream.ID
response.PushURL = stream.PushURL
// 获取房间内其他用户的拉流地址
pullURLs := make([]PullURLInfo, 0)
for _, p := range room.GetOtherParticipants(req.UserID) {
if p.PullURL != "" {
pullURLs = append(pullURLs, PullURLInfo{
UserID: p.UserID,
URL: p.PullURL,
})
}
}
response.PullURLs = pullURLs
}
// 获取参与者列表
participants := make([]ParticipantInfo, 0)
for _, p := range room.GetAllParticipants() {
platform := "h5"
if p.Type == mediaserver.ParticipantTypeRTMP {
platform = "miniprogram"
}
participants = append(participants, ParticipantInfo{
UserID: p.UserID,
Platform: platform,
HasAudio: p.HasAudio,
HasVideo: p.HasVideo,
})
}
response.Participants = participants
utils.SuccessWithData(c, response, "加入成功")
}
// LeaveCallRoomHandler 离开通话房间
// POST /api/call/leave
func LeaveCallRoomHandler(c *gin.Context) {
var req LeaveCallRoomRequest
if err := c.ShouldBindJSON(&req); err != nil {
utils.BadRequest(c, "参数错误: "+err.Error())
return
}
ms := mediaserver.GetServer()
room := ms.GetRoom(req.RoomID)
if room == nil {
utils.SuccessWithData(c, gin.H{"left": true}, "已离开")
return
}
// 获取参与者信息
participant := room.GetParticipant(req.UserID)
if participant != nil && participant.Type == mediaserver.ParticipantTypeRTMP {
// 清理 RTMP 流
rtmpServer := ms.GetRTMP()
if rtmpServer != nil && participant.StreamID != "" {
rtmpServer.RemoveStream(participant.StreamID)
}
}
// 移除参与者
room.RemoveParticipant(req.UserID)
// 如果房间空了，移除房间
if room.IsEmpty() {
ms.RemoveRoom(req.RoomID)
}
utils.SuccessWithData(c, gin.H{"left": true}, "已离开")
}
// GetCallRoomHandler 获取通话房间信息
// GET /api/call/room/:room_id
func GetCallRoomHandler(c *gin.Context) {
roomID := c.Param("room_id")
if roomID == "" {
utils.BadRequest(c, "房间ID不能为空")
return
}
ms := mediaserver.GetServer()
room := ms.GetRoom(roomID)
if room == nil {
utils.NotFound(c, "房间不存在")
return
}
info := room.GetInfo()
participants := make([]ParticipantInfo, 0)
for _, p := range info.Participants {
platform := "h5"
if p.Type == mediaserver.ParticipantTypeRTMP {
platform = "miniprogram"
}
participants = append(participants, ParticipantInfo{
UserID: p.UserID,
Platform: platform,
HasAudio: p.HasAudio,
HasVideo: p.HasVideo,
})
}
utils.SuccessWithData(c, RoomInfoResponse{
RoomID: info.ID,
CallType: info.CallType,
IsGroupCall: info.IsGroupCall,
ParticipantCount: info.ParticipantCount,
Participants: participants,
}, "获取成功")
}
// WebRTCOfferHandler 处理 WebRTC Offer
// POST /api/call/offer
func WebRTCOfferHandler(c *gin.Context) {
var req WebRTCOfferRequest
if err := c.ShouldBindJSON(&req); err != nil {
utils.BadRequest(c, "参数错误: "+err.Error())
return
}
ms := mediaserver.GetServer()
sfu := ms.GetSFU()
if sfu == nil {
utils.Error(c, 503, "SFU服务未启用")
return
}
// 处理 Offer 并返回 Answer
answerSDP, err := sfu.HandleOffer(req.RoomID, req.UserID, req.SDP)
if err != nil {
utils.InternalError(c, "处理Offer失败: "+err.Error())
return
}
utils.SuccessWithData(c, gin.H{
"sdp": answerSDP,
}, "处理成功")
}
// WebRTCICEHandler 处理 WebRTC ICE 候选
// POST /api/call/ice
func WebRTCICEHandler(c *gin.Context) {
var req WebRTCICERequest
if err := c.ShouldBindJSON(&req); err != nil {
utils.BadRequest(c, "参数错误: "+err.Error())
return
}
ms := mediaserver.GetServer()
sfu := ms.GetSFU()
if sfu == nil {
utils.Error(c, 503, "SFU服务未启用")
return
}
// 添加 ICE 候选
if err := sfu.HandleICECandidate(req.RoomID, req.UserID, req.Candidate); err != nil {
utils.InternalError(c, "处理ICE失败: "+err.Error())
return
}
utils.SuccessWithData(c, gin.H{"added": true}, "添加成功")
}
// GetICEServersHandler 获取 ICE 服务器配置
// GET /api/call/ice-servers
func GetICEServersHandler(c *gin.Context) {
userID := c.Query("user_id")
if userID == "" {
userID = "anonymous"
}
servers := getICEServers(userID)
utils.SuccessWithData(c, gin.H{
"ice_servers": servers,
}, "获取成功")
}
// getICEServers 获取 ICE 服务器配置
func getICEServers(userID string) []ICEServerConfig {
servers := []ICEServerConfig{}
turnPublicIP := viper.GetString("turn.public_ip")
turnPort := viper.GetInt("turn.listen_port")
if turnPublicIP != "" && turnPort > 0 {
portStr := fmt.Sprintf("%d", turnPort)
// STUN 服务器
servers = append(servers, ICEServerConfig{
URLs: []string{
"stun:" + turnPublicIP + ":" + portStr,
},
})
// TURN 服务器（如果启用）
if viper.GetBool("turn.enabled") {
username, credential := mediaserver.GetTURNCredentials(userID)
servers = append(servers, ICEServerConfig{
URLs: []string{
"turn:" + turnPublicIP + ":" + portStr,
},
Username: username,
Credential: credential,
})
}
}
return servers
}
// RegisterCallRoutes 注册通话相关路由
func RegisterCallRoutes(router *gin.RouterGroup) {
callGroup := router.Group("/call")
{
callGroup.POST("/room", CreateCallRoomHandler)
callGroup.GET("/room/:room_id", GetCallRoomHandler)
callGroup.POST("/join", JoinCallRoomHandler)
callGroup.POST("/leave", LeaveCallRoomHandler)
callGroup.POST("/offer", WebRTCOfferHandler)
callGroup.POST("/ice", WebRTCICEHandler)
callGroup.GET("/ice-servers", GetICEServersHandler)
}
}

183
internal/api/media_handler.go Normal file
View File

@@ -0,0 +1,183 @@
/**
* package api
*
* 媒体服务监控 API 处理器
* 功能：
* 1. 查看媒体服务器状态
* 2. 查看活跃房间和流
* 3. 管理桥接器
*/
package api
import (
"xk-websocket-v2/internal/mediaserver"
"xk-websocket-v2/internal/utils"
"github.com/gin-gonic/gin"
)
// MediaServerStatusResponse 媒体服务器状态响应
type MediaServerStatusResponse struct {
Enabled bool `json:"enabled"`
Config map[string]interface{} `json:"config"`
RoomCount int `json:"room_count"`
SFUTrackCount int `json:"sfu_track_count"`
RTMPStreamCount int `json:"rtmp_stream_count"`
}
// GetMediaServerStatusHandler 获取媒体服务器状态
// GET /api/media/status
func GetMediaServerStatusHandler(c *gin.Context) {
ms := mediaserver.GetServer()
config := ms.GetConfig()
if config == nil {
utils.Error(c, 503, "媒体服务未配置")
return
}
response := MediaServerStatusResponse{
Enabled: config.Enabled,
RoomCount: ms.GetRoomCount(),
Config: map[string]interface{}{
"public_ip": config.PublicIP,
"webrtc_port": config.WebRTCPort,
"rtmp_port": config.RTMPPort,
"http_flv_port": config.HTTPFLVPort,
"max_room_size": config.MaxRoomSize,
},
}
// 获取 SFU 轨道数
if sfu := ms.GetSFU(); sfu != nil {
response.SFUTrackCount = sfu.GetTrackCount()
}
// 获取 RTMP 流数
if rtmp := ms.GetRTMP(); rtmp != nil {
response.RTMPStreamCount = rtmp.GetStreamCount()
}
utils.SuccessWithData(c, response, "获取成功")
}
// GetAllRoomsHandler 获取所有房间
// GET /api/media/rooms
func GetAllRoomsHandler(c *gin.Context) {
ms := mediaserver.GetServer()
rooms := ms.GetAllRooms()
utils.SuccessWithData(c, gin.H{
"count": len(rooms),
"rooms": rooms,
}, "获取成功")
}
// GetRoomDetailHandler 获取房间详情
// GET /api/media/rooms/:room_id
func GetRoomDetailHandler(c *gin.Context) {
roomID := c.Param("room_id")
if roomID == "" {
utils.BadRequest(c, "房间ID不能为空")
return
}
ms := mediaserver.GetServer()
room := ms.GetRoom(roomID)
if room == nil {
utils.NotFound(c, "房间不存在")
return
}
info := room.GetInfo()
// 获取桥接信息
bm := mediaserver.GetBridgeManager()
bridge := bm.GetBridge(roomID)
var bridgeInfo map[string]interface{}
if bridge != nil && bridge.IsRunning() {
w2r, r2w := bridge.GetAllConverters()
w2rCount, r2wCount := bridge.GetConverterCount()
bridgeInfo = map[string]interface{}{
"running": true,
"webrtc_to_rtmp": w2rCount,
"rtmp_to_webrtc": r2wCount,
"w2r_converters": w2r,
"r2w_converters": r2w,
}
}
utils.SuccessWithData(c, gin.H{
"room": info,
"bridge": bridgeInfo,
}, "获取成功")
}
// GetRTMPStreamsHandler 获取 RTMP 流列表
// GET /api/media/rtmp/streams
func GetRTMPStreamsHandler(c *gin.Context) {
ms := mediaserver.GetServer()
rtmp := ms.GetRTMP()
if rtmp == nil {
utils.Error(c, 503, "RTMP服务未启用")
return
}
roomID := c.Query("room_id")
var streams []*mediaserver.RTMPStream
if roomID != "" {
streams = rtmp.GetStreamsByRoom(roomID)
} else {
// 获取所有流（简化实现：只返回数量）
utils.SuccessWithData(c, gin.H{
"count": rtmp.GetStreamCount(),
}, "获取成功")
return
}
utils.SuccessWithData(c, gin.H{
"count": len(streams),
"streams": streams,
}, "获取成功")
}
// CloseRoomHandler 关闭房间
// POST /api/media/rooms/:room_id/close
func CloseRoomHandler(c *gin.Context) {
roomID := c.Param("room_id")
if roomID == "" {
utils.BadRequest(c, "房间ID不能为空")
return
}
ms := mediaserver.GetServer()
room := ms.GetRoom(roomID)
if room == nil {
utils.NotFound(c, "房间不存在")
return
}
// 移除桥接
bm := mediaserver.GetBridgeManager()
bm.RemoveBridge(roomID)
// 移除房间
ms.RemoveRoom(roomID)
utils.Success(c, "房间已关闭")
}
// RegisterMediaRoutes 注册媒体相关路由
func RegisterMediaRoutes(router *gin.RouterGroup) {
mediaGroup := router.Group("/media")
{
mediaGroup.GET("/status", GetMediaServerStatusHandler)
mediaGroup.GET("/rooms", GetAllRoomsHandler)
mediaGroup.GET("/rooms/:room_id", GetRoomDetailHandler)
mediaGroup.POST("/rooms/:room_id/close", CloseRoomHandler)
mediaGroup.GET("/rtmp/streams", GetRTMPStreamsHandler)
}
}

482
internal/mediaserver/bridge.go Normal file
View File

@@ -0,0 +1,482 @@
/**
* package mediaserver
*
* WebRTC ↔ RTMP 媒体桥接
* 功能：
* 1. WebRTC 流转 RTMP（H5用户 -> 小程序用户）
* 2. RTMP 流转 WebRTC（小程序用户 -> H5用户）
* 3. 音频转码：Opus ↔ AAC
* 4. 视频透传：H.264
*/
package mediaserver
import (
"fmt"
"log"
"sync"
"time"
"github.com/pion/rtp"
"github.com/pion/webrtc/v3"
)
// Bridge WebRTC-RTMP 媒体桥
type Bridge struct {
roomID string
mu sync.RWMutex
running bool
stopChan chan struct{}
// WebRTC -> RTMP 转换器
webrtcToRTMP map[string]*WebRTCToRTMPConverter
// RTMP -> WebRTC 转换器
rtmpToWebRTC map[string]*RTMPToWebRTCConverter
}
// WebRTCToRTMPConverter WebRTC 转 RTMP 转换器
type WebRTCToRTMPConverter struct {
UserID string
StreamID string
RoomID string
Running bool
stopChan chan struct{}
// 输入: WebRTC Track
VideoTrack *webrtc.TrackRemote
AudioTrack *webrtc.TrackRemote
// 输出: RTMP 流信息
OutputURL string // RTMP 推流地址
// 编解码器
videoCodec string // h264
audioCodec string // opus -> aac
// 缓冲区
videoBuffer chan *rtp.Packet
audioBuffer chan *rtp.Packet
}
// RTMPToWebRTCConverter RTMP 转 WebRTC 转换器
type RTMPToWebRTCConverter struct {
UserID string
StreamID string
RoomID string
Running bool
stopChan chan struct{}
// 输入: RTMP 拉流地址
InputURL string
// 输出: WebRTC Track
VideoTrack *webrtc.TrackLocalStaticRTP
AudioTrack *webrtc.TrackLocalStaticRTP
// 编解码器
videoCodec string // h264
audioCodec string // aac -> opus
}
// BridgeManager 桥接管理器
type BridgeManager struct {
bridges map[string]*Bridge // roomID -> bridge
mu sync.RWMutex
}
// 全局桥接管理器
var bridgeManager *BridgeManager
var bridgeOnce sync.Once
// GetBridgeManager 获取桥接管理器单例
func GetBridgeManager() *BridgeManager {
bridgeOnce.Do(func() {
bridgeManager = &BridgeManager{
bridges: make(map[string]*Bridge),
}
})
return bridgeManager
}
// GetOrCreateBridge 获取或创建房间的桥接
func (bm *BridgeManager) GetOrCreateBridge(roomID string) *Bridge {
bm.mu.Lock()
defer bm.mu.Unlock()
if bridge, exists := bm.bridges[roomID]; exists {
return bridge
}
bridge := NewBridge(roomID)
bm.bridges[roomID] = bridge
return bridge
}
// RemoveBridge 移除房间的桥接
func (bm *BridgeManager) RemoveBridge(roomID string) {
bm.mu.Lock()
defer bm.mu.Unlock()
if bridge, exists := bm.bridges[roomID]; exists {
bridge.Stop()
delete(bm.bridges, roomID)
}
}
// GetBridge 获取桥接
func (bm *BridgeManager) GetBridge(roomID string) *Bridge {
bm.mu.RLock()
defer bm.mu.RUnlock()
return bm.bridges[roomID]
}
// NewBridge 创建新桥接
func NewBridge(roomID string) *Bridge {
return &Bridge{
roomID: roomID,
stopChan: make(chan struct{}),
webrtcToRTMP: make(map[string]*WebRTCToRTMPConverter),
rtmpToWebRTC: make(map[string]*RTMPToWebRTCConverter),
}
}
// Start 启动桥接
func (b *Bridge) Start() {
b.mu.Lock()
defer b.mu.Unlock()
if b.running {
return
}
b.running = true
log.Printf("🌉 [Bridge] 启动房间桥接 | Room:%s", b.roomID)
}
// Stop 停止桥接
func (b *Bridge) Stop() {
b.mu.Lock()
defer b.mu.Unlock()
if !b.running {
return
}
// 关闭停止信号
close(b.stopChan)
// 停止所有 WebRTC -> RTMP 转换器
for _, converter := range b.webrtcToRTMP {
converter.Stop()
}
b.webrtcToRTMP = make(map[string]*WebRTCToRTMPConverter)
// 停止所有 RTMP -> WebRTC 转换器
for _, converter := range b.rtmpToWebRTC {
converter.Stop()
}
b.rtmpToWebRTC = make(map[string]*RTMPToWebRTCConverter)
b.running = false
log.Printf("🌉 [Bridge] 停止房间桥接 | Room:%s", b.roomID)
}
// AddWebRTCToRTMP 添加 WebRTC 到 RTMP 的转换
func (b *Bridge) AddWebRTCToRTMP(userID string, videoTrack, audioTrack *webrtc.TrackRemote, outputURL string) (*WebRTCToRTMPConverter, error) {
b.mu.Lock()
defer b.mu.Unlock()
if !b.running {
b.running = true
}
streamID := fmt.Sprintf("w2r_%s_%s_%d", b.roomID, userID, time.Now().UnixNano())
converter := &WebRTCToRTMPConverter{
UserID: userID,
StreamID: streamID,
RoomID: b.roomID,
Running: true,
stopChan: make(chan struct{}),
VideoTrack: videoTrack,
AudioTrack: audioTrack,
OutputURL: outputURL,
videoCodec: "h264",
audioCodec: "opus",
videoBuffer: make(chan *rtp.Packet, 100),
audioBuffer: make(chan *rtp.Packet, 100),
}
b.webrtcToRTMP[userID] = converter
// 启动转换协程
go converter.Run()
log.Printf("🌉 [Bridge] 添加 WebRTC->RTMP | Room:%s User:%s Stream:%s", b.roomID, userID, streamID)
return converter, nil
}
// AddRTMPToWebRTC 添加 RTMP 到 WebRTC 的转换
func (b *Bridge) AddRTMPToWebRTC(userID string, inputURL string) (*RTMPToWebRTCConverter, error) {
b.mu.Lock()
defer b.mu.Unlock()
if !b.running {
b.running = true
}
streamID := fmt.Sprintf("r2w_%s_%s_%d", b.roomID, userID, time.Now().UnixNano())
// 创建本地轨道
videoTrack, err := webrtc.NewTrackLocalStaticRTP(
webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264},
fmt.Sprintf("video_%s", streamID),
fmt.Sprintf("stream_%s", userID),
)
if err != nil {
return nil, fmt.Errorf("create video track failed: %w", err)
}
audioTrack, err := webrtc.NewTrackLocalStaticRTP(
webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus},
fmt.Sprintf("audio_%s", streamID),
fmt.Sprintf("stream_%s", userID),
)
if err != nil {
return nil, fmt.Errorf("create audio track failed: %w", err)
}
converter := &RTMPToWebRTCConverter{
UserID: userID,
StreamID: streamID,
RoomID: b.roomID,
Running: true,
stopChan: make(chan struct{}),
InputURL: inputURL,
VideoTrack: videoTrack,
AudioTrack: audioTrack,
videoCodec: "h264",
audioCodec: "aac",
}
b.rtmpToWebRTC[userID] = converter
// 启动转换协程
go converter.Run()
log.Printf("🌉 [Bridge] 添加 RTMP->WebRTC | Room:%s User:%s Stream:%s", b.roomID, userID, streamID)
return converter, nil
}
// RemoveWebRTCToRTMP 移除 WebRTC 到 RTMP 的转换
func (b *Bridge) RemoveWebRTCToRTMP(userID string) {
b.mu.Lock()
defer b.mu.Unlock()
if converter, exists := b.webrtcToRTMP[userID]; exists {
converter.Stop()
delete(b.webrtcToRTMP, userID)
log.Printf("🌉 [Bridge] 移除 WebRTC->RTMP | Room:%s User:%s", b.roomID, userID)
}
}
// RemoveRTMPToWebRTC 移除 RTMP 到 WebRTC 的转换
func (b *Bridge) RemoveRTMPToWebRTC(userID string) {
b.mu.Lock()
defer b.mu.Unlock()
if converter, exists := b.rtmpToWebRTC[userID]; exists {
converter.Stop()
delete(b.rtmpToWebRTC, userID)
log.Printf("🌉 [Bridge] 移除 RTMP->WebRTC | Room:%s User:%s", b.roomID, userID)
}
}
// GetRTMPToWebRTCTracks 获取 RTMP->WebRTC 转换器的输出轨道
func (b *Bridge) GetRTMPToWebRTCTracks(userID string) (*webrtc.TrackLocalStaticRTP, *webrtc.TrackLocalStaticRTP) {
b.mu.RLock()
defer b.mu.RUnlock()
if converter, exists := b.rtmpToWebRTC[userID]; exists {
return converter.VideoTrack, converter.AudioTrack
}
return nil, nil
}
// --- WebRTCToRTMPConverter 方法 ---
// Run 执行 WebRTC 到 RTMP 的转换
func (c *WebRTCToRTMPConverter) Run() {
log.Printf("🌉 [W2R] 转换开始 | Stream:%s", c.StreamID)
defer log.Printf("🌉 [W2R] 转换结束 | Stream:%s", c.StreamID)
// 启动视频读取
if c.VideoTrack != nil {
go c.readVideoTrack()
}
// 启动音频读取
if c.AudioTrack != nil {
go c.readAudioTrack()
}
// 启动 RTMP 推流
go c.pushToRTMP()
// 等待停止信号
<-c.stopChan
}
// readVideoTrack 读取视频轨道
func (c *WebRTCToRTMPConverter) readVideoTrack() {
buf := make([]byte, 1500)
for c.Running {
n, _, err := c.VideoTrack.Read(buf)
if err != nil {
if c.Running {
log.Printf("⚠️ [W2R] 读取视频失败: %v", err)
}
return
}
// 解析 RTP 包
packet := &rtp.Packet{}
if err := packet.Unmarshal(buf[:n]); err != nil {
continue
}
select {
case c.videoBuffer <- packet:
default:
// 缓冲区满，丢弃
}
}
}
// readAudioTrack 读取音频轨道
func (c *WebRTCToRTMPConverter) readAudioTrack() {
buf := make([]byte, 1500)
for c.Running {
n, _, err := c.AudioTrack.Read(buf)
if err != nil {
if c.Running {
log.Printf("⚠️ [W2R] 读取音频失败: %v", err)
}
return
}
// 解析 RTP 包
packet := &rtp.Packet{}
if err := packet.Unmarshal(buf[:n]); err != nil {
continue
}
select {
case c.audioBuffer <- packet:
default:
// 缓冲区满，丢弃
}
}
}
// pushToRTMP 推送到 RTMP
func (c *WebRTCToRTMPConverter) pushToRTMP() {
// TODO: 实际的 RTMP 推流实现
// 1. 连接到 RTMP 服务器
// 2. 从 videoBuffer/audioBuffer 读取 RTP 包
// 3. 解封装 RTP，获取裸数据
// 4. 视频: H.264 NAL 单元 -> FLV Tag
// 5. 音频: Opus -> AAC (需要转码)
// 6. 发送 FLV Tag 到 RTMP 服务器
for c.Running {
select {
case <-c.stopChan:
return
case videoPacket := <-c.videoBuffer:
// 处理视频包
_ = videoPacket
case audioPacket := <-c.audioBuffer:
// 处理音频包
_ = audioPacket
}
}
}
// Stop 停止转换
func (c *WebRTCToRTMPConverter) Stop() {
if !c.Running {
return
}
c.Running = false
close(c.stopChan)
}
// --- RTMPToWebRTCConverter 方法 ---
// Run 执行 RTMP 到 WebRTC 的转换
func (c *RTMPToWebRTCConverter) Run() {
log.Printf("🌉 [R2W] 转换开始 | Stream:%s URL:%s", c.StreamID, c.InputURL)
defer log.Printf("🌉 [R2W] 转换结束 | Stream:%s", c.StreamID)
// TODO: 实际的 RTMP 拉流实现
// 1. 连接到 RTMP 服务器
// 2. 拉取 FLV 流
// 3. 解析 FLV Tag
// 4. 视频: H.264 NAL 单元 -> RTP 包
// 5. 音频: AAC -> Opus (需要转码)
// 6. 写入 WebRTC Track
// 等待停止信号
<-c.stopChan
}
// Stop 停止转换
func (c *RTMPToWebRTCConverter) Stop() {
if !c.Running {
return
}
c.Running = false
close(c.stopChan)
}
// GetConverterCount 获取转换器数量
func (b *Bridge) GetConverterCount() (int, int) {
b.mu.RLock()
defer b.mu.RUnlock()
return len(b.webrtcToRTMP), len(b.rtmpToWebRTC)
}
// IsRunning 是否运行中
func (b *Bridge) IsRunning() bool {
b.mu.RLock()
defer b.mu.RUnlock()
return b.running
}
// GetAllConverters 获取所有转换器信息
func (b *Bridge) GetAllConverters() ([]map[string]interface{}, []map[string]interface{}) {
b.mu.RLock()
defer b.mu.RUnlock()
w2r := make([]map[string]interface{}, 0, len(b.webrtcToRTMP))
for _, c := range b.webrtcToRTMP {
w2r = append(w2r, map[string]interface{}{
"user_id": c.UserID,
"stream_id": c.StreamID,
"output_url": c.OutputURL,
"running": c.Running,
})
}
r2w := make([]map[string]interface{}, 0, len(b.rtmpToWebRTC))
for _, c := range b.rtmpToWebRTC {
r2w = append(r2w, map[string]interface{}{
"user_id": c.UserID,
"stream_id": c.StreamID,
"input_url": c.InputURL,
"running": c.Running,
})
}
return w2r, r2w
}

35
internal/mediaserver/errors.go Normal file
View File

@@ -0,0 +1,35 @@
/**
* package mediaserver
*
* 错误定义
*/
package mediaserver
import "errors"
var (
// 房间相关错误
ErrRoomNotFound = errors.New("room not found")
ErrRoomFull = errors.New("room is full")
ErrRoomClosed = errors.New("room is closed")
ErrAlreadyInRoom = errors.New("user already in room")
ErrNotInRoom = errors.New("user not in room")
// 参与者相关错误
ErrParticipantNotFound = errors.New("participant not found")
ErrInvalidParticipantType = errors.New("invalid participant type")
// 媒体相关错误
ErrMediaServerDisabled = errors.New("media server is disabled")
ErrSFUNotReady = errors.New("SFU server is not ready")
ErrRTMPNotReady = errors.New("RTMP server is not ready")
ErrStreamNotFound = errors.New("stream not found")
ErrInvalidOffer = errors.New("invalid SDP offer")
ErrInvalidAnswer = errors.New("invalid SDP answer")
ErrICEConnectionFailed = errors.New("ICE connection failed")
// 编解码相关错误
ErrUnsupportedCodec = errors.New("unsupported codec")
ErrTranscodeFailed = errors.New("transcode failed")
)

299
internal/mediaserver/room.go Normal file
View File

@@ -0,0 +1,299 @@
/**
* package mediaserver
*
* 通话房间管理
* 功能：
* 1. 管理房间内的参与者
* 2. 管理 WebRTC 和 RTMP 流
* 3. 处理用户加入/离开
*/
package mediaserver
import (
"log"
"sync"
"time"
)
// ParticipantType 参与者类型
type ParticipantType int
const (
ParticipantTypeWebRTC ParticipantType = iota // H5/App WebRTC 用户
ParticipantTypeRTMP // 小程序 RTMP 用户
)
// Participant 房间参与者
type Participant struct {
UserID string `json:"user_id"`
Type ParticipantType `json:"type"`
JoinedAt time.Time `json:"joined_at"`
HasAudio bool `json:"has_audio"`
HasVideo bool `json:"has_video"`
// WebRTC 相关
PeerConnection interface{} `json:"-"` // *webrtc.PeerConnection
// RTMP 相关
PushURL string `json:"push_url,omitempty"` // 推流地址
PullURL string `json:"pull_url,omitempty"` // 拉流地址
StreamID string `json:"stream_id,omitempty"` // 流ID
}
// Room 通话房间
type Room struct {
ID string `json:"id"`
CreatedAt time.Time `json:"created_at"`
MaxSize int `json:"max_size"`
Participants map[string]*Participant `json:"participants"`
// 房间状态
CallType string `json:"call_type"` // audio/video
IsGroupCall bool `json:"is_group_call"`
mu sync.RWMutex
closeChan chan struct{}
closed bool
}
// RoomInfo 房间信息（用于API返回）
type RoomInfo struct {
ID string `json:"id"`
CreatedAt time.Time `json:"created_at"`
ParticipantCount int `json:"participant_count"`
CallType string `json:"call_type"`
IsGroupCall bool `json:"is_group_call"`
Participants []ParticipantInfo `json:"participants"`
}
// ParticipantInfo 参与者信息（用于API返回）
type ParticipantInfo struct {
UserID string `json:"user_id"`
Type ParticipantType `json:"type"`
TypeName string `json:"type_name"`
HasAudio bool `json:"has_audio"`
HasVideo bool `json:"has_video"`
JoinedAt time.Time `json:"joined_at"`
}
// NewRoom 创建新房间
func NewRoom(id string, maxSize int) *Room {
return &Room{
ID: id,
CreatedAt: time.Now(),
MaxSize: maxSize,
Participants: make(map[string]*Participant),
closeChan: make(chan struct{}),
}
}
// SetCallType 设置通话类型
func (r *Room) SetCallType(callType string, isGroup bool) {
r.mu.Lock()
defer r.mu.Unlock()
r.CallType = callType
r.IsGroupCall = isGroup
}
// AddParticipant 添加参与者
func (r *Room) AddParticipant(userID string, pType ParticipantType) (*Participant, error) {
r.mu.Lock()
defer r.mu.Unlock()
if r.closed {
return nil, ErrRoomClosed
}
// 检查是否已存在
if p, exists := r.Participants[userID]; exists {
return p, nil
}
// 检查房间容量
if len(r.Participants) >= r.MaxSize {
return nil, ErrRoomFull
}
participant := &Participant{
UserID: userID,
Type: pType,
JoinedAt: time.Now(),
}
r.Participants[userID] = participant
log.Printf("🎥 [Room:%s] 用户 %s 加入 (类型: %d)", r.ID, userID, pType)
return participant, nil
}
// RemoveParticipant 移除参与者
func (r *Room) RemoveParticipant(userID string) {
r.mu.Lock()
defer r.mu.Unlock()
if p, exists := r.Participants[userID]; exists {
// 清理 PeerConnection
if p.PeerConnection != nil {
// TODO: 关闭 PeerConnection
}
delete(r.Participants, userID)
log.Printf("🎥 [Room:%s] 用户 %s 离开", r.ID, userID)
}
// 如果房间空了，标记为可清理
if len(r.Participants) == 0 {
log.Printf("🎥 [Room:%s] 房间已空，等待清理", r.ID)
}
}
// GetParticipant 获取参与者
func (r *Room) GetParticipant(userID string) *Participant {
r.mu.RLock()
defer r.mu.RUnlock()
return r.Participants[userID]
}
// GetAllParticipants 获取所有参与者
func (r *Room) GetAllParticipants() []*Participant {
r.mu.RLock()
defer r.mu.RUnlock()
participants := make([]*Participant, 0, len(r.Participants))
for _, p := range r.Participants {
participants = append(participants, p)
}
return participants
}
// GetOtherParticipants 获取除指定用户外的其他参与者
func (r *Room) GetOtherParticipants(excludeUserID string) []*Participant {
r.mu.RLock()
defer r.mu.RUnlock()
participants := make([]*Participant, 0, len(r.Participants)-1)
for _, p := range r.Participants {
if p.UserID != excludeUserID {
participants = append(participants, p)
}
}
return participants
}
// GetParticipantCount 获取参与者数量
func (r *Room) GetParticipantCount() int {
r.mu.RLock()
defer r.mu.RUnlock()
return len(r.Participants)
}
// IsEmpty 房间是否为空
func (r *Room) IsEmpty() bool {
r.mu.RLock()
defer r.mu.RUnlock()
return len(r.Participants) == 0
}
// IsFull 房间是否已满
func (r *Room) IsFull() bool {
r.mu.RLock()
defer r.mu.RUnlock()
return len(r.Participants) >= r.MaxSize
}
// GetInfo 获取房间信息
func (r *Room) GetInfo() *RoomInfo {
r.mu.RLock()
defer r.mu.RUnlock()
participants := make([]ParticipantInfo, 0, len(r.Participants))
for _, p := range r.Participants {
typeName := "WebRTC"
if p.Type == ParticipantTypeRTMP {
typeName = "RTMP"
}
participants = append(participants, ParticipantInfo{
UserID: p.UserID,
Type: p.Type,
TypeName: typeName,
HasAudio: p.HasAudio,
HasVideo: p.HasVideo,
JoinedAt: p.JoinedAt,
})
}
return &RoomInfo{
ID: r.ID,
CreatedAt: r.CreatedAt,
ParticipantCount: len(r.Participants),
CallType: r.CallType,
IsGroupCall: r.IsGroupCall,
Participants: participants,
}
}
// UpdateParticipantMedia 更新参与者媒体状态
func (r *Room) UpdateParticipantMedia(userID string, hasAudio, hasVideo bool) {
r.mu.Lock()
defer r.mu.Unlock()
if p, exists := r.Participants[userID]; exists {
p.HasAudio = hasAudio
p.HasVideo = hasVideo
}
}
// SetParticipantRTMPURLs 设置参与者的RTMP推拉流地址
func (r *Room) SetParticipantRTMPURLs(userID, pushURL, pullURL, streamID string) {
r.mu.Lock()
defer r.mu.Unlock()
if p, exists := r.Participants[userID]; exists {
p.PushURL = pushURL
p.PullURL = pullURL
p.StreamID = streamID
}
}
// Close 关闭房间
func (r *Room) Close() {
r.mu.Lock()
defer r.mu.Unlock()
if r.closed {
return
}
r.closed = true
close(r.closeChan)
// 清理所有参与者
for userID, p := range r.Participants {
if p.PeerConnection != nil {
// TODO: 关闭 PeerConnection
}
delete(r.Participants, userID)
}
log.Printf("🎥 [Room:%s] 已关闭", r.ID)
}
// IsClosed 房间是否已关闭
func (r *Room) IsClosed() bool {
r.mu.RLock()
defer r.mu.RUnlock()
return r.closed
}
// Broadcast 向房间内所有参与者广播消息（用于信令）
func (r *Room) Broadcast(excludeUserID string, message interface{}) {
r.mu.RLock()
defer r.mu.RUnlock()
for userID := range r.Participants {
if userID != excludeUserID {
// TODO: 通过 WebSocket 发送消息
_ = message
}
}
}

501
internal/mediaserver/rtmp.go Normal file
View File

@@ -0,0 +1,501 @@
/**
* package mediaserver
*
* RTMP 服务（用于微信小程序 live-pusher/live-player）
* 功能：
* 1. 接收小程序推流
* 2. 提供拉流服务
* 3. 支持 HTTP-FLV 协议
*
* 注意：当前版本使用简化实现，实际 RTMP 协议处理需要集成完整的 RTMP 服务器
* 生产环境建议使用 nginx-rtmp 或独立部署 livego
*/
package mediaserver
import (
"crypto/hmac"
"crypto/sha1"
"encoding/base64"
"fmt"
"io"
"log"
"net"
"net/http"
"sync"
"time"
"github.com/spf13/viper"
)
// RTMPServer RTMP 服务器
type RTMPServer struct {
config *MediaServerConfig
mu sync.RWMutex
running bool
streams map[string]*RTMPStream // streamID -> stream
// 服务器监听器
rtmpListener net.Listener
httpServer *http.Server
}
// RTMPStream RTMP 流信息
type RTMPStream struct {
ID string `json:"id"`
RoomID string `json:"room_id"`
UserID string `json:"user_id"`
PushURL string `json:"push_url"`
PullURL string `json:"pull_url"`
FLVURL string `json:"flv_url"`
CreatedAt time.Time `json:"created_at"`
IsActive bool `json:"is_active"`
// 流数据缓存
dataChan chan []byte `json:"-"`
stopChan chan struct{} `json:"-"`
}
// NewRTMPServer 创建 RTMP 服务器
func NewRTMPServer(config *MediaServerConfig) *RTMPServer {
return &RTMPServer{
config: config,
streams: make(map[string]*RTMPStream),
}
}
// Start 启动 RTMP 服务器
func (r *RTMPServer) Start() {
r.mu.Lock()
if r.running {
r.mu.Unlock()
return
}
r.running = true
r.mu.Unlock()
// 启动 RTMP 服务器（简化版）
go r.startRTMPServer()
// 启动 HTTP-FLV 服务器
go r.startHTTPFLVServer()
// 启动流清理任务
go r.cleanupTask()
log.Printf("🚀 [RTMP] RTMP 服务已启动 | RTMP端口: %d | HTTP-FLV端口: %d",
r.config.RTMPPort, r.config.HTTPFLVPort)
}
// startRTMPServer 启动 RTMP 服务（简化版）
func (r *RTMPServer) startRTMPServer() {
addr := fmt.Sprintf(":%d", r.config.RTMPPort)
var err error
r.rtmpListener, err = net.Listen("tcp", addr)
if err != nil {
log.Printf("❌ [RTMP] RTMP 监听失败: %v", err)
return
}
log.Printf("🎬 [RTMP] RTMP 服务监听: %s", addr)
for {
r.mu.RLock()
if !r.running {
r.mu.RUnlock()
return
}
r.mu.RUnlock()
conn, err := r.rtmpListener.Accept()
if err != nil {
if !r.running {
return
}
log.Printf("⚠️ [RTMP] 接受连接失败: %v", err)
continue
}
// 处理 RTMP 连接
go r.handleRTMPConnection(conn)
}
}
// handleRTMPConnection 处理 RTMP 连接
// 简化版：基本的 RTMP 握手和流处理
func (r *RTMPServer) handleRTMPConnection(conn net.Conn) {
defer conn.Close()
// RTMP 协议处理
// 步骤1: RTMP 握手
if err := r.doHandshake(conn); err != nil {
log.Printf("⚠️ [RTMP] 握手失败: %v", err)
return
}
log.Printf("🎬 [RTMP] 新连接: %s", conn.RemoteAddr().String())
// 步骤2: 读取 RTMP 消息
// 简化实现：等待数据并转发
buf := make([]byte, 4096)
for {
r.mu.RLock()
if !r.running {
r.mu.RUnlock()
return
}
r.mu.RUnlock()
conn.SetReadDeadline(time.Now().Add(30 * time.Second))
n, err := conn.Read(buf)
if err != nil {
if err != io.EOF {
log.Printf("⚠️ [RTMP] 读取失败: %v", err)
}
return
}
// TODO: 解析 RTMP 消息并路由到对应的流
_ = n
}
}
// doHandshake 执行 RTMP 握手
func (r *RTMPServer) doHandshake(conn net.Conn) error {
// RTMP 握手协议
// C0 + C1 -> S0 + S1 + S2 -> C2
c0c1 := make([]byte, 1537) // C0 (1) + C1 (1536)
conn.SetReadDeadline(time.Now().Add(10 * time.Second))
if _, err := io.ReadFull(conn, c0c1); err != nil {
return fmt.Errorf("read C0C1 failed: %w", err)
}
// 验证版本号 (C0 应该是 0x03)
if c0c1[0] != 0x03 {
return fmt.Errorf("invalid RTMP version: %d", c0c1[0])
}
// 发送 S0 + S1 + S2
s0s1s2 := make([]byte, 3073) // S0 (1) + S1 (1536) + S2 (1536)
s0s1s2[0] = 0x03 // RTMP version 3
// S1: 时间戳(4) + 零(4) + 随机数据(1528)
// S2: 客户端时间戳(4) + 服务器时间戳(4) + 随机数据(1528)
copy(s0s1s2[1:1537], c0c1[1:]) // S1 = C1
copy(s0s1s2[1537:], c0c1[1:]) // S2 = C1
conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if _, err := conn.Write(s0s1s2); err != nil {
return fmt.Errorf("write S0S1S2 failed: %w", err)
}
// 读取 C2
c2 := make([]byte, 1536)
conn.SetReadDeadline(time.Now().Add(10 * time.Second))
if _, err := io.ReadFull(conn, c2); err != nil {
return fmt.Errorf("read C2 failed: %w", err)
}
return nil
}
// startHTTPFLVServer 启动 HTTP-FLV 服务
func (r *RTMPServer) startHTTPFLVServer() {
addr := fmt.Sprintf(":%d", r.config.HTTPFLVPort)
mux := http.NewServeMux()
mux.HandleFunc("/live/", r.handleFLVRequest)
mux.HandleFunc("/api/streams", r.handleStreamsAPI)
r.httpServer = &http.Server{
Addr: addr,
Handler: mux,
}
log.Printf("🎬 [RTMP] HTTP-FLV 服务监听: %s", addr)
if err := r.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Printf("❌ [RTMP] HTTP-FLV 启动失败: %v", err)
}
}
// handleFLVRequest 处理 FLV 请求
func (r *RTMPServer) handleFLVRequest(w http.ResponseWriter, req *http.Request) {
// 解析流ID: /live/{streamID}.flv
path := req.URL.Path
if len(path) < 7 { // "/live/" = 6, 至少还需要1个字符
http.Error(w, "Invalid path", http.StatusBadRequest)
return
}
// 提取 streamID
streamPath := path[6:] // 去掉 "/live/"
if len(streamPath) > 4 && streamPath[len(streamPath)-4:] == ".flv" {
streamPath = streamPath[:len(streamPath)-4]
}
r.mu.RLock()
stream, exists := r.streams[streamPath]
r.mu.RUnlock()
if !exists || !stream.IsActive {
http.Error(w, "Stream not found", http.StatusNotFound)
return
}
// 设置响应头
w.Header().Set("Content-Type", "video/x-flv")
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Transfer-Encoding", "chunked")
// TODO: 实际的 FLV 数据推送
// 当前为占位实现
log.Printf("🎬 [RTMP] FLV 请求: %s", streamPath)
// 发送 FLV header
flvHeader := []byte{0x46, 0x4C, 0x56, 0x01, 0x05, 0x00, 0x00, 0x00, 0x09, 0x00, 0x00, 0x00, 0x00}
w.Write(flvHeader)
// 保持连接直到客户端断开
<-req.Context().Done()
}
// handleStreamsAPI 处理流列表 API
func (r *RTMPServer) handleStreamsAPI(w http.ResponseWriter, req *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Access-Control-Allow-Origin", "*")
r.mu.RLock()
defer r.mu.RUnlock()
streams := make([]map[string]interface{}, 0)
for _, s := range r.streams {
if s.IsActive {
streams = append(streams, map[string]interface{}{
"id": s.ID,
"room_id": s.RoomID,
"user_id": s.UserID,
"pull_url": s.PullURL,
"flv_url": s.FLVURL,
})
}
}
fmt.Fprintf(w, `{"streams":%d,"data":%v}`, len(streams), streams)
}
// Stop 停止 RTMP 服务器
func (r *RTMPServer) Stop() {
r.mu.Lock()
defer r.mu.Unlock()
r.running = false
// 关闭 RTMP 监听器
if r.rtmpListener != nil {
r.rtmpListener.Close()
}
// 关闭 HTTP 服务器
if r.httpServer != nil {
r.httpServer.Close()
}
// 关闭所有流
for _, stream := range r.streams {
if stream.stopChan != nil {
close(stream.stopChan)
}
}
r.streams = make(map[string]*RTMPStream)
log.Println("🛑 [RTMP] 已停止")
}
// GenerateStreamURLs 为用户生成推拉流地址
func (r *RTMPServer) GenerateStreamURLs(roomID, userID string) (*RTMPStream, error) {
r.mu.Lock()
defer r.mu.Unlock()
if !r.running {
return nil, ErrRTMPNotReady
}
// 生成唯一的流ID
streamID := fmt.Sprintf("%s_%s_%d", roomID, userID, time.Now().UnixNano())
// 获取服务器配置
publicIP := r.config.PublicIP
if publicIP == "" {
publicIP = viper.GetString("turn.public_ip")
}
// 生成鉴权 Token
token := generateStreamToken(streamID)
// 生成 URL
stream := &RTMPStream{
ID: streamID,
RoomID: roomID,
UserID: userID,
PushURL: fmt.Sprintf("rtmp://%s:%d/live/%s?token=%s", publicIP, r.config.RTMPPort, streamID, token),
PullURL: fmt.Sprintf("rtmp://%s:%d/live/%s", publicIP, r.config.RTMPPort, streamID),
FLVURL: fmt.Sprintf("http://%s:%d/live/%s.flv", publicIP, r.config.HTTPFLVPort, streamID),
CreatedAt: time.Now(),
IsActive: true,
dataChan: make(chan []byte, 100),
stopChan: make(chan struct{}),
}
r.streams[streamID] = stream
log.Printf("🎬 [RTMP] 创建流 | Room:%s User:%s Stream:%s", roomID, userID, streamID)
return stream, nil
}
// GetStream 获取流信息
func (r *RTMPServer) GetStream(streamID string) *RTMPStream {
r.mu.RLock()
defer r.mu.RUnlock()
return r.streams[streamID]
}
// GetStreamsByRoom 获取房间内所有流
func (r *RTMPServer) GetStreamsByRoom(roomID string) []*RTMPStream {
r.mu.RLock()
defer r.mu.RUnlock()
streams := make([]*RTMPStream, 0)
for _, s := range r.streams {
if s.RoomID == roomID && s.IsActive {
streams = append(streams, s)
}
}
return streams
}
// GetStreamsByUser 获取用户的所有流
func (r *RTMPServer) GetStreamsByUser(userID string) []*RTMPStream {
r.mu.RLock()
defer r.mu.RUnlock()
streams := make([]*RTMPStream, 0)
for _, s := range r.streams {
if s.UserID == userID && s.IsActive {
streams = append(streams, s)
}
}
return streams
}
// RemoveStream 移除流
func (r *RTMPServer) RemoveStream(streamID string) {
r.mu.Lock()
defer r.mu.Unlock()
if stream, exists := r.streams[streamID]; exists {
stream.IsActive = false
if stream.stopChan != nil {
close(stream.stopChan)
}
delete(r.streams, streamID)
log.Printf("🎬 [RTMP] 移除流 | Stream:%s", streamID)
}
}
// RemoveStreamsByUser 移除用户的所有流
func (r *RTMPServer) RemoveStreamsByUser(userID string) {
r.mu.Lock()
defer r.mu.Unlock()
for streamID, stream := range r.streams {
if stream.UserID == userID {
stream.IsActive = false
if stream.stopChan != nil {
close(stream.stopChan)
}
delete(r.streams, streamID)
log.Printf("🎬 [RTMP] 移除用户流 | User:%s Stream:%s", userID, streamID)
}
}
}
// RemoveStreamsByRoom 移除房间的所有流
func (r *RTMPServer) RemoveStreamsByRoom(roomID string) {
r.mu.Lock()
defer r.mu.Unlock()
for streamID, stream := range r.streams {
if stream.RoomID == roomID {
stream.IsActive = false
if stream.stopChan != nil {
close(stream.stopChan)
}
delete(r.streams, streamID)
log.Printf("🎬 [RTMP] 移除房间流 | Room:%s Stream:%s", roomID, streamID)
}
}
}
// GetStreamCount 获取流数量
func (r *RTMPServer) GetStreamCount() int {
r.mu.RLock()
defer r.mu.RUnlock()
return len(r.streams)
}
// cleanupTask 清理过期流
func (r *RTMPServer) cleanupTask() {
ticker := time.NewTicker(5 * time.Minute)
defer ticker.Stop()
for {
select {
case <-ticker.C:
r.cleanupExpiredStreams()
}
r.mu.RLock()
if !r.running {
r.mu.RUnlock()
return
}
r.mu.RUnlock()
}
}
// cleanupExpiredStreams 清理过期的流
func (r *RTMPServer) cleanupExpiredStreams() {
r.mu.Lock()
defer r.mu.Unlock()
expireTime := time.Now().Add(-1 * time.Hour) // 1小时过期
for streamID, stream := range r.streams {
if stream.CreatedAt.Before(expireTime) && !stream.IsActive {
if stream.stopChan != nil {
close(stream.stopChan)
}
delete(r.streams, streamID)
log.Printf("🗑️ [RTMP] 清理过期流 | Stream:%s", streamID)
}
}
}
// generateStreamToken 生成流鉴权 Token
func generateStreamToken(streamID string) string {
secret := viper.GetString("turn.shared_secret")
timestamp := time.Now().Add(24 * time.Hour).Unix()
data := fmt.Sprintf("%s:%d", streamID, timestamp)
mac := hmac.New(sha1.New, []byte(secret))
mac.Write([]byte(data))
return base64.URLEncoding.EncodeToString(mac.Sum(nil))
}
// ValidateStreamToken 验证流鉴权 Token
func ValidateStreamToken(streamID, token string) bool {
expectedToken := generateStreamToken(streamID)
return token == expectedToken
}

229
internal/mediaserver/server.go Normal file
View File

@@ -0,0 +1,229 @@
/**
* package mediaserver
*
* 媒体服务器主入口
* 功能：
* 1. 集成 WebRTC SFU（支持 H5/App 端音视频通话）
* 2. 集成 RTMP 服务（支持微信小程序 live-pusher/live-player）
* 3. 实现 WebRTC ↔ RTMP 媒体桥接（跨平台互通）
*/
package mediaserver
import (
"log"
"sync"
"github.com/spf13/viper"
)
// MediaServer 媒体服务器主结构
type MediaServer struct {
sfu *SFUServer // WebRTC SFU 服务
rtmp *RTMPServer // RTMP 服务
rooms map[string]*Room // 通话房间管理
mu sync.RWMutex // 房间锁
config *MediaServerConfig // 配置
}
// MediaServerConfig 媒体服务器配置
type MediaServerConfig struct {
Enabled bool `mapstructure:"enabled"`
PublicIP string `mapstructure:"public_ip"`
WebRTCPort int `mapstructure:"webrtc_port"`
RTMPPort int `mapstructure:"rtmp_port"`
HTTPFLVPort int `mapstructure:"http_flv_port"`
// SFU 配置
MaxRoomSize int `mapstructure:"max_room_size"` // 单房间最大人数
// 编解码配置
VideoCodec string `mapstructure:"video_codec"` // 视频编解码：h264
AudioCodec string `mapstructure:"audio_codec"` // 音频编解码：opus/aac
}
// 全局媒体服务器实例
var (
server *MediaServer
once sync.Once
)
// GetServer 获取媒体服务器单例
func GetServer() *MediaServer {
once.Do(func() {
server = newMediaServer()
})
return server
}
// newMediaServer 创建媒体服务器实例
func newMediaServer() *MediaServer {
config := loadConfig()
ms := &MediaServer{
rooms: make(map[string]*Room),
config: config,
}
if config.Enabled {
ms.sfu = NewSFUServer(config)
ms.rtmp = NewRTMPServer(config)
}
return ms
}
// loadConfig 从 viper 加载配置
func loadConfig() *MediaServerConfig {
config := &MediaServerConfig{
Enabled: viper.GetBool("media.enabled"),
PublicIP: viper.GetString("media.public_ip"),
WebRTCPort: viper.GetInt("media.webrtc_port"),
RTMPPort: viper.GetInt("media.rtmp_port"),
HTTPFLVPort: viper.GetInt("media.http_flv_port"),
MaxRoomSize: viper.GetInt("media.max_room_size"),
VideoCodec: viper.GetString("media.video_codec"),
AudioCodec: viper.GetString("media.audio_codec"),
}
// 设置默认值
if config.PublicIP == "" {
config.PublicIP = viper.GetString("turn.public_ip") // 复用 TURN 的公网IP
}
if config.WebRTCPort == 0 {
config.WebRTCPort = 8080
}
if config.RTMPPort == 0 {
config.RTMPPort = 1935
}
if config.HTTPFLVPort == 0 {
config.HTTPFLVPort = 7001
}
if config.MaxRoomSize == 0 {
config.MaxRoomSize = 9 // 默认最多9人
}
if config.VideoCodec == "" {
config.VideoCodec = "h264"
}
if config.AudioCodec == "" {
config.AudioCodec = "opus"
}
return config
}
// Start 启动媒体服务器
func Start() {
ms := GetServer()
if !ms.config.Enabled {
log.Println("ℹ️ [MediaServer] 服务已禁用")
return
}
log.Println("🎥 [MediaServer] 正在启动...")
// 启动 WebRTC SFU
if ms.sfu != nil {
go ms.sfu.Start()
}
// 启动 RTMP 服务
if ms.rtmp != nil {
go ms.rtmp.Start()
}
log.Printf("🎥 [MediaServer] 已启动 | WebRTC: %d | RTMP: %d | HTTP-FLV: %d",
ms.config.WebRTCPort, ms.config.RTMPPort, ms.config.HTTPFLVPort)
}
// Stop 停止媒体服务器
func (ms *MediaServer) Stop() {
log.Println("🛑 [MediaServer] 正在停止...")
// 关闭所有房间
ms.mu.Lock()
for _, room := range ms.rooms {
room.Close()
}
ms.rooms = make(map[string]*Room)
ms.mu.Unlock()
// 停止 SFU
if ms.sfu != nil {
ms.sfu.Stop()
}
// 停止 RTMP
if ms.rtmp != nil {
ms.rtmp.Stop()
}
log.Println("🛑 [MediaServer] 已停止")
}
// GetOrCreateRoom 获取或创建房间
func (ms *MediaServer) GetOrCreateRoom(roomID string) *Room {
ms.mu.Lock()
defer ms.mu.Unlock()
if room, exists := ms.rooms[roomID]; exists {
return room
}
room := NewRoom(roomID, ms.config.MaxRoomSize)
ms.rooms[roomID] = room
log.Printf("🎥 [MediaServer] 创建房间: %s", roomID)
return room
}
// GetRoom 获取房间
func (ms *MediaServer) GetRoom(roomID string) *Room {
ms.mu.RLock()
defer ms.mu.RUnlock()
return ms.rooms[roomID]
}
// RemoveRoom 移除房间
func (ms *MediaServer) RemoveRoom(roomID string) {
ms.mu.Lock()
defer ms.mu.Unlock()
if room, exists := ms.rooms[roomID]; exists {
room.Close()
delete(ms.rooms, roomID)
log.Printf("🎥 [MediaServer] 移除房间: %s", roomID)
}
}
// GetConfig 获取配置
func (ms *MediaServer) GetConfig() *MediaServerConfig {
return ms.config
}
// GetSFU 获取 SFU 服务
func (ms *MediaServer) GetSFU() *SFUServer {
return ms.sfu
}
// GetRTMP 获取 RTMP 服务
func (ms *MediaServer) GetRTMP() *RTMPServer {
return ms.rtmp
}
// GetRoomCount 获取房间数量
func (ms *MediaServer) GetRoomCount() int {
ms.mu.RLock()
defer ms.mu.RUnlock()
return len(ms.rooms)
}
// GetAllRooms 获取所有房间信息（用于调试/监控）
func (ms *MediaServer) GetAllRooms() []*RoomInfo {
ms.mu.RLock()
defer ms.mu.RUnlock()
infos := make([]*RoomInfo, 0, len(ms.rooms))
for _, room := range ms.rooms {
infos = append(infos, room.GetInfo())
}
return infos
}

395
internal/mediaserver/sfu.go Normal file
View File

@@ -0,0 +1,395 @@
/**
* package mediaserver
*
* WebRTC SFU (Selective Forwarding Unit) 服务
* 功能：
* 1. 接收 WebRTC 流
* 2. 转发给同房间的其他用户
* 3. 支持音视频分离转发
*/
package mediaserver
import (
"crypto/hmac"
"crypto/sha1"
"encoding/base64"
"encoding/json"
"fmt"
"log"
"strings"
"sync"
"time"
"github.com/pion/interceptor"
"github.com/pion/interceptor/pkg/intervalpli"
"github.com/pion/webrtc/v3"
"github.com/spf13/viper"
)
// SFUServer WebRTC SFU 服务器
type SFUServer struct {
config *MediaServerConfig
api *webrtc.API
mu sync.RWMutex
running bool
// Track 管理
trackLocals map[string]*webrtc.TrackLocalStaticRTP // trackID -> localTrack
}
// NewSFUServer 创建 SFU 服务器
func NewSFUServer(config *MediaServerConfig) *SFUServer {
return &SFUServer{
config: config,
trackLocals: make(map[string]*webrtc.TrackLocalStaticRTP),
}
}
// Start 启动 SFU 服务器
func (s *SFUServer) Start() {
s.mu.Lock()
if s.running {
s.mu.Unlock()
return
}
s.running = true
s.mu.Unlock()
// 创建 MediaEngine
m := &webrtc.MediaEngine{}
// 注册默认编解码器
if err := m.RegisterDefaultCodecs(); err != nil {
log.Printf("❌ [SFU] 注册编解码器失败: %v", err)
return
}
// 创建拦截器注册表
i := &interceptor.Registry{}
// 注册 PLI 拦截器（用于请求关键帧）
intervalPliFactory, err := intervalpli.NewReceiverInterceptor()
if err != nil {
log.Printf("❌ [SFU] 创建 PLI 拦截器失败: %v", err)
return
}
i.Add(intervalPliFactory)
// 使用拦截器
if err := webrtc.RegisterDefaultInterceptors(m, i); err != nil {
log.Printf("❌ [SFU] 注册拦截器失败: %v", err)
return
}
// 创建 API
s.api = webrtc.NewAPI(webrtc.WithMediaEngine(m), webrtc.WithInterceptorRegistry(i))
log.Printf("🚀 [SFU] WebRTC SFU 已启动 | 端口: %d", s.config.WebRTCPort)
}
// Stop 停止 SFU 服务器
func (s *SFUServer) Stop() {
s.mu.Lock()
defer s.mu.Unlock()
s.running = false
s.trackLocals = make(map[string]*webrtc.TrackLocalStaticRTP)
log.Println("🛑 [SFU] 已停止")
}
// CreatePeerConnection 为用户创建 PeerConnection
func (s *SFUServer) CreatePeerConnection(roomID, userID string) (*webrtc.PeerConnection, error) {
s.mu.RLock()
if !s.running || s.api == nil {
s.mu.RUnlock()
return nil, ErrSFUNotReady
}
s.mu.RUnlock()
// 获取 ICE 服务器配置
iceServers := s.getICEServers(userID)
// 创建 PeerConnection 配置
config := webrtc.Configuration{
ICEServers: iceServers,
}
// 创建 PeerConnection
pc, err := s.api.NewPeerConnection(config)
if err != nil {
return nil, fmt.Errorf("create peer connection failed: %w", err)
}
// 监听 ICE 连接状态
pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
log.Printf("🔗 [SFU] Room:%s User:%s ICE状态: %s", roomID, userID, state.String())
if state == webrtc.ICEConnectionStateFailed || state == webrtc.ICEConnectionStateDisconnected {
// 连接断开，清理资源
s.handleDisconnect(roomID, userID)
}
})
// 监听轨道
pc.OnTrack(func(remoteTrack *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
s.handleTrack(roomID, userID, remoteTrack, receiver)
})
return pc, nil
}
// getICEServers 获取 ICE 服务器配置
func (s *SFUServer) getICEServers(userID string) []webrtc.ICEServer {
servers := []webrtc.ICEServer{}
// 添加 STUN 服务器
turnPublicIP := viper.GetString("turn.public_ip")
turnPort := viper.GetInt("turn.listen_port")
if turnPublicIP != "" && turnPort > 0 {
stunURL := fmt.Sprintf("stun:%s:%d", turnPublicIP, turnPort)
servers = append(servers, webrtc.ICEServer{
URLs: []string{stunURL},
})
// 添加 TURN 服务器（如果启用）
if viper.GetBool("turn.enabled") {
turnURL := fmt.Sprintf("turn:%s:%d", turnPublicIP, turnPort)
// 生成 TURN 凭证
username, credential := generateTURNCredentials(userID)
servers = append(servers, webrtc.ICEServer{
URLs: []string{turnURL},
Username: username,
Credential: credential,
})
}
}
return servers
}
// handleTrack 处理远程轨道
func (s *SFUServer) handleTrack(roomID, userID string, remoteTrack *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
trackID := fmt.Sprintf("%s_%s_%s", roomID, userID, remoteTrack.Kind().String())
log.Printf("🎬 [SFU] 收到轨道 | Room:%s User:%s Kind:%s ID:%s",
roomID, userID, remoteTrack.Kind().String(), trackID)
// 创建本地轨道用于转发
localTrack, err := webrtc.NewTrackLocalStaticRTP(
remoteTrack.Codec().RTPCodecCapability,
trackID,
fmt.Sprintf("stream_%s", userID),
)
if err != nil {
log.Printf("❌ [SFU] 创建本地轨道失败: %v", err)
return
}
// 保存轨道
s.mu.Lock()
s.trackLocals[trackID] = localTrack
s.mu.Unlock()
// 转发 RTP 包
go func() {
buf := make([]byte, 1500)
for {
n, _, readErr := remoteTrack.Read(buf)
if readErr != nil {
log.Printf("⚠️ [SFU] 读取轨道失败 | Track:%s Error:%v", trackID, readErr)
break
}
// 写入本地轨道（会自动转发给所有订阅者）
if _, writeErr := localTrack.Write(buf[:n]); writeErr != nil {
log.Printf("⚠️ [SFU] 写入轨道失败 | Track:%s Error:%v", trackID, writeErr)
break
}
}
// 清理轨道
s.mu.Lock()
delete(s.trackLocals, trackID)
s.mu.Unlock()
}()
// 通知房间内其他用户有新轨道
s.notifyNewTrack(roomID, userID, localTrack)
}
// notifyNewTrack 通知房间内其他用户有新轨道
func (s *SFUServer) notifyNewTrack(roomID, senderUserID string, track *webrtc.TrackLocalStaticRTP) {
ms := GetServer()
room := ms.GetRoom(roomID)
if room == nil {
return
}
// 获取其他参与者
for _, p := range room.GetOtherParticipants(senderUserID) {
if p.Type != ParticipantTypeWebRTC {
continue // 只处理 WebRTC 用户
}
if p.PeerConnection == nil {
continue
}
pc, ok := p.PeerConnection.(*webrtc.PeerConnection)
if !ok {
continue
}
// 添加轨道到对方的 PeerConnection
if _, err := pc.AddTrack(track); err != nil {
log.Printf("⚠️ [SFU] 添加轨道到用户 %s 失败: %v", p.UserID, err)
}
}
}
// handleDisconnect 处理断开连接
func (s *SFUServer) handleDisconnect(roomID, userID string) {
ms := GetServer()
room := ms.GetRoom(roomID)
if room == nil {
return
}
room.RemoveParticipant(userID)
// 清理该用户的所有轨道
s.mu.Lock()
for trackID := range s.trackLocals {
// 检查 trackID 是否属于该用户
prefix := fmt.Sprintf("%s_%s_", roomID, userID)
if strings.HasPrefix(trackID, prefix) {
delete(s.trackLocals, trackID)
}
}
s.mu.Unlock()
// 如果房间空了，移除房间
if room.IsEmpty() {
ms.RemoveRoom(roomID)
}
}
// GetTrackCount 获取轨道数量
func (s *SFUServer) GetTrackCount() int {
s.mu.RLock()
defer s.mu.RUnlock()
return len(s.trackLocals)
}
// HandleOffer 处理 SDP Offer
func (s *SFUServer) HandleOffer(roomID, userID string, offerSDP string) (string, error) {
ms := GetServer()
room := ms.GetOrCreateRoom(roomID)
// 添加参与者
participant, err := room.AddParticipant(userID, ParticipantTypeWebRTC)
if err != nil {
return "", err
}
// 创建 PeerConnection
pc, err := s.CreatePeerConnection(roomID, userID)
if err != nil {
return "", err
}
participant.PeerConnection = pc
// 设置远程描述
offer := webrtc.SessionDescription{
Type: webrtc.SDPTypeOffer,
SDP: offerSDP,
}
if err := pc.SetRemoteDescription(offer); err != nil {
return "", fmt.Errorf("set remote description failed: %w", err)
}
// 添加房间内其他用户的轨道
s.addExistingTracks(pc, roomID, userID)
// 创建 Answer
answer, err := pc.CreateAnswer(nil)
if err != nil {
return "", fmt.Errorf("create answer failed: %w", err)
}
// 设置本地描述
if err := pc.SetLocalDescription(answer); err != nil {
return "", fmt.Errorf("set local description failed: %w", err)
}
// 等待 ICE 收集完成
<-webrtc.GatheringCompletePromise(pc)
return pc.LocalDescription().SDP, nil
}
// addExistingTracks 添加房间内已有的轨道
func (s *SFUServer) addExistingTracks(pc *webrtc.PeerConnection, roomID, excludeUserID string) {
s.mu.RLock()
defer s.mu.RUnlock()
prefix := fmt.Sprintf("%s_", roomID)
excludePrefix := fmt.Sprintf("%s_%s_", roomID, excludeUserID)
for trackID, track := range s.trackLocals {
// 检查是否是同房间的轨道，且不是自己的
if strings.HasPrefix(trackID, prefix) && !strings.HasPrefix(trackID, excludePrefix) {
if _, err := pc.AddTrack(track); err != nil {
log.Printf("⚠️ [SFU] 添加已有轨道失败: %v", err)
}
}
}
}
// HandleICECandidate 处理 ICE 候选
func (s *SFUServer) HandleICECandidate(roomID, userID string, candidateJSON string) error {
ms := GetServer()
room := ms.GetRoom(roomID)
if room == nil {
return ErrRoomNotFound
}
participant := room.GetParticipant(userID)
if participant == nil {
return ErrParticipantNotFound
}
pc, ok := participant.PeerConnection.(*webrtc.PeerConnection)
if !ok || pc == nil {
return ErrSFUNotReady
}
var candidate webrtc.ICECandidateInit
if err := json.Unmarshal([]byte(candidateJSON), &candidate); err != nil {
return fmt.Errorf("parse ICE candidate failed: %w", err)
}
if err := pc.AddICECandidate(candidate); err != nil {
return fmt.Errorf("add ICE candidate failed: %w", err)
}
return nil
}
// generateTURNCredentials 生成 TURN 凭证
func generateTURNCredentials(userID string) (string, string) {
timestamp := time.Now().Add(24 * time.Hour).Unix()
username := fmt.Sprintf("%d:%s", timestamp, userID)
secret := viper.GetString("turn.shared_secret")
mac := hmac.New(sha1.New, []byte(secret))
mac.Write([]byte(username))
password := base64.StdEncoding.EncodeToString(mac.Sum(nil))
return username, password
}
// GetTURNCredentials 获取 TURN 凭证（供外部调用）
func GetTURNCredentials(userID string) (string, string) {
return generateTURNCredentials(userID)
}