From 2485cd3ac9ea7456dda592e925fc4dd363b8cb7d Mon Sep 17 00:00:00 2001 From: liqi Date: Mon, 15 Dec 2025 09:04:14 +0800 Subject: [PATCH] =?UTF-8?q?=E5=BE=AE=E4=BF=A1=E5=B0=8F=E7=A8=8B=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/server/main.go | 10 + configs/config.yaml | 23 +- go.mod | 15 + go.sum | 32 +++ internal/api/call_handler.go | 419 +++++++++++++++++++++++++++ internal/api/media_handler.go | 183 ++++++++++++ internal/mediaserver/bridge.go | 482 +++++++++++++++++++++++++++++++ internal/mediaserver/errors.go | 35 +++ internal/mediaserver/room.go | 299 ++++++++++++++++++++ internal/mediaserver/rtmp.go | 501 +++++++++++++++++++++++++++++++++ internal/mediaserver/server.go | 229 +++++++++++++++ internal/mediaserver/sfu.go | 395 ++++++++++++++++++++++++++ 12 files changed, 2622 insertions(+), 1 deletion(-) create mode 100644 internal/api/call_handler.go create mode 100644 internal/api/media_handler.go create mode 100644 internal/mediaserver/bridge.go create mode 100644 internal/mediaserver/errors.go create mode 100644 internal/mediaserver/room.go create mode 100644 internal/mediaserver/rtmp.go create mode 100644 internal/mediaserver/server.go create mode 100644 internal/mediaserver/sfu.go diff --git a/cmd/server/main.go b/cmd/server/main.go index be3c93e..6905a12 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -27,6 +27,7 @@ import ( "xk-websocket-v2/internal/api" "xk-websocket-v2/internal/manager" + "xk-websocket-v2/internal/mediaserver" "xk-websocket-v2/internal/middleware" "xk-websocket-v2/internal/model" "xk-websocket-v2/internal/service" @@ -229,6 +230,9 @@ func main() { // 步骤5: 启动TURN服务器（用于WebRTC音视频通话） go turnserver.Start() + // 步骤5.5: 启动媒体服务器（WebRTC SFU + RTMP） + go mediaserver.Start() + // 步骤6: 创建Gin路由引擎 r := gin.Default() @@ -391,6 +395,12 @@ func main() { // 朋友圈管理 api.RegisterMomentRoutes(authGroup) + + // 通话管理（音视频） + api.RegisterCallRoutes(authGroup) + + // 媒体服务器管理 + api.RegisterMediaRoutes(authGroup) } } diff --git a/configs/config.yaml b/configs/config.yaml index 4407dae..d6852f1 100644 --- a/configs/config.yaml +++ b/configs/config.yaml @@ -56,4 +56,25 @@ turn: # UDP 端口范围下限 min_port: 49152 # UDP 端口范围上限 - max_port: 65535 \ No newline at end of file + max_port: 65535 + +# ========================================== +# 媒体服务器配置 (WebRTC SFU + RTMP) +# ========================================== +media: + # 是否启用媒体服务器 + enabled: true + # 服务器公网 IP (默认复用 turn.public_ip) + public_ip: "" + # WebRTC SFU 端口 + webrtc_port: 8080 + # RTMP 服务端口 + rtmp_port: 1935 + # HTTP-FLV 端口 + http_flv_port: 7001 + # 单房间最大人数 + max_room_size: 9 + # 视频编解码器 (h264) + video_codec: "h264" + # 音频编解码器 (opus) + audio_codec: "opus" \ No newline at end of file diff --git a/go.mod b/go.mod index 44fcc14..27a1fc6 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,10 @@ require ( github.com/golang-jwt/jwt/v5 v5.3.0 github.com/gorilla/websocket v1.5.3 github.com/panjf2000/ants/v2 v2.11.3 + github.com/pion/interceptor v0.1.29 + github.com/pion/rtp v1.8.7 github.com/pion/turn/v2 v2.1.6 + github.com/pion/webrtc/v3 v3.3.6 github.com/spf13/viper v1.21.0 golang.org/x/crypto v0.45.0 gorm.io/driver/mysql v1.6.0 @@ -22,6 +25,7 @@ require ( github.com/bytedance/sonic/loader v0.4.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/gabriel-vasile/mimetype v1.4.11 // indirect @@ -33,6 +37,7 @@ require ( github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/goccy/go-json v0.10.5 // indirect github.com/goccy/go-yaml v1.19.0 // indirect + github.com/google/uuid v1.3.1 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect github.com/json-iterator/go v1.1.12 // indirect @@ -42,18 +47,27 @@ require ( github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect + github.com/pion/datachannel v1.5.8 // indirect github.com/pion/dtls/v2 v2.2.12 // indirect + github.com/pion/ice/v2 v2.3.38 // indirect github.com/pion/logging v0.2.4 // indirect + github.com/pion/mdns v0.0.12 // indirect github.com/pion/randutil v0.1.0 // indirect + github.com/pion/rtcp v1.2.14 // indirect + github.com/pion/sctp v1.8.19 // indirect + github.com/pion/sdp/v3 v3.0.9 // indirect + github.com/pion/srtp/v2 v2.0.20 // indirect github.com/pion/stun v0.6.1 // indirect github.com/pion/transport/v2 v2.2.10 // indirect github.com/pion/transport/v3 v3.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect github.com/quic-go/qpack v0.6.0 // indirect github.com/quic-go/quic-go v0.57.1 // indirect github.com/sagikazarmark/locafero v0.12.0 // indirect github.com/spf13/afero v1.15.0 // indirect github.com/spf13/cast v1.10.0 // indirect github.com/spf13/pflag v1.0.10 // indirect + github.com/stretchr/testify v1.11.1 // indirect github.com/subosito/gotenv v1.6.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.3.1 // indirect @@ -66,4 +80,5 @@ require ( golang.org/x/sys v0.38.0 // indirect golang.org/x/text v0.31.0 // indirect google.golang.org/protobuf v1.36.10 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 46ea8fd..4137700 100644 --- a/go.sum +++ b/go.sum @@ -48,6 +48,8 @@ github.com/golang-jwt/jwt/v5 v5.3.0/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArs github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4= +github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= @@ -58,8 +60,11 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= @@ -81,24 +86,49 @@ github.com/panjf2000/ants/v2 v2.11.3 h1:AfI0ngBoXJmYOpDh9m516vjqoUu2sLrIVgppI9TZ github.com/panjf2000/ants/v2 v2.11.3/go.mod h1:8u92CYMUc6gyvTIw8Ru7Mt7+/ESnJahz5EVtqfrilek= github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= +github.com/pion/datachannel v1.5.8 h1:ph1P1NsGkazkjrvyMfhRBUAWMxugJjq2HfQifaOoSNo= +github.com/pion/datachannel v1.5.8/go.mod h1:PgmdpoaNBLX9HNzNClmdki4DYW5JtI7Yibu8QzbL3tI= github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s= github.com/pion/dtls/v2 v2.2.12 h1:KP7H5/c1EiVAAKUmXyCzPiQe5+bCJrpOeKg/L05dunk= github.com/pion/dtls/v2 v2.2.12/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE= +github.com/pion/ice/v2 v2.3.38 h1:DEpt13igPfvkE2+1Q+6e8mP30dtWnQD3CtMIKoRDRmA= +github.com/pion/ice/v2 v2.3.38/go.mod h1:mBF7lnigdqgtB+YHkaY/Y6s6tsyRyo4u4rPGRuOjUBQ= +github.com/pion/interceptor v0.1.29 h1:39fsnlP1U8gw2JzOFWdfCU82vHvhW9o0rZnZF56wF+M= +github.com/pion/interceptor v0.1.29/go.mod h1:ri+LGNjRUc5xUNtDEPzfdkmSqISixVTBF/z/Zms/6T4= github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms= github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8= github.com/pion/logging v0.2.4/go.mod h1:DffhXTKYdNZU+KtJ5pyQDjvOAh/GsNSyv1lbkFbe3so= +github.com/pion/mdns v0.0.12 h1:CiMYlY+O0azojWDmxdNr7ADGrnZ+V6Ilfner+6mSVK8= +github.com/pion/mdns v0.0.12/go.mod h1:VExJjv8to/6Wqm1FXK+Ii/Z9tsVk/F5sD/N70cnYFbk= github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= +github.com/pion/rtcp v1.2.12/go.mod h1:sn6qjxvnwyAkkPzPULIbVqSKI5Dv54Rv7VG0kNxh9L4= +github.com/pion/rtcp v1.2.14 h1:KCkGV3vJ+4DAJmvP0vaQShsb0xkRfWkO540Gy102KyE= +github.com/pion/rtcp v1.2.14/go.mod h1:sn6qjxvnwyAkkPzPULIbVqSKI5Dv54Rv7VG0kNxh9L4= +github.com/pion/rtp v1.8.3/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU= +github.com/pion/rtp v1.8.7 h1:qslKkG8qxvQ7hqaxkmL7Pl0XcUm+/Er7nMnu6Vq+ZxM= +github.com/pion/rtp v1.8.7/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU= +github.com/pion/sctp v1.8.19 h1:2CYuw+SQ5vkQ9t0HdOPccsCz1GQMDuVy5PglLgKVBW8= +github.com/pion/sctp v1.8.19/go.mod h1:P6PbDVA++OJMrVNg2AL3XtYHV4uD6dvfyOovCgMs0PE= +github.com/pion/sdp/v3 v3.0.9 h1:pX++dCHoHUwq43kuwf3PyJfHlwIj4hXA7Vrifiq0IJY= +github.com/pion/sdp/v3 v3.0.9/go.mod h1:B5xmvENq5IXJimIO4zfp6LAe1fD9N+kFv+V/1lOdz8M= +github.com/pion/srtp/v2 v2.0.20 h1:HNNny4s+OUmG280ETrCdgFndp4ufx3/uy85EawYEhTk= +github.com/pion/srtp/v2 v2.0.20/go.mod h1:0KJQjA99A6/a0DOVTu1PhDSw0CXF2jTkqOoMg3ODqdA= github.com/pion/stun v0.6.1 h1:8lp6YejULeHBF8NmV8e2787BogQhduZugh5PdhDyyN4= github.com/pion/stun v0.6.1/go.mod h1:/hO7APkX4hZKu/D0f2lHzNyvdkTGtIy3NDmLR7kSz/8= github.com/pion/transport/v2 v2.2.1/go.mod h1:cXXWavvCnFF6McHTft3DWS9iic2Mftcz1Aq29pGcU5g= +github.com/pion/transport/v2 v2.2.3/go.mod h1:q2U/tf9FEfnSBGSW6w5Qp5PFWRLRj3NjLhCCgpRK4p0= github.com/pion/transport/v2 v2.2.4/go.mod h1:q2U/tf9FEfnSBGSW6w5Qp5PFWRLRj3NjLhCCgpRK4p0= github.com/pion/transport/v2 v2.2.10 h1:ucLBLE8nuxiHfvkFKnkDQRYWYfp8ejf4YBOPfaQpw6Q= github.com/pion/transport/v2 v2.2.10/go.mod h1:sq1kSLWs+cHW9E+2fJP95QudkzbK7wscs8yYgQToO5E= +github.com/pion/transport/v3 v3.0.1/go.mod h1:UY7kiITrlMv7/IKgd5eTUcaahZx5oUN3l9SzK5f5xE0= github.com/pion/transport/v3 v3.1.1 h1:Tr684+fnnKlhPceU+ICdrw6KKkTms+5qHMgw6bIkYOM= github.com/pion/transport/v3 v3.1.1/go.mod h1:+c2eewC5WJQHiAA46fkMMzoYZSuGzA/7E2FPrOYHctQ= +github.com/pion/turn/v2 v2.1.3/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY= github.com/pion/turn/v2 v2.1.6 h1:Xr2niVsiPTB0FPtt+yAWKFUkU1eotQbGgpTIld4x1Gc= github.com/pion/turn/v2 v2.1.6/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY= +github.com/pion/webrtc/v3 v3.3.6 h1:7XAh4RPtlY1Vul6/GmZrv7z+NnxKA6If0KStXBI2ZLE= +github.com/pion/webrtc/v3 v3.3.6/go.mod h1:zyN7th4mZpV27eXybfR/cnUf3J2DRy8zw/mdjD9JTNM= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= @@ -126,6 +156,7 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= @@ -209,6 +240,7 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ= diff --git a/internal/api/call_handler.go b/internal/api/call_handler.go new file mode 100644 index 0000000..73ef303 --- /dev/null +++ b/internal/api/call_handler.go @@ -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) + } +} diff --git a/internal/api/media_handler.go b/internal/api/media_handler.go new file mode 100644 index 0000000..6576587 --- /dev/null +++ b/internal/api/media_handler.go @@ -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) + } +} + diff --git a/internal/mediaserver/bridge.go b/internal/mediaserver/bridge.go new file mode 100644 index 0000000..c8a3099 --- /dev/null +++ b/internal/mediaserver/bridge.go @@ -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 +} diff --git a/internal/mediaserver/errors.go b/internal/mediaserver/errors.go new file mode 100644 index 0000000..7ca1608 --- /dev/null +++ b/internal/mediaserver/errors.go @@ -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") +) + diff --git a/internal/mediaserver/room.go b/internal/mediaserver/room.go new file mode 100644 index 0000000..2ae7831 --- /dev/null +++ b/internal/mediaserver/room.go @@ -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 + } + } +} + diff --git a/internal/mediaserver/rtmp.go b/internal/mediaserver/rtmp.go new file mode 100644 index 0000000..5fe5695 --- /dev/null +++ b/internal/mediaserver/rtmp.go @@ -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 +} diff --git a/internal/mediaserver/server.go b/internal/mediaserver/server.go new file mode 100644 index 0000000..49e520f --- /dev/null +++ b/internal/mediaserver/server.go @@ -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 +} + diff --git a/internal/mediaserver/sfu.go b/internal/mediaserver/sfu.go new file mode 100644 index 0000000..30ea7ab --- /dev/null +++ b/internal/mediaserver/sfu.go @@ -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) +}