diff --git a/go.mod b/go.mod index 27a1fc6..c447ce4 100644 --- a/go.mod +++ b/go.mod @@ -13,6 +13,8 @@ require ( github.com/pion/turn/v2 v2.1.6 github.com/pion/webrtc/v3 v3.3.6 github.com/spf13/viper v1.21.0 + github.com/yutopp/go-flv v0.3.1 + github.com/yutopp/go-rtmp v0.0.7 golang.org/x/crypto v0.45.0 gorm.io/driver/mysql v1.6.0 gorm.io/gorm v1.31.1 @@ -38,12 +40,15 @@ require ( 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/hashicorp/errwrap v1.1.0 // indirect + github.com/hashicorp/go-multierror v1.1.0 // 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 github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/leodido/go-urn v1.4.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect + github.com/mitchellh/mapstructure v1.4.1 // indirect 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 @@ -60,10 +65,12 @@ require ( 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/pkg/errors v0.9.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/sirupsen/logrus v1.7.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 @@ -72,6 +79,7 @@ require ( github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.3.1 // indirect github.com/wlynxg/anet v0.0.5 // indirect + github.com/yutopp/go-amf0 v0.1.0 // indirect go.uber.org/mock v0.6.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/arch v0.23.0 // indirect diff --git a/go.sum b/go.sum index 4137700..f2d6719 100644 --- a/go.sum +++ b/go.sum @@ -15,6 +15,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/fortytw2/leaktest v1.2.0 h1:cj6GCiwJDH7l3tMHLjZDo0QqPtrXJiWSI9JgpeQKw+Q= +github.com/fortytw2/leaktest v1.2.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g= github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= @@ -52,6 +54,11 @@ 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/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= +github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= +github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= +github.com/hashicorp/go-multierror v1.1.0 h1:B9UzwGQJehnUY1yNrnwREHc3fGbC2xefo8g4TbElacI= +github.com/hashicorp/go-multierror v1.1.0/go.mod h1:spPvp8C1qA32ftKqdAHm4hHTbPw+vmowP0z+KUhOZdA= github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= @@ -71,6 +78,8 @@ github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mitchellh/mapstructure v1.4.1 h1:CpVNEelQCZBooIPDn+AR3NpivK/TIKU8bDxdASFVQag= +github.com/mitchellh/mapstructure v1.4.1/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -129,6 +138,8 @@ 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/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= 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= @@ -139,6 +150,8 @@ github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjR github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= github.com/sagikazarmark/locafero v0.12.0 h1:/NQhBAkUb4+fH1jivKHWusDYFjMOOKU88eegjfxfHb4= github.com/sagikazarmark/locafero v0.12.0/go.mod h1:sZh36u/YSZ918v0Io+U9ogLYQJ9tLLBmM4eneO6WwsI= +github.com/sirupsen/logrus v1.7.0 h1:ShrD1U9pZB12TX0cVy0DtePoCH97K8EtX+mg7ZARUtM= +github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I= github.com/spf13/afero v1.15.0/go.mod h1:NC2ByUVxtQs4b3sIUphxK0NioZnmxgyCrfzeuq8lxMg= github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY= @@ -151,9 +164,11 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+ github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= 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= @@ -170,6 +185,12 @@ github.com/wlynxg/anet v0.0.3/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguH github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +github.com/yutopp/go-amf0 v0.1.0 h1:a3UeBZG7nRF0zfvmPn2iAfNo1RGzUpHz1VyJD2oGrik= +github.com/yutopp/go-amf0 v0.1.0/go.mod h1:QzDOBr9RV6sQh6E5GFEJROZbU0iQKijORBmprkb3FIk= +github.com/yutopp/go-flv v0.3.1 h1:4ILK6OgCJgUNm2WOjaucWM5lUHE0+sLNPdjq3L0Xtjk= +github.com/yutopp/go-flv v0.3.1/go.mod h1:pAlHPSVRMv5aCUKmGOS/dZn/ooTgnc09qOPmiUNMubs= +github.com/yutopp/go-rtmp v0.0.7 h1:sKKm1MVV3ANbJHZlf3Kq8ecq99y5U7XnDUDxSjuK7KU= +github.com/yutopp/go-rtmp v0.0.7/go.mod h1:KSwrC9Xj5Kf18EUlk1g7CScecjXfIqc0J5q+S0u6Irc= go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= @@ -201,10 +222,12 @@ golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.3.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= diff --git a/internal/mediaserver/rtmp.go b/internal/mediaserver/rtmp.go index 5fe5695..85b541f 100644 --- a/internal/mediaserver/rtmp.go +++ b/internal/mediaserver/rtmp.go @@ -2,17 +2,17 @@ * package mediaserver * * RTMP 服务（用于微信小程序 live-pusher/live-player） - * 功能： - * 1. 接收小程序推流 - * 2. 提供拉流服务 - * 3. 支持 HTTP-FLV 协议 + * 使用 github.com/yutopp/go-rtmp 实现完整的 RTMP 协议 * - * 注意：当前版本使用简化实现，实际 RTMP 协议处理需要集成完整的 RTMP 服务器 - * 生产环境建议使用 nginx-rtmp 或独立部署 livego + * 功能： + * 1. 接收小程序推流（publish） + * 2. 提供拉流服务（play） + * 3. 支持 HTTP-FLV 协议 */ package mediaserver import ( + "bytes" "crypto/hmac" "crypto/sha1" "encoding/base64" @@ -21,22 +21,33 @@ import ( "log" "net" "net/http" + "strings" "sync" "time" "github.com/spf13/viper" + "github.com/yutopp/go-flv" + flvtag "github.com/yutopp/go-flv/tag" + "github.com/yutopp/go-rtmp" + rtmpmsg "github.com/yutopp/go-rtmp/message" ) // RTMPServer RTMP 服务器 type RTMPServer struct { - config *MediaServerConfig - mu sync.RWMutex - running bool - streams map[string]*RTMPStream // streamID -> stream - + config *MediaServerConfig + mu sync.RWMutex + running bool + streams map[string]*RTMPStream // streamID -> stream + streamsMu sync.RWMutex + // 服务器监听器 rtmpListener net.Listener httpServer *http.Server + rtmpServer *rtmp.Server + + // 订阅者管理 + subscribers map[string]map[string]*Subscriber // streamID -> subscriberID -> subscriber + subMu sync.RWMutex } // RTMPStream RTMP 流信息 @@ -49,17 +60,33 @@ type RTMPStream struct { FLVURL string `json:"flv_url"` CreatedAt time.Time `json:"created_at"` IsActive bool `json:"is_active"` - - // 流数据缓存 - dataChan chan []byte `json:"-"` - stopChan chan struct{} `json:"-"` + + // 流数据 + mu sync.RWMutex + flvHeader []byte // FLV header + metaData []byte // FLV metadata tag + videoHeader []byte // Video sequence header (AVC) + audioHeader []byte // Audio sequence header (AAC) + gopCache [][]byte // GOP cache for new subscribers + + // 控制 + stopChan chan struct{} `json:"-"` +} + +// Subscriber 订阅者 +type Subscriber struct { + ID string + StreamID string + DataChan chan []byte + Done chan struct{} } // NewRTMPServer 创建 RTMP 服务器 func NewRTMPServer(config *MediaServerConfig) *RTMPServer { return &RTMPServer{ - config: config, - streams: make(map[string]*RTMPStream), + config: config, + streams: make(map[string]*RTMPStream), + subscribers: make(map[string]map[string]*Subscriber), } } @@ -72,149 +99,467 @@ func (r *RTMPServer) Start() { } r.running = true r.mu.Unlock() - - // 启动 RTMP 服务器（简化版） + + // 启动 RTMP 服务器 go r.startRTMPServer() - + // 启动 HTTP-FLV 服务器 go r.startHTTPFLVServer() - + // 启动流清理任务 go r.cleanupTask() - - log.Printf("🚀 [RTMP] RTMP 服务已启动 | RTMP端口: %d | HTTP-FLV端口: %d", + + log.Printf("🚀 [RTMP] RTMP 服务已启动 | RTMP端口: %d | HTTP-FLV端口: %d", r.config.RTMPPort, r.config.HTTPFLVPort) } -// startRTMPServer 启动 RTMP 服务（简化版） +// 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 + + // 创建 RTMP 服务器 + r.rtmpServer = rtmp.NewServer(&rtmp.ServerConfig{ + OnConnect: func(conn net.Conn) (io.ReadWriteCloser, *rtmp.ConnConfig) { + log.Printf("🎬 [RTMP] 新连接: %s", conn.RemoteAddr().String()) + return conn, &rtmp.ConnConfig{ + Handler: &RTMPHandler{ + server: r, + conn: conn, + }, + ControlState: rtmp.StreamControlStateConfig{ + DefaultBandwidthWindowSize: 6 * 1024 * 1024, + }, } - log.Printf("⚠️ [RTMP] 接受连接失败: %v", err) - continue + }, + }) + + // 启动服务 + if err := r.rtmpServer.Serve(r.rtmpListener); err != nil { + if r.running { + log.Printf("❌ [RTMP] 服务异常: %v", err) } - - // 处理 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) +// RTMPHandler RTMP 连接处理器 +type RTMPHandler struct { + rtmp.DefaultHandler + server *RTMPServer + conn net.Conn + streamID string + isPublish bool +} + +// OnServe 连接建立时调用 +func (h *RTMPHandler) OnServe(conn *rtmp.Conn) { + log.Printf("🎬 [RTMP] OnServe: %s", h.conn.RemoteAddr().String()) +} + +// OnConnect 处理 connect 命令 +func (h *RTMPHandler) OnConnect(timestamp uint32, cmd *rtmpmsg.NetConnectionConnect) error { + log.Printf("🎬 [RTMP] OnConnect: app=%v", cmd.Command.App) + return nil +} + +// OnCreateStream 处理 createStream 命令 +func (h *RTMPHandler) OnCreateStream(timestamp uint32, cmd *rtmpmsg.NetConnectionCreateStream) error { + log.Printf("🎬 [RTMP] OnCreateStream") + return nil +} + +// OnReleaseStream 处理 releaseStream 命令 +func (h *RTMPHandler) OnReleaseStream(timestamp uint32, cmd *rtmpmsg.NetConnectionReleaseStream) error { + streamName := cmd.StreamName + if idx := strings.Index(streamName, "?"); idx != -1 { + streamName = streamName[:idx] + } + log.Printf("🎬 [RTMP] OnReleaseStream: %s", streamName) + return nil +} + +// OnDeleteStream 处理 deleteStream 命令 +func (h *RTMPHandler) OnDeleteStream(timestamp uint32, cmd *rtmpmsg.NetStreamDeleteStream) error { + log.Printf("🎬 [RTMP] OnDeleteStream") + return nil +} + +// OnFCPublish 处理 FCPublish 命令 +func (h *RTMPHandler) OnFCPublish(timestamp uint32, cmd *rtmpmsg.NetStreamFCPublish) error { + log.Printf("🎬 [RTMP] OnFCPublish: %s", cmd.StreamName) + return nil +} + +// OnFCUnpublish 处理 FCUnpublish 命令 +func (h *RTMPHandler) OnFCUnpublish(timestamp uint32, cmd *rtmpmsg.NetStreamFCUnpublish) error { + log.Printf("🎬 [RTMP] OnFCUnpublish: %s", cmd.StreamName) + return nil +} + +// OnPublish 处理 publish 命令 +func (h *RTMPHandler) OnPublish(_ *rtmp.StreamContext, timestamp uint32, cmd *rtmpmsg.NetStreamPublish) error { + log.Printf("🎬 [RTMP] OnPublish: name=%s, type=%s", cmd.PublishingName, cmd.PublishingType) + + // 解析 stream name，去掉 token 参数 + streamName := cmd.PublishingName + if idx := strings.Index(streamName, "?"); idx != -1 { + streamName = streamName[:idx] + } + + h.streamID = streamName + h.isPublish = true + + log.Printf("🎬 [RTMP] 解析后的流ID: %s", h.streamID) + + // 检查流是否已注册 + h.server.streamsMu.RLock() + stream, exists := h.server.streams[h.streamID] + h.server.streamsMu.RUnlock() + + if !exists { + log.Printf("⚠️ [RTMP] 流未注册，自动创建: %s", h.streamID) + // 自动创建流（用于测试） + stream = &RTMPStream{ + ID: h.streamID, + CreatedAt: time.Now(), + IsActive: true, + stopChan: make(chan struct{}), + gopCache: make([][]byte, 0), + } + h.server.streamsMu.Lock() + h.server.streams[h.streamID] = stream + h.server.streamsMu.Unlock() + } else { + log.Printf("✅ [RTMP] 找到已注册的流: %s", h.streamID) + } + + stream.mu.Lock() + stream.IsActive = true + stream.mu.Unlock() + + log.Printf("✅ [RTMP] 开始推流: %s", h.streamID) + + return nil +} + +// OnPlay 处理 play 命令 +func (h *RTMPHandler) OnPlay(ctx *rtmp.StreamContext, timestamp uint32, cmd *rtmpmsg.NetStreamPlay) error { + log.Printf("🎬 [RTMP] OnPlay: name=%s", cmd.StreamName) + + h.streamID = cmd.StreamName + h.isPublish = false + + // 检查流是否存在 + h.server.streamsMu.RLock() + stream, exists := h.server.streams[h.streamID] + h.server.streamsMu.RUnlock() + + if !exists || !stream.IsActive { + log.Printf("⚠️ [RTMP] 流不存在或未激活: %s", h.streamID) + return fmt.Errorf("stream not found: %s", h.streamID) + } + + log.Printf("✅ [RTMP] 开始播放: %s", h.streamID) + + return nil +} + +// OnSetDataFrame 处理 metadata +func (h *RTMPHandler) OnSetDataFrame(timestamp uint32, data *rtmpmsg.NetStreamSetDataFrame) error { + log.Printf("🎬 [RTMP] OnSetDataFrame") + + if h.streamID == "" { + return nil + } + + h.server.streamsMu.RLock() + stream, exists := h.server.streams[h.streamID] + h.server.streamsMu.RUnlock() + + if exists && len(data.Payload) > 0 { + stream.mu.Lock() + stream.metaData = data.Payload + stream.mu.Unlock() + } + + return nil +} + +// OnAudio 处理音频数据 +func (h *RTMPHandler) OnAudio(timestamp uint32, payload io.Reader) error { + if h.streamID == "" || !h.isPublish { + return nil + } + + // 读取音频数据 + data, err := io.ReadAll(payload) + if err != nil { + return err + } + + if len(data) == 0 { + return nil + } + + h.server.streamsMu.RLock() + stream, exists := h.server.streams[h.streamID] + h.server.streamsMu.RUnlock() + + if !exists { + return nil + } + + // 保存 AAC sequence header + if len(data) > 1 { + soundFormat := (data[0] >> 4) & 0x0f + if soundFormat == 10 { // AAC + aacPacketType := data[1] + if aacPacketType == 0 { // Sequence header + stream.mu.Lock() + stream.audioHeader = make([]byte, len(data)) + copy(stream.audioHeader, data) + stream.mu.Unlock() + } + } + } + + // 创建 FLV audio tag + flvData := h.createFLVAudioTag(timestamp, data) + if flvData != nil { + h.broadcastToSubscribers(h.streamID, flvData) + } + + return nil +} + +// OnVideo 处理视频数据 +func (h *RTMPHandler) OnVideo(timestamp uint32, payload io.Reader) error { + if h.streamID == "" || !h.isPublish { + return nil + } + + // 读取视频数据 + data, err := io.ReadAll(payload) + if err != nil { + return err + } + + if len(data) == 0 { + return nil + } + + h.server.streamsMu.RLock() + stream, exists := h.server.streams[h.streamID] + h.server.streamsMu.RUnlock() + + if !exists { + return nil + } + + // 解析视频帧信息 + frameType := (data[0] >> 4) & 0x0f + codecID := data[0] & 0x0f + + // 保存 AVC sequence header + if codecID == 7 && len(data) > 1 { // AVC + avcPacketType := data[1] + if avcPacketType == 0 { // Sequence header + stream.mu.Lock() + stream.videoHeader = make([]byte, len(data)) + copy(stream.videoHeader, data) + stream.mu.Unlock() + } + + // 关键帧时清空 GOP 缓存 + if frameType == 1 { // Keyframe + stream.mu.Lock() + stream.gopCache = make([][]byte, 0) + stream.mu.Unlock() + } + } + + // 创建 FLV video tag + flvData := h.createFLVVideoTag(timestamp, data) + if flvData != nil { + // 缓存到 GOP + stream.mu.Lock() + stream.gopCache = append(stream.gopCache, flvData) + // 限制 GOP 缓存大小 + if len(stream.gopCache) > 300 { + stream.gopCache = stream.gopCache[len(stream.gopCache)-300:] + } + stream.mu.Unlock() + + h.broadcastToSubscribers(h.streamID, flvData) + } + + return nil +} + +// OnUnknownMessage 处理未知消息 +func (h *RTMPHandler) OnUnknownMessage(timestamp uint32, msg rtmpmsg.Message) error { + return nil +} + +// OnUnknownCommandMessage 处理未知命令消息 +func (h *RTMPHandler) OnUnknownCommandMessage(timestamp uint32, cmd *rtmpmsg.CommandMessage) error { + return nil +} + +// OnUnknownDataMessage 处理未知数据消息 +func (h *RTMPHandler) OnUnknownDataMessage(timestamp uint32, data *rtmpmsg.DataMessage) error { + return nil +} + +// createFLVAudioTag 创建 FLV 音频 tag +func (h *RTMPHandler) createFLVAudioTag(timestamp uint32, data []byte) []byte { + // FLV tag 格式: + // TagType (1 byte): 8 = audio + // DataSize (3 bytes): big-endian + // Timestamp (3 bytes): big-endian + // TimestampExtended (1 byte) + // StreamID (3 bytes): always 0 + // Data + // PreviousTagSize (4 bytes): big-endian + + dataSize := len(data) + tagSize := 11 + dataSize + 4 // header + data + previous tag size + + tag := make([]byte, tagSize) + + // Tag type + tag[0] = 8 // Audio + + // Data size (24-bit big-endian) + tag[1] = byte((dataSize >> 16) & 0xff) + tag[2] = byte((dataSize >> 8) & 0xff) + tag[3] = byte(dataSize & 0xff) + + // Timestamp (24-bit big-endian) + tag[4] = byte((timestamp >> 16) & 0xff) + tag[5] = byte((timestamp >> 8) & 0xff) + tag[6] = byte(timestamp & 0xff) + + // Timestamp extended + tag[7] = byte((timestamp >> 24) & 0xff) + + // Stream ID (always 0) + tag[8] = 0 + tag[9] = 0 + tag[10] = 0 + + // Data + copy(tag[11:], data) + + // Previous tag size (11 + dataSize) + prevTagSize := 11 + dataSize + tag[11+dataSize] = byte((prevTagSize >> 24) & 0xff) + tag[11+dataSize+1] = byte((prevTagSize >> 16) & 0xff) + tag[11+dataSize+2] = byte((prevTagSize >> 8) & 0xff) + tag[11+dataSize+3] = byte(prevTagSize & 0xff) + + return tag +} + +// createFLVVideoTag 创建 FLV 视频 tag +func (h *RTMPHandler) createFLVVideoTag(timestamp uint32, data []byte) []byte { + dataSize := len(data) + tagSize := 11 + dataSize + 4 + + tag := make([]byte, tagSize) + + // Tag type + tag[0] = 9 // Video + + // Data size (24-bit big-endian) + tag[1] = byte((dataSize >> 16) & 0xff) + tag[2] = byte((dataSize >> 8) & 0xff) + tag[3] = byte(dataSize & 0xff) + + // Timestamp (24-bit big-endian) + tag[4] = byte((timestamp >> 16) & 0xff) + tag[5] = byte((timestamp >> 8) & 0xff) + tag[6] = byte(timestamp & 0xff) + + // Timestamp extended + tag[7] = byte((timestamp >> 24) & 0xff) + + // Stream ID (always 0) + tag[8] = 0 + tag[9] = 0 + tag[10] = 0 + + // Data + copy(tag[11:], data) + + // Previous tag size + prevTagSize := 11 + dataSize + tag[11+dataSize] = byte((prevTagSize >> 24) & 0xff) + tag[11+dataSize+1] = byte((prevTagSize >> 16) & 0xff) + tag[11+dataSize+2] = byte((prevTagSize >> 8) & 0xff) + tag[11+dataSize+3] = byte(prevTagSize & 0xff) + + return tag +} + +// broadcastToSubscribers 向所有订阅者广播数据 +func (h *RTMPHandler) broadcastToSubscribers(streamID string, data []byte) { + h.server.subMu.RLock() + subs, exists := h.server.subscribers[streamID] + if !exists { + h.server.subMu.RUnlock() 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 + + for _, sub := range subs { + select { + case sub.DataChan <- data: + default: + // 缓冲区满，跳过 } - 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 } + h.server.subMu.RUnlock() } -// 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) +// OnClose 连接关闭 +func (h *RTMPHandler) OnClose() { + log.Printf("🎬 [RTMP] OnClose: stream=%s, publish=%v", h.streamID, h.isPublish) + + if h.isPublish && h.streamID != "" { + h.server.streamsMu.Lock() + if stream, exists := h.server.streams[h.streamID]; exists { + stream.mu.Lock() + stream.IsActive = false + stream.mu.Unlock() + } + h.server.streamsMu.Unlock() + log.Printf("⏹️ [RTMP] 停止推流: %s", h.streamID) } - - // 验证版本号 (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) } @@ -224,51 +569,153 @@ func (r *RTMPServer) startHTTPFLVServer() { 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个字符 + if len(path) < 7 { 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() + + r.streamsMu.RLock() stream, exists := r.streams[streamPath] - r.mu.RUnlock() - + r.streamsMu.RUnlock() + if !exists || !stream.IsActive { http.Error(w, "Stream not found", http.StatusNotFound) return } - + + log.Printf("🎬 [RTMP] FLV 请求: %s", streamPath) + // 设置响应头 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) - + w.Header().Set("Connection", "keep-alive") + w.Header().Set("Cache-Control", "no-cache") + // 发送 FLV header + // FLV signature (3 bytes) + version (1 byte) + flags (1 byte) + header size (4 bytes) flvHeader := []byte{0x46, 0x4C, 0x56, 0x01, 0x05, 0x00, 0x00, 0x00, 0x09, 0x00, 0x00, 0x00, 0x00} w.Write(flvHeader) - - // 保持连接直到客户端断开 - <-req.Context().Done() + + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + + // 创建订阅者 + subID := fmt.Sprintf("%d", time.Now().UnixNano()) + sub := &Subscriber{ + ID: subID, + StreamID: streamPath, + DataChan: make(chan []byte, 100), + Done: make(chan struct{}), + } + + // 注册订阅者 + r.subMu.Lock() + if r.subscribers[streamPath] == nil { + r.subscribers[streamPath] = make(map[string]*Subscriber) + } + r.subscribers[streamPath][subID] = sub + r.subMu.Unlock() + + // 清理 + defer func() { + r.subMu.Lock() + delete(r.subscribers[streamPath], subID) + r.subMu.Unlock() + close(sub.Done) + }() + + // 发送缓存的头信息 + stream.mu.RLock() + + // 发送 metadata + if len(stream.metaData) > 0 { + w.Write(stream.metaData) + } + + // 发送视频 sequence header + if len(stream.videoHeader) > 0 { + flvTag := createFLVTagFromData(9, 0, stream.videoHeader) + w.Write(flvTag) + } + + // 发送音频 sequence header + if len(stream.audioHeader) > 0 { + flvTag := createFLVTagFromData(8, 0, stream.audioHeader) + w.Write(flvTag) + } + + // 发送 GOP 缓存 + for _, data := range stream.gopCache { + w.Write(data) + } + stream.mu.RUnlock() + + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + + // 持续发送数据 + for { + select { + case <-req.Context().Done(): + return + case data := <-sub.DataChan: + if _, err := w.Write(data); err != nil { + return + } + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + } + } +} + +// createFLVTagFromData 从原始数据创建 FLV tag +func createFLVTagFromData(tagType byte, timestamp uint32, data []byte) []byte { + dataSize := len(data) + tagSize := 11 + dataSize + 4 + + tag := make([]byte, tagSize) + + tag[0] = tagType + tag[1] = byte((dataSize >> 16) & 0xff) + tag[2] = byte((dataSize >> 8) & 0xff) + tag[3] = byte(dataSize & 0xff) + tag[4] = byte((timestamp >> 16) & 0xff) + tag[5] = byte((timestamp >> 8) & 0xff) + tag[6] = byte(timestamp & 0xff) + tag[7] = byte((timestamp >> 24) & 0xff) + tag[8] = 0 + tag[9] = 0 + tag[10] = 0 + + copy(tag[11:], data) + + prevTagSize := 11 + dataSize + tag[11+dataSize] = byte((prevTagSize >> 24) & 0xff) + tag[11+dataSize+1] = byte((prevTagSize >> 16) & 0xff) + tag[11+dataSize+2] = byte((prevTagSize >> 8) & 0xff) + tag[11+dataSize+3] = byte(prevTagSize & 0xff) + + return tag } // 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() - + + r.streamsMu.RLock() + defer r.streamsMu.RUnlock() + streams := make([]map[string]interface{}, 0) for _, s := range r.streams { if s.IsActive { @@ -281,7 +728,7 @@ func (r *RTMPServer) handleStreamsAPI(w http.ResponseWriter, req *http.Request) }) } } - + fmt.Fprintf(w, `{"streams":%d,"data":%v}`, len(streams), streams) } @@ -289,51 +736,73 @@ func (r *RTMPServer) handleStreamsAPI(w http.ResponseWriter, req *http.Request) func (r *RTMPServer) Stop() { r.mu.Lock() defer r.mu.Unlock() - + r.running = false - + + // 关闭 RTMP 服务器 + if r.rtmpServer != nil { + r.rtmpServer.Close() + } + // 关闭 RTMP 监听器 if r.rtmpListener != nil { r.rtmpListener.Close() } - + // 关闭 HTTP 服务器 if r.httpServer != nil { r.httpServer.Close() } - + // 关闭所有流 + r.streamsMu.Lock() for _, stream := range r.streams { + stream.IsActive = false if stream.stopChan != nil { - close(stream.stopChan) + select { + case <-stream.stopChan: + default: + close(stream.stopChan) + } } } r.streams = make(map[string]*RTMPStream) - + r.streamsMu.Unlock() + + // 关闭所有订阅者 + r.subMu.Lock() + for _, subs := range r.subscribers { + for _, sub := range subs { + close(sub.DataChan) + } + } + r.subscribers = make(map[string]map[string]*Subscriber) + r.subMu.Unlock() + log.Println("🛑 [RTMP] 已停止") } // GenerateStreamURLs 为用户生成推拉流地址 func (r *RTMPServer) GenerateStreamURLs(roomID, userID string) (*RTMPStream, error) { - r.mu.Lock() - defer r.mu.Unlock() - + r.mu.RLock() if !r.running { + r.mu.RUnlock() return nil, ErrRTMPNotReady } - + r.mu.RUnlock() + // 生成唯一的流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, @@ -344,28 +813,31 @@ func (r *RTMPServer) GenerateStreamURLs(roomID, userID string) (*RTMPStream, err 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{}), + gopCache: make([][]byte, 0), } - + + r.streamsMu.Lock() r.streams[streamID] = stream + r.streamsMu.Unlock() + 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() + r.streamsMu.RLock() + defer r.streamsMu.RUnlock() return r.streams[streamID] } // GetStreamsByRoom 获取房间内所有流 func (r *RTMPServer) GetStreamsByRoom(roomID string) []*RTMPStream { - r.mu.RLock() - defer r.mu.RUnlock() - + r.streamsMu.RLock() + defer r.streamsMu.RUnlock() + streams := make([]*RTMPStream, 0) for _, s := range r.streams { if s.RoomID == roomID && s.IsActive { @@ -377,9 +849,9 @@ func (r *RTMPServer) GetStreamsByRoom(roomID string) []*RTMPStream { // GetStreamsByUser 获取用户的所有流 func (r *RTMPServer) GetStreamsByUser(userID string) []*RTMPStream { - r.mu.RLock() - defer r.mu.RUnlock() - + r.streamsMu.RLock() + defer r.streamsMu.RUnlock() + streams := make([]*RTMPStream, 0) for _, s := range r.streams { if s.UserID == userID && s.IsActive { @@ -391,57 +863,84 @@ func (r *RTMPServer) GetStreamsByUser(userID string) []*RTMPStream { // RemoveStream 移除流 func (r *RTMPServer) RemoveStream(streamID string) { - r.mu.Lock() - defer r.mu.Unlock() - + r.streamsMu.Lock() + defer r.streamsMu.Unlock() + if stream, exists := r.streams[streamID]; exists { stream.IsActive = false if stream.stopChan != nil { - close(stream.stopChan) + select { + case <-stream.stopChan: + default: + close(stream.stopChan) + } } delete(r.streams, streamID) log.Printf("🎬 [RTMP] 移除流 | Stream:%s", streamID) } + + // 移除订阅者 + r.subMu.Lock() + delete(r.subscribers, streamID) + r.subMu.Unlock() } // RemoveStreamsByUser 移除用户的所有流 func (r *RTMPServer) RemoveStreamsByUser(userID string) { - r.mu.Lock() - defer r.mu.Unlock() - + r.streamsMu.Lock() + defer r.streamsMu.Unlock() + for streamID, stream := range r.streams { if stream.UserID == userID { stream.IsActive = false if stream.stopChan != nil { - close(stream.stopChan) + select { + case <-stream.stopChan: + default: + close(stream.stopChan) + } } delete(r.streams, streamID) log.Printf("🎬 [RTMP] 移除用户流 | User:%s Stream:%s", userID, streamID) + + // 移除订阅者 + r.subMu.Lock() + delete(r.subscribers, streamID) + r.subMu.Unlock() } } } // RemoveStreamsByRoom 移除房间的所有流 func (r *RTMPServer) RemoveStreamsByRoom(roomID string) { - r.mu.Lock() - defer r.mu.Unlock() - + r.streamsMu.Lock() + defer r.streamsMu.Unlock() + for streamID, stream := range r.streams { if stream.RoomID == roomID { stream.IsActive = false if stream.stopChan != nil { - close(stream.stopChan) + select { + case <-stream.stopChan: + default: + close(stream.stopChan) + } } delete(r.streams, streamID) log.Printf("🎬 [RTMP] 移除房间流 | Room:%s Stream:%s", roomID, streamID) + + // 移除订阅者 + r.subMu.Lock() + delete(r.subscribers, streamID) + r.subMu.Unlock() } } } // GetStreamCount 获取流数量 func (r *RTMPServer) GetStreamCount() int { - r.mu.RLock() - defer r.mu.RUnlock() + r.streamsMu.RLock() + defer r.streamsMu.RUnlock() return len(r.streams) } @@ -449,13 +948,13 @@ func (r *RTMPServer) GetStreamCount() int { 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() @@ -467,15 +966,19 @@ func (r *RTMPServer) cleanupTask() { // cleanupExpiredStreams 清理过期的流 func (r *RTMPServer) cleanupExpiredStreams() { - r.mu.Lock() - defer r.mu.Unlock() - + r.streamsMu.Lock() + defer r.streamsMu.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) + select { + case <-stream.stopChan: + default: + close(stream.stopChan) + } } delete(r.streams, streamID) log.Printf("🗑️ [RTMP] 清理过期流 | Stream:%s", streamID) @@ -488,7 +991,7 @@ 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)) @@ -499,3 +1002,10 @@ func ValidateStreamToken(streamID, token string) bool { expectedToken := generateStreamToken(streamID) return token == expectedToken } + +// 确保不使用的导入被使用 +var ( + _ = bytes.NewReader + _ = flv.NewEncoder + _ = flvtag.TagTypeVideo +)