From 2d03f4b27981cf97c54ff9af5b5df6a62fd3c8dc Mon Sep 17 00:00:00 2001 From: xugo Date: Tue, 28 Apr 2026 11:31:57 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(api):=20=E5=85=BC=E5=AE=B9=20zlm=20?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=E4=B8=8E=E5=9B=9E=E8=B0=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/config.go | 61 ++- server/hook_builtin_http_plugin.go | 35 ++ server/http_notify.go | 203 ++++++- server/router.go | 1 + server/router_test.go | 24 +- server/router_zlm_compat.go | 328 ++++++++++++ server/server.go | 37 ++ server/zlm_compat_config.go | 111 ++++ server/zlm_compat_ffmpeg.go | 160 ++++++ server/zlm_compat_test.go | 830 +++++++++++++++++++++++++++++ server/zlm_compat_types.go | 253 +++++++++ 11 files changed, 2006 insertions(+), 37 deletions(-) create mode 100644 server/router_zlm_compat.go create mode 100644 server/zlm_compat_config.go create mode 100644 server/zlm_compat_ffmpeg.go create mode 100644 server/zlm_compat_test.go create mode 100644 server/zlm_compat_types.go diff --git a/config/config.go b/config/config.go index 9262297..a97bf03 100644 --- a/config/config.go +++ b/config/config.go @@ -85,22 +85,53 @@ type GB28181MediaConfig struct { MultiPortMaxIncrement uint16 `json:"multi_port_max_increment"` // 多端口范围 ListenPort+1至ListenPort+MultiPortMax } +// ZlmCompatHookConfig ZLM 兼容 hook URL 配置 +// 为什么独立结构体：隔离 ZLM 适配层，lalmax 原有字段保持不变 +type ZlmCompatHookConfig struct { + ZlmOnStreamChanged string `json:"zlm_on_stream_changed"` + ZlmOnServerKeepalive string `json:"zlm_on_server_keepalive"` + ZlmOnStreamNoneReader string `json:"zlm_on_stream_none_reader"` + ZlmOnRtpServerTimeout string `json:"zlm_on_rtp_server_timeout"` + ZlmOnRecordMp4 string `json:"zlm_on_record_mp4"` + ZlmOnPublish string `json:"zlm_on_publish"` + ZlmOnPlay string `json:"zlm_on_play"` + ZlmOnStreamNotFound string `json:"zlm_on_stream_not_found"` + ZlmOnServerStarted string `json:"zlm_on_server_started"` +} + +// HasZlmHooks 任一 ZLM 兼容 hook 字段有值即返回 true +// 为什么：ZLM 回调与 lalmax 原有回调二选一，此方法为判断条件 +func (c ZlmCompatHookConfig) HasZlmHooks() bool { + return c.ZlmOnStreamChanged != "" || + c.ZlmOnServerKeepalive != "" || + c.ZlmOnStreamNoneReader != "" || + c.ZlmOnRtpServerTimeout != "" || + c.ZlmOnRecordMp4 != "" || + c.ZlmOnPublish != "" || + c.ZlmOnPlay != "" || + c.ZlmOnStreamNotFound != "" +} + type HttpNotifyConfig struct { - Enable bool `json:"enable"` - UpdateIntervalSec int `json:"update_interval_sec"` - OnServerStart string `json:"on_server_start"` - OnUpdate string `json:"on_update"` - OnGroupStart string `json:"on_group_start"` - OnGroupStop string `json:"on_group_stop"` - OnStreamActive string `json:"on_stream_active"` - OnPubStart string `json:"on_pub_start"` - OnPubStop string `json:"on_pub_stop"` - OnSubStart string `json:"on_sub_start"` - OnSubStop string `json:"on_sub_stop"` - OnRelayPullStart string `json:"on_relay_pull_start"` - OnRelayPullStop string `json:"on_relay_pull_stop"` - OnRtmpConnect string `json:"on_rtmp_connect"` - OnHlsMakeTs string `json:"on_hls_make_ts"` + Enable bool `json:"enable"` + UpdateIntervalSec int `json:"update_interval_sec"` + KeepaliveIntervalSec int `json:"keepalive_interval_sec"` + OnServerStart string `json:"on_server_start"` + OnUpdate string `json:"on_update"` + OnGroupStart string `json:"on_group_start"` + OnGroupStop string `json:"on_group_stop"` + OnStreamActive string `json:"on_stream_active"` + OnPubStart string `json:"on_pub_start"` + OnPubStop string `json:"on_pub_stop"` + OnSubStart string `json:"on_sub_start"` + OnSubStop string `json:"on_sub_stop"` + OnRelayPullStart string `json:"on_relay_pull_start"` + OnRelayPullStop string `json:"on_relay_pull_stop"` + OnRtmpConnect string `json:"on_rtmp_connect"` + OnHlsMakeTs string `json:"on_hls_make_ts"` + + // --- ZLM 兼容 hook 配置 --- + ZlmCompatHookConfig } type LogicConfig struct { diff --git a/server/hook_builtin_http_plugin.go b/server/hook_builtin_http_plugin.go index 008da1c..bee1e9e 100644 --- a/server/hook_builtin_http_plugin.go +++ b/server/hook_builtin_http_plugin.go @@ -24,6 +24,9 @@ func (p *hookBuiltinHTTPPlugin) OnHookEvent(event HookEvent) error { if p.hub.cfg.OnServerStart != "" { p.hub.asyncPostEvent(p.hub.cfg.OnServerStart, event) } + if p.hub.cfg.ZlmOnServerStarted != "" { + p.hub.asyncPostEvent(p.hub.cfg.ZlmOnServerStarted, event) + } case HookEventUpdate: if p.hub.cfg.OnUpdate != "" { p.hub.asyncPostEvent(p.hub.cfg.OnUpdate, event) @@ -72,6 +75,38 @@ func (p *hookBuiltinHTTPPlugin) OnHookEvent(event HookEvent) error { if p.hub.cfg.OnHlsMakeTs != "" { p.hub.asyncPostEvent(p.hub.cfg.OnHlsMakeTs, event) } + case HookEventStreamChanged: + if p.hub.cfg.ZlmOnStreamChanged != "" { + p.hub.asyncPostEvent(p.hub.cfg.ZlmOnStreamChanged, event) + } + case HookEventServerKeepalive: + if p.hub.cfg.ZlmOnServerKeepalive != "" { + p.hub.asyncPostEvent(p.hub.cfg.ZlmOnServerKeepalive, event) + } + case HookEventStreamNoneReader: + if p.hub.cfg.ZlmOnStreamNoneReader != "" { + p.hub.asyncPostEvent(p.hub.cfg.ZlmOnStreamNoneReader, event) + } + case HookEventRtpServerTimeout: + if p.hub.cfg.ZlmOnRtpServerTimeout != "" { + p.hub.asyncPostEvent(p.hub.cfg.ZlmOnRtpServerTimeout, event) + } + case HookEventRecordMp4: + if p.hub.cfg.ZlmOnRecordMp4 != "" { + p.hub.asyncPostEvent(p.hub.cfg.ZlmOnRecordMp4, event) + } + case HookEventPublish: + if p.hub.cfg.ZlmOnPublish != "" { + p.hub.asyncPostEvent(p.hub.cfg.ZlmOnPublish, event) + } + case HookEventPlay: + if p.hub.cfg.ZlmOnPlay != "" { + p.hub.asyncPostEvent(p.hub.cfg.ZlmOnPlay, event) + } + case HookEventStreamNotFound: + if p.hub.cfg.ZlmOnStreamNotFound != "" { + p.hub.asyncPostEvent(p.hub.cfg.ZlmOnStreamNotFound, event) + } } return nil diff --git a/server/http_notify.go b/server/http_notify.go index 83f646a..b00114d 100644 --- a/server/http_notify.go +++ b/server/http_notify.go @@ -69,21 +69,33 @@ type HookEvent struct { } const ( - HookEventServerStart = "on_server_start" - HookEventUpdate = "on_update" - HookEventGroupStart = "on_group_start" - HookEventGroupStop = "on_group_stop" - HookEventStreamActive = "on_stream_active" - HookEventPubStart = "on_pub_start" - HookEventPubStop = "on_pub_stop" - HookEventSubStart = "on_sub_start" - HookEventSubStop = "on_sub_stop" - HookEventRelayPullStart = "on_relay_pull_start" - HookEventRelayPullStop = "on_relay_pull_stop" - HookEventRtmpConnect = "on_rtmp_connect" - HookEventHlsMakeTs = "on_hls_make_ts" + HookEventServerStart = "on_server_start" + HookEventUpdate = "on_update" + HookEventGroupStart = "on_group_start" + HookEventGroupStop = "on_group_stop" + HookEventStreamActive = "on_stream_active" + HookEventPubStart = "on_pub_start" + HookEventPubStop = "on_pub_stop" + HookEventSubStart = "on_sub_start" + HookEventSubStop = "on_sub_stop" + HookEventRelayPullStart = "on_relay_pull_start" + HookEventRelayPullStop = "on_relay_pull_stop" + HookEventRtmpConnect = "on_rtmp_connect" + HookEventHlsMakeTs = "on_hls_make_ts" + HookEventStreamChanged = "on_stream_changed" + HookEventServerKeepalive = "on_server_keepalive" + HookEventStreamNoneReader = "on_stream_none_reader" + HookEventRtpServerTimeout = "on_rtp_server_timeout" + HookEventRecordMp4 = "on_record_mp4" + HookEventPublish = "on_publish" + HookEventPlay = "on_play" + HookEventStreamNotFound = "on_stream_not_found" ) +// SubCountFn 查询指定流当前的 sub 数量 +// 为什么用回调：避免 HttpNotify 直接依赖 lalsvr，保持解耦 +type SubCountFn func(streamName string) int + type HttpNotify struct { cfg config.HttpNotifyConfig @@ -92,6 +104,8 @@ type HttpNotify struct { client *http.Client + subCountFn SubCountFn + eventID atomic.Int64 subID atomic.Int64 historyMux sync.RWMutex @@ -104,6 +118,20 @@ type HttpNotify struct { httpPosts map[string]*hookHTTPPostWorker } +// SetSubCountFn 注入 sub 数量查询函数，用于 on_stream_none_reader 判断 +func (h *HttpNotify) SetSubCountFn(fn SubCountFn) { + h.subCountFn = fn +} + +// UpdateZlmHookConfig 运行时更新 ZLM 兼容 hook 配置 +// 为什么：gb28181 通过 setServerConfig 动态设置 hook URL，需要立即生效 +func (h *HttpNotify) UpdateZlmHookConfig(zlmCfg config.ZlmCompatHookConfig) { + h.cfg.ZlmCompatHookConfig = zlmCfg + h.cfg.Enable = true + Log.Infof("zlm compat hook config updated. on_stream_changed=%s, on_server_keepalive=%s, on_publish=%s, on_play=%s", + zlmCfg.ZlmOnStreamChanged, zlmCfg.ZlmOnServerKeepalive, zlmCfg.ZlmOnPublish, zlmCfg.ZlmOnPlay) +} + func NewHttpNotify(cfg config.HttpNotifyConfig, serverId string) *HttpNotify { httpNotify := &HttpNotify{ cfg: cfg, @@ -155,21 +183,83 @@ func (h *HttpNotify) NotifyStreamActive(info HookGroupInfo) { func (h *HttpNotify) NotifyPubStart(info base.PubStartInfo) { info.ServerId = h.serverId h.publish(HookEventPubStart, info) + + if !h.cfg.HasZlmHooks() { + return + } + // --- ZLM 兼容：派生 on_publish + on_stream_changed --- + h.publish(HookEventPublish, ZlmOnPublishPayload{ + MediaServerID: h.serverId, + App: info.AppName, + Schema: info.Protocol, + Stream: info.StreamName, + Vhost: "__defaultVhost__", + }) + h.publish(HookEventStreamChanged, ZlmOnStreamChangedPayload{ + Regist: true, + App: info.AppName, + Stream: info.StreamName, + AppName: info.AppName, + StreamName: info.StreamName, + Schema: info.Protocol, + MediaServerID: h.serverId, + Vhost: "__defaultVhost__", + }) } func (h *HttpNotify) NotifyPubStop(info base.PubStopInfo) { info.ServerId = h.serverId h.publish(HookEventPubStop, info) + + if !h.cfg.HasZlmHooks() { + return + } + // --- ZLM 兼容：派生 on_stream_changed(regist=false) --- + h.publish(HookEventStreamChanged, ZlmOnStreamChangedPayload{ + Regist: false, + App: info.AppName, + Stream: info.StreamName, + AppName: info.AppName, + StreamName: info.StreamName, + Schema: info.Protocol, + MediaServerID: h.serverId, + Vhost: "__defaultVhost__", + }) } func (h *HttpNotify) NotifySubStart(info base.SubStartInfo) { info.ServerId = h.serverId h.publish(HookEventSubStart, info) + + if !h.cfg.HasZlmHooks() { + return + } + // --- ZLM 兼容：派生 on_play --- + h.publish(HookEventPlay, ZlmOnPlayPayload{ + MediaServerID: h.serverId, + App: info.AppName, + Schema: info.Protocol, + Stream: info.StreamName, + Vhost: "__defaultVhost__", + }) } func (h *HttpNotify) NotifySubStop(info base.SubStopInfo) { info.ServerId = h.serverId h.publish(HookEventSubStop, info) + + if h.cfg.ZlmOnStreamNoneReader == "" || h.subCountFn == nil { + return + } + // 检查该流是否已无观看者，触发 on_stream_none_reader + if h.subCountFn(info.StreamName) <= 0 { + h.NotifyStreamNoneReader(ZlmOnStreamNoneReaderPayload{ + App: info.AppName, + Schema: info.Protocol, + Stream: info.StreamName, + Vhost: "__defaultVhost__", + }) + } } func (h *HttpNotify) NotifyPullStart(info base.PullStartInfo) { @@ -192,6 +282,61 @@ func (h *HttpNotify) NotifyOnHlsMakeTs(info base.HlsMakeTsInfo) { h.publish(HookEventHlsMakeTs, info) } +func (h *HttpNotify) NotifyStreamChanged(info ZlmOnStreamChangedPayload) { + if info.MediaServerID == "" { + info.MediaServerID = h.serverId + } + h.publish(HookEventStreamChanged, info) +} + +func (h *HttpNotify) NotifyServerKeepalive() { + h.publish(HookEventServerKeepalive, ZlmOnServerKeepalivePayload{ + MediaServerID: h.serverId, + }) +} + +func (h *HttpNotify) NotifyStreamNoneReader(info ZlmOnStreamNoneReaderPayload) { + if info.MediaServerID == "" { + info.MediaServerID = h.serverId + } + h.publish(HookEventStreamNoneReader, info) +} + +func (h *HttpNotify) NotifyRtpServerTimeout(info ZlmOnRtpServerTimeoutPayload) { + if info.MediaServerID == "" { + info.MediaServerID = h.serverId + } + h.publish(HookEventRtpServerTimeout, info) +} + +func (h *HttpNotify) NotifyRecordMp4(info ZlmOnRecordMp4Payload) { + if info.MediaServerID == "" { + info.MediaServerID = h.serverId + } + h.publish(HookEventRecordMp4, info) +} + +func (h *HttpNotify) NotifyPublish(info ZlmOnPublishPayload) { + if info.MediaServerID == "" { + info.MediaServerID = h.serverId + } + h.publish(HookEventPublish, info) +} + +func (h *HttpNotify) NotifyPlay(info ZlmOnPlayPayload) { + if info.MediaServerID == "" { + info.MediaServerID = h.serverId + } + h.publish(HookEventPlay, info) +} + +func (h *HttpNotify) NotifyStreamNotFound(info ZlmOnStreamNotFoundPayload) { + if info.MediaServerID == "" { + info.MediaServerID = h.serverId + } + h.publish(HookEventStreamNotFound, info) +} + // ----- implement INotifyHandler interface ---------------------------------------------------------------------------- func (h *HttpNotify) OnServerStart(info base.LalInfo) { @@ -496,6 +641,38 @@ func populateHookEventMeta(event *HookEvent, info interface{}) { event.appName = v.App case base.HlsMakeTsInfo: event.streamName = v.StreamName + case ZlmOnStreamChangedPayload: + event.appName = v.App + event.streamName = v.Stream + if event.appName == "" { + event.appName = v.AppName + } + if event.streamName == "" { + event.streamName = v.StreamName + } + case ZlmOnStreamNoneReaderPayload: + event.appName = v.App + event.streamName = v.Stream + case ZlmOnRtpServerTimeoutPayload: + event.streamName = v.StreamID + case ZlmOnRecordMp4Payload: + event.appName = v.App + event.streamName = v.Stream + case ZlmOnPublishPayload: + event.appName = v.App + event.streamName = v.Stream + case ZlmOnPlayPayload: + event.appName = v.App + event.streamName = v.Stream + case ZlmOnStreamNotFoundPayload: + event.appName = v.App + event.streamName = v.Stream + if event.appName == "" { + event.appName = v.AppName + } + if event.streamName == "" { + event.streamName = v.StreamName + } } } diff --git a/server/router.go b/server/router.go index ac2ddd0..db26a7c 100644 --- a/server/router.go +++ b/server/router.go @@ -15,4 +15,5 @@ func (s *LalMaxServer) InitRouter(router *gin.Engine) { s.initHookRouter(router, auth) s.initStatRouter(router, auth) s.initCtrlRouter(router, auth) + s.initZlmCompatRouter(router, auth) } diff --git a/server/router_test.go b/server/router_test.go index b8ec5fb..b167316 100644 --- a/server/router_test.go +++ b/server/router_test.go @@ -399,7 +399,8 @@ func TestStopRelayPullAllowsGet(t *testing.T) { func TestHookHubRecentAndSubscribe(t *testing.T) { hub := NewHttpNotify(config.HttpNotifyConfig{}, "hub-test") - _, ch, cancel := hub.Subscribe(1) + // NotifyPubStart 会派生 on_stream_changed，需要足够缓冲 + _, ch, cancel := hub.Subscribe(8) defer cancel() hub.NotifyPubStart(base.PubStartInfo{}) @@ -413,12 +414,16 @@ func TestHookHubRecentAndSubscribe(t *testing.T) { t.Fatal("wait hook event timeout") } - events := hub.Recent(1) - if len(events) != 1 { - t.Fatalf("unexpected recent len: %d", len(events)) + events := hub.Recent(0) + found := false + for _, e := range events { + if e.Event == HookEventPubStart { + found = true + break + } } - if events[0].Event != HookEventPubStart { - t.Fatalf("unexpected recent event: %+v", events[0]) + if !found { + t.Fatalf("on_pub_start not found in recent events") } } @@ -755,7 +760,8 @@ func TestHookRecentEndpoint(t *testing.T) { svr.notifyHub.NotifyPubStop(base.PubStopInfo{}) r := httptest.NewRecorder() - req := httptest.NewRequest("GET", "/api/hook/recent?limit=1", nil) + // 用 event filter 精确查询，因为 NotifyPubStop 会派生 on_stream_changed + req := httptest.NewRequest("GET", "/api/hook/recent?limit=10&event=on_pub_stop", nil) svr.router.ServeHTTP(r, req) resp := r.Result() if resp.StatusCode != http.StatusOK { @@ -774,8 +780,8 @@ func TestHookRecentEndpoint(t *testing.T) { if out.ErrorCode != base.ErrorCodeSucc { t.Fatalf("unexpected response: %+v", out) } - if len(out.Data.Events) != 1 { - t.Fatalf("unexpected event count: %d", len(out.Data.Events)) + if len(out.Data.Events) < 1 { + t.Fatalf("expected at least 1 on_pub_stop event, got: %d", len(out.Data.Events)) } if out.Data.Events[0].Event != HookEventPubStop { t.Fatalf("unexpected event: %+v", out.Data.Events[0]) diff --git a/server/router_zlm_compat.go b/server/router_zlm_compat.go new file mode 100644 index 0000000..67b7561 --- /dev/null +++ b/server/router_zlm_compat.go @@ -0,0 +1,328 @@ +package server + +import ( + "encoding/json" + "net/http" + "strconv" + + "github.com/gin-gonic/gin" + "github.com/q191201771/lal/pkg/base" + "github.com/q191201771/lal/pkg/logic" + config "github.com/q191201771/lalmax/config" +) + +// initZlmCompatRouter 注册 /index/api/* ZLM 兼容路由 +// 为什么独立文件：隔离 ZLM 兼容层，不影响现有 lalmax API +func (s *LalMaxServer) initZlmCompatRouter(router *gin.Engine, handlers ...gin.HandlerFunc) { + zlm := router.Group("/index/api", handlers...) + zlm.POST("/openRtpServer", s.zlmOpenRtpServerHandler) + zlm.POST("/closeRtpServer", s.zlmCloseRtpServerHandler) + zlm.POST("/close_streams", s.zlmCloseStreamsHandler) + zlm.POST("/getServerConfig", s.zlmGetServerConfigHandler) + zlm.POST("/setServerConfig", s.zlmSetServerConfigHandler) + zlm.POST("/restartServer", s.zlmRestartServerHandler) + zlm.POST("/startRecord", s.zlmStartRecordHandler) + zlm.POST("/stopRecord", s.zlmStopRecordHandler) + zlm.POST("/addStreamProxy", s.zlmAddStreamProxyHandler) + zlm.POST("/getSnap", s.zlmGetSnapHandler) +} + +// ---------- openRtpServer ---------- + +func (s *LalMaxServer) zlmOpenRtpServerHandler(c *gin.Context) { + var req ZlmOpenRtpServerReq + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusOK, ZlmOpenRtpServerResp{Code: -300, Msg: "invalid params"}) + return + } + + isTcpFlag := 0 + if req.TCPMode > 0 { + isTcpFlag = 1 + } + + resp := s.rtpPubMgr.Start(base.ApiCtrlStartRtpPubReq{ + StreamName: req.StreamID, + Port: req.Port, + IsTcpFlag: isTcpFlag, + }) + + if resp.ErrorCode != base.ErrorCodeSucc { + c.JSON(http.StatusOK, ZlmOpenRtpServerResp{Code: -1, Msg: resp.Desp}) + return + } + + Log.Infof("zlm compat openRtpServer. stream_id=%s, port=%d", req.StreamID, resp.Data.Port) + c.JSON(http.StatusOK, ZlmOpenRtpServerResp{Code: 0, Port: resp.Data.Port}) +} + +// ---------- closeRtpServer ---------- + +func (s *LalMaxServer) zlmCloseRtpServerHandler(c *gin.Context) { + var req ZlmCloseRtpServerReq + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusOK, ZlmCloseRtpServerResp{Code: -300}) + return + } + + _, err := s.rtpPubMgr.Stop(req.StreamID, "") + if err != nil { + Log.Infof("zlm compat closeRtpServer not found. stream_id=%s", req.StreamID) + c.JSON(http.StatusOK, ZlmCloseRtpServerResp{Code: 0, Hit: 0}) + return + } + + Log.Infof("zlm compat closeRtpServer. stream_id=%s", req.StreamID) + c.JSON(http.StatusOK, ZlmCloseRtpServerResp{Code: 0, Hit: 1}) +} + +// ---------- close_streams ---------- + +func (s *LalMaxServer) zlmCloseStreamsHandler(c *gin.Context) { + var req ZlmCloseStreamsReq + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusOK, ZlmCloseStreamsResp{Code: -300}) + return + } + + streamName := req.Stream + if streamName == "" { + c.JSON(http.StatusOK, ZlmCloseStreamsResp{Code: 0, CountHit: 0, CountClosed: 0}) + return + } + + // 尝试通过 kick_session 关闭所有匹配的 session + groups := s.lalsvr.StatAllGroup() + hit := 0 + closed := 0 + for _, g := range groups { + if g.StreamName != streamName { + continue + } + hit++ + // 关闭 pub session + if g.StatPub.SessionId != "" { + resp := s.lalsvr.CtrlKickSession(base.ApiCtrlKickSessionReq{ + StreamName: streamName, + SessionId: g.StatPub.SessionId, + }) + if resp.ErrorCode == base.ErrorCodeSucc { + closed++ + } + } + } + + // 也尝试关闭 RTP pub session + if _, err := s.rtpPubMgr.Stop(streamName, ""); err == nil { + if hit == 0 { + hit++ + } + closed++ + } + + Log.Infof("zlm compat close_streams. stream=%s, hit=%d, closed=%d", streamName, hit, closed) + c.JSON(http.StatusOK, ZlmCloseStreamsResp{Code: 0, CountHit: hit, CountClosed: closed}) +} + +// ---------- getServerConfig ---------- + +func (s *LalMaxServer) zlmGetServerConfigHandler(c *gin.Context) { + cfg := buildZlmServerConfig(s.conf) + c.JSON(http.StatusOK, ZlmGetServerConfigResp{Code: 0, Data: []map[string]any{cfg}}) +} + +// ---------- setServerConfig ---------- + +func (s *LalMaxServer) zlmSetServerConfigHandler(c *gin.Context) { + var params map[string]*string + if err := c.ShouldBindJSON(¶ms); err != nil { + c.JSON(http.StatusOK, ZlmSetServerConfigResp{ + ZlmFixedHeader: ZlmFixedHeader{Code: -300, Msg: "invalid params"}, + }) + return + } + + changed := 0 + zlmCfg := s.conf.HttpNotifyConfig.ZlmCompatHookConfig + + hookMap := map[string]*string{ + "hook.on_stream_changed": &zlmCfg.ZlmOnStreamChanged, + "hook.on_server_keepalive": &zlmCfg.ZlmOnServerKeepalive, + "hook.on_stream_none_reader": &zlmCfg.ZlmOnStreamNoneReader, + "hook.on_rtp_server_timeout": &zlmCfg.ZlmOnRtpServerTimeout, + "hook.on_record_mp4": &zlmCfg.ZlmOnRecordMp4, + "hook.on_publish": &zlmCfg.ZlmOnPublish, + "hook.on_play": &zlmCfg.ZlmOnPlay, + "hook.on_stream_not_found": &zlmCfg.ZlmOnStreamNotFound, + "hook.on_server_started": &zlmCfg.ZlmOnServerStarted, + } + + for key, target := range hookMap { + if v, ok := params[key]; ok && v != nil && *v != *target { + *target = *v + changed++ + } + } + + if changed > 0 { + s.notifyHub.UpdateZlmHookConfig(zlmCfg) + s.conf.HttpNotifyConfig.ZlmCompatHookConfig = zlmCfg + } + + // 处理 keepalive 间隔 + if v, ok := params["hook.alive_interval"]; ok && v != nil { + if interval, err := strconv.Atoi(*v); err == nil && interval > 0 { + s.conf.HttpNotifyConfig.KeepaliveIntervalSec = interval + changed++ + } + } + + Log.Infof("zlm compat setServerConfig. changed=%d", changed) + c.JSON(http.StatusOK, ZlmSetServerConfigResp{ + ZlmFixedHeader: ZlmFixedHeader{Code: 0}, + Changed: changed, + }) +} + +// ---------- restartServer ---------- + +func (s *LalMaxServer) zlmRestartServerHandler(c *gin.Context) { + // 为什么不重启：lalmax 不需要像 ZLM 那样通过重启来重绑端口 + Log.Infof("zlm compat restartServer (noop)") + c.JSON(http.StatusOK, ZlmFixedHeader{Code: 0, Msg: "ok"}) +} + +// ---------- addStreamProxy ---------- + +func (s *LalMaxServer) zlmAddStreamProxyHandler(c *gin.Context) { + var req ZlmAddStreamProxyReq + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusOK, ZlmAddStreamProxyResp{ZlmFixedHeader: ZlmFixedHeader{Code: -300, Msg: "invalid params"}}) + return + } + + streamName := req.Stream + if streamName == "" { + c.JSON(http.StatusOK, ZlmAddStreamProxyResp{ZlmFixedHeader: ZlmFixedHeader{Code: -300, Msg: "stream is required"}}) + return + } + + pullReq := base.ApiCtrlStartRelayPullReq{ + Url: req.URL, + StreamName: streamName, + PullTimeoutMs: int(req.TimeoutSec * 1000), + PullRetryNum: req.RetryCount, + AutoStopPullAfterNoOutMs: base.AutoStopPullAfterNoOutMsNever, + RtspMode: req.RTPType, + } + if pullReq.PullRetryNum == 0 { + pullReq.PullRetryNum = base.PullRetryNumNever + } + if pullReq.PullTimeoutMs == 0 { + pullReq.PullTimeoutMs = logic.DefaultApiCtrlStartRelayPullReqPullTimeoutMs + } + + resp := s.lalsvr.CtrlStartRelayPull(pullReq) + if resp.ErrorCode != base.ErrorCodeSucc { + c.JSON(http.StatusOK, ZlmAddStreamProxyResp{ZlmFixedHeader: ZlmFixedHeader{Code: -1, Msg: resp.Desp}}) + return + } + + Log.Infof("zlm compat addStreamProxy. stream=%s, session_id=%s", streamName, resp.Data.SessionId) + var out ZlmAddStreamProxyResp + out.Code = 0 + out.Data.Key = resp.Data.SessionId + c.JSON(http.StatusOK, out) +} + +// ---------- startRecord ---------- + +func (s *LalMaxServer) zlmStartRecordHandler(c *gin.Context) { + var req ZlmStartRecordReq + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusOK, ZlmStartRecordResp{ZlmFixedHeader: ZlmFixedHeader{Code: -300, Msg: "invalid params"}}) + return + } + + rtmpAddr := extractHostPort(s.conf, "rtmp") + if rtmpAddr == "" { + c.JSON(http.StatusOK, ZlmStartRecordResp{ZlmFixedHeader: ZlmFixedHeader{Code: -1, Msg: "rtmp not configured"}}) + return + } + + _, err := s.recorder.startRecord(rtmpAddr, req.App, req.Stream, req.Type, req.MaxSecond) + if err != nil { + Log.Errorf("zlm compat startRecord failed. stream=%s, err=%v", req.Stream, err) + c.JSON(http.StatusOK, ZlmStartRecordResp{ZlmFixedHeader: ZlmFixedHeader{Code: -1, Msg: err.Error()}, Result: false}) + return + } + + Log.Infof("zlm compat startRecord. app=%s, stream=%s, type=%d", req.App, req.Stream, req.Type) + c.JSON(http.StatusOK, ZlmStartRecordResp{ZlmFixedHeader: ZlmFixedHeader{Code: 0}, Result: true}) +} + +// ---------- stopRecord ---------- + +func (s *LalMaxServer) zlmStopRecordHandler(c *gin.Context) { + var req ZlmStopRecordReq + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusOK, ZlmStopRecordResp{ZlmFixedHeader: ZlmFixedHeader{Code: -300, Msg: "invalid params"}}) + return + } + + file, err := s.recorder.stopRecord(req.App, req.Stream, req.Type) + if err != nil { + Log.Infof("zlm compat stopRecord not recording. app=%s, stream=%s, err=%v", req.App, req.Stream, err) + c.JSON(http.StatusOK, ZlmStopRecordResp{ZlmFixedHeader: ZlmFixedHeader{Code: 0}, Result: false}) + return + } + + Log.Infof("zlm compat stopRecord. app=%s, stream=%s, file=%s", req.App, req.Stream, file) + c.JSON(http.StatusOK, ZlmStopRecordResp{ZlmFixedHeader: ZlmFixedHeader{Code: 0}, Result: true}) +} + +// ---------- getSnap ---------- + +func (s *LalMaxServer) zlmGetSnapHandler(c *gin.Context) { + var req ZlmGetSnapReq + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusOK, ZlmFixedHeader{Code: -300, Msg: "invalid params"}) + return + } + + if req.URL == "" { + c.JSON(http.StatusOK, ZlmFixedHeader{Code: -300, Msg: "url is required"}) + return + } + + data, err := getSnap(req.URL, req.TimeoutSec) + if err != nil { + Log.Errorf("zlm compat getSnap failed. url=%s, err=%v", req.URL, err) + c.JSON(http.StatusOK, ZlmFixedHeader{Code: -1, Msg: err.Error()}) + return + } + + Log.Infof("zlm compat getSnap. url=%s, size=%d", req.URL, len(data)) + c.Data(http.StatusOK, "image/jpeg", data) +} + +// extractHostPort 从 lal 原始配置中提取指定协议的 host:port +// 为什么有默认值：ZLM 模式下 gb28181 假设 RTMP 总在标准端口可用 +func extractHostPort(conf *config.Config, protocol string) string { + var raw lalRawPorts + if len(conf.LalRawContent) > 0 { + _ = json.Unmarshal(conf.LalRawContent, &raw) + } + switch protocol { + case "rtmp": + addr := raw.Rtmp.Addr + if addr == "" { + return "127.0.0.1:1935" + } + if addr[0] == ':' { + return "127.0.0.1" + addr + } + return addr + } + return "" +} diff --git a/server/server.go b/server/server.go index 3526223..37d2c42 100644 --- a/server/server.go +++ b/server/server.go @@ -39,6 +39,7 @@ type LalMaxServer struct { httpfmp4svr *httpfmp4.HttpFmp4Server hlssvr *hls.HlsServer rtpPubMgr *rtppub.Manager + recorder *ffmpegRecorder } func NewLalMaxServer(conf *config.Config) (*LalMaxServer, error) { @@ -58,8 +59,19 @@ func NewLalMaxServer(conf *config.Config) (*LalMaxServer, error) { stats: maxlogic.NewStatAggregator(maxlogic.GetGroupManagerInstance()), notifyHub: notifyHub, rtpPubMgr: rtppub.NewManager(lalsvr, conf.GB28181Config.MediaConfig), + recorder: newFfmpegRecorder(""), } + // 注入 sub 数量查询，用于 on_stream_none_reader 判断 + notifyHub.SetSubCountFn(func(streamName string) int { + for _, g := range lalsvr.StatAllGroup() { + if g.StreamName == streamName { + return len(g.StatSubs) + } + } + return 0 + }) + if conf.SrtConfig.Enable { maxsvr.srtsvr = srt.NewSrtServer(conf.SrtConfig.Addr, lalsvr, func(option *srt.SrtOption) { option.Latency = 300 @@ -133,6 +145,7 @@ func (s *LalMaxServer) Run() (err error) { } go s.runPeriodicUpdate(ctx) + go s.runPeriodicKeepalive(ctx) go func() { nazalog.Infof("lalmax http listen. addr=%s", s.conf.HttpConfig.ListenAddr) @@ -179,6 +192,30 @@ func (s *LalMaxServer) runPeriodicUpdate(ctx context.Context) { } } +// runPeriodicKeepalive ZLM 兼容：定时发送 on_server_keepalive +func (s *LalMaxServer) runPeriodicKeepalive(ctx context.Context) { + if s == nil || s.notifyHub == nil { + return + } + + intervalSec := s.conf.HttpNotifyConfig.KeepaliveIntervalSec + if intervalSec <= 0 { + return + } + + ticker := time.NewTicker(time.Duration(intervalSec) * time.Second) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + s.notifyHub.NotifyServerKeepalive() + } + } +} + func (s *LalMaxServer) HookHub() *HttpNotify { return s.notifyHub } diff --git a/server/zlm_compat_config.go b/server/zlm_compat_config.go new file mode 100644 index 0000000..d92cf93 --- /dev/null +++ b/server/zlm_compat_config.go @@ -0,0 +1,111 @@ +package server + +import ( + "encoding/json" + "fmt" + "net" + "strconv" + + config "github.com/q191201771/lalmax/config" +) + +// lalRawPorts 从 LalRawContent 中提取 lal 的端口配置 +type lalRawPorts struct { + Rtmp struct { + Addr string `json:"addr"` + SslAddr string `json:"rtmps_addr"` + } `json:"rtmp"` + Rtsp struct { + Addr string `json:"addr"` + SslAddr string `json:"rtsps_addr"` + } `json:"rtsp"` +} + +// buildZlmServerConfig 将 lalmax 配置转换为 ZLM getServerConfig 响应格式 +// 为什么：owl 的 ZLMDriver.Connect 依赖 data[0] 中的 http.port / rtmp.port 等字段来更新端口信息 +func buildZlmServerConfig(conf *config.Config) map[string]any { + cfg := make(map[string]any) + + cfg["general.mediaServerId"] = conf.ServerId + + // 从 lal raw config 中提取 rtmp/rtsp 端口 + var lalPorts lalRawPorts + if len(conf.LalRawContent) > 0 { + _ = json.Unmarshal(conf.LalRawContent, &lalPorts) + } + cfg["rtmp.port"] = extractPort(lalPorts.Rtmp.Addr) + cfg["rtmp.sslport"] = extractPort(lalPorts.Rtmp.SslAddr) + cfg["rtsp.port"] = extractPort(lalPorts.Rtsp.Addr) + cfg["rtsp.sslport"] = extractPort(lalPorts.Rtsp.SslAddr) + cfg["http.port"] = extractPort(conf.HttpConfig.ListenAddr) + cfg["http.sslport"] = extractPort(conf.HttpConfig.HttpsListenAddr) + + // rtp_proxy 端口从 gb28181 配置获取 + cfg["rtp_proxy.port"] = strconv.Itoa(int(conf.GB28181Config.MediaConfig.ListenPort)) + rtpBase := int(conf.GB28181Config.MediaConfig.ListenPort) + rtpMax := rtpBase + int(conf.GB28181Config.MediaConfig.MultiPortMaxIncrement) + if rtpBase > 0 && rtpMax > rtpBase { + cfg["rtp_proxy.port_range"] = fmt.Sprintf("%d-%d", rtpBase+1, rtpMax) + } else { + cfg["rtp_proxy.port_range"] = "30000-35000" + } + + // --- RTC 配置 --- + if conf.RtcConfig.Enable { + cfg["rtc.port"] = strconv.Itoa(conf.RtcConfig.ICEUDPMuxPort) + cfg["rtc.tcpPort"] = strconv.Itoa(conf.RtcConfig.ICETCPMuxPort) + } else { + cfg["rtc.port"] = "0" + cfg["rtc.tcpPort"] = "0" + } + + // --- Hook 配置 --- + cfg["hook.enable"] = boolStr(conf.HttpNotifyConfig.Enable) + cfg["hook.alive_interval"] = strconv.Itoa(conf.HttpNotifyConfig.KeepaliveIntervalSec) + cfg["hook.on_stream_changed"] = conf.HttpNotifyConfig.ZlmOnStreamChanged + cfg["hook.on_server_keepalive"] = conf.HttpNotifyConfig.ZlmOnServerKeepalive + cfg["hook.on_stream_none_reader"] = conf.HttpNotifyConfig.ZlmOnStreamNoneReader + cfg["hook.on_rtp_server_timeout"] = conf.HttpNotifyConfig.ZlmOnRtpServerTimeout + cfg["hook.on_record_mp4"] = conf.HttpNotifyConfig.ZlmOnRecordMp4 + cfg["hook.on_server_started"] = conf.HttpNotifyConfig.ZlmOnServerStarted + cfg["hook.on_publish"] = conf.HttpNotifyConfig.ZlmOnPublish + cfg["hook.on_play"] = conf.HttpNotifyConfig.ZlmOnPlay + cfg["hook.on_flow_report"] = "" + cfg["hook.on_http_access"] = "" + cfg["hook.on_rtsp_auth"] = "" + cfg["hook.on_rtsp_realm"] = "" + cfg["hook.on_shell_login"] = "" + cfg["hook.on_send_rtp_stopped"] = "" + cfg["hook.on_server_exited"] = "" + cfg["hook.on_stream_not_found"] = conf.HttpNotifyConfig.ZlmOnStreamNotFound + cfg["hook.on_record_ts"] = "" + cfg["hook.timeoutSec"] = "10" + cfg["hook.retry"] = "1" + cfg["hook.retry_delay"] = "3" + cfg["hook.stream_changed_schemas"] = "" + + // --- 默认值填充 --- + cfg["api.secret"] = "" + cfg["api.apiDebug"] = "1" + + return cfg +} + +// extractPort 从 ":1935" 或 "0.0.0.0:1935" 格式中提取端口号字符串 +func extractPort(addr string) string { + if addr == "" { + return "0" + } + _, portStr, err := net.SplitHostPort(addr) + if err != nil { + return "0" + } + return portStr +} + +func boolStr(v bool) string { + if v { + return "1" + } + return "0" +} diff --git a/server/zlm_compat_ffmpeg.go b/server/zlm_compat_ffmpeg.go new file mode 100644 index 0000000..275389d --- /dev/null +++ b/server/zlm_compat_ffmpeg.go @@ -0,0 +1,160 @@ +package server + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "sync" + "time" +) + +// ffmpegRecorder 管理 ffmpeg 录像进程 +// 为什么用 ffmpeg：lal 内核无按需录像 API，ffmpeg 可从 RTMP 拉流写 MP4，与 ZLM 行为一致 +type ffmpegRecorder struct { + mu sync.Mutex + sessions map[string]*recordSession + outputDir string +} + +type recordSession struct { + cmd *exec.Cmd + cancel context.CancelFunc + app string + stream string + file string + start time.Time +} + +func newFfmpegRecorder(outputDir string) *ffmpegRecorder { + if outputDir == "" { + outputDir = "./record" + } + return &ffmpegRecorder{ + sessions: make(map[string]*recordSession), + outputDir: outputDir, + } +} + +// recordKey 生成录像会话唯一标识 +func recordKey(app, stream string, typ int) string { + return fmt.Sprintf("%d/%s/%s", typ, app, stream) +} + +// startRecord 启动 ffmpeg 从 RTMP 拉流并录制为 MP4 +func (r *ffmpegRecorder) startRecord(rtmpAddr, app, stream string, typ int, maxSecond int) (string, error) { + key := recordKey(app, stream, typ) + + r.mu.Lock() + defer r.mu.Unlock() + + if _, ok := r.sessions[key]; ok { + return "", fmt.Errorf("already recording: %s", key) + } + + dir := filepath.Join(r.outputDir, app, stream) + if err := os.MkdirAll(dir, 0o755); err != nil { + return "", fmt.Errorf("create record dir: %w", err) + } + + filename := fmt.Sprintf("%s_%s.mp4", stream, time.Now().Format("20060102_150405")) + outPath := filepath.Join(dir, filename) + + srcURL := fmt.Sprintf("rtmp://%s/%s/%s", rtmpAddr, app, stream) + + ctx, cancel := context.WithCancel(context.Background()) + + args := []string{ + "-hide_banner", + "-loglevel", "warning", + "-i", srcURL, + "-c", "copy", + "-movflags", "+faststart", + } + if maxSecond > 0 { + args = append(args, "-t", fmt.Sprintf("%d", maxSecond)) + } + args = append(args, "-y", outPath) + + cmd := exec.CommandContext(ctx, "ffmpeg", args...) + cmd.Stdout = nil + cmd.Stderr = nil + + if err := cmd.Start(); err != nil { + cancel() + return "", fmt.Errorf("ffmpeg start: %w", err) + } + + sess := &recordSession{ + cmd: cmd, + cancel: cancel, + app: app, + stream: stream, + file: outPath, + start: time.Now(), + } + r.sessions[key] = sess + + go func() { + _ = cmd.Wait() + r.mu.Lock() + delete(r.sessions, key) + r.mu.Unlock() + Log.Infof("ffmpeg record finished. key=%s, file=%s", key, outPath) + }() + + Log.Infof("ffmpeg record started. key=%s, file=%s, src=%s", key, outPath, srcURL) + return outPath, nil +} + +// stopRecord 终止 ffmpeg 录像进程 +func (r *ffmpegRecorder) stopRecord(app, stream string, typ int) (string, error) { + key := recordKey(app, stream, typ) + + r.mu.Lock() + sess, ok := r.sessions[key] + if !ok { + r.mu.Unlock() + return "", fmt.Errorf("not recording: %s", key) + } + delete(r.sessions, key) + r.mu.Unlock() + + sess.cancel() + _ = sess.cmd.Wait() + Log.Infof("ffmpeg record stopped. key=%s, file=%s, duration=%s", key, sess.file, time.Since(sess.start)) + return sess.file, nil +} + +// getSnap 用 ffmpeg 从指定 URL 截取一帧 JPEG 图片 +func getSnap(srcURL string, timeoutSec int) ([]byte, error) { + if timeoutSec <= 0 { + timeoutSec = 10 + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutSec)*time.Second) + defer cancel() + + args := []string{ + "-hide_banner", + "-loglevel", "warning", + "-i", srcURL, + "-vframes", "1", + "-f", "image2", + "-vcodec", "mjpeg", + "pipe:1", + } + + cmd := exec.CommandContext(ctx, "ffmpeg", args...) + out, err := cmd.Output() + if err != nil { + return nil, fmt.Errorf("ffmpeg snap: %w", err) + } + + if len(out) == 0 { + return nil, fmt.Errorf("ffmpeg snap: empty output") + } + + return out, nil +} diff --git a/server/zlm_compat_test.go b/server/zlm_compat_test.go new file mode 100644 index 0000000..17f3075 --- /dev/null +++ b/server/zlm_compat_test.go @@ -0,0 +1,830 @@ +package server + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + config "github.com/q191201771/lalmax/config" + + "github.com/q191201771/lal/pkg/base" +) + +// =========================================================================== +// REST API 兼容测试 +// =========================================================================== + +func TestZlmCompatOpenRtpServer(t *testing.T) { + body := `{"port":0,"tcp_mode":0,"stream_id":"zlm_compat_rtp_test"}` + r := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/index/api/openRtpServer", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r, req) + + if r.Code != http.StatusOK { + t.Fatalf("unexpected status: %d, body: %s", r.Code, r.Body.String()) + } + + var resp ZlmOpenRtpServerResp + if err := json.NewDecoder(r.Body).Decode(&resp); err != nil { + t.Fatal(err) + } + if resp.Code != 0 { + t.Fatalf("expected code=0, got %d msg=%s", resp.Code, resp.Msg) + } + if resp.Port == 0 { + t.Fatal("expected non-zero port") + } + + // 清理：关闭刚开启的 RTP 服务 + t.Cleanup(func() { + closeBody := `{"stream_id":"zlm_compat_rtp_test"}` + cr := httptest.NewRecorder() + creq := httptest.NewRequest("POST", "/index/api/closeRtpServer", strings.NewReader(closeBody)) + creq.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(cr, creq) + }) +} + +func TestZlmCompatCloseRtpServer(t *testing.T) { + // 先开启 + openBody := `{"port":0,"tcp_mode":0,"stream_id":"zlm_close_rtp_test"}` + r := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/index/api/openRtpServer", strings.NewReader(openBody)) + req.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r, req) + + if r.Code != http.StatusOK { + t.Fatalf("open failed: %d %s", r.Code, r.Body.String()) + } + + // 再关闭 + closeBody := `{"stream_id":"zlm_close_rtp_test"}` + r = httptest.NewRecorder() + req = httptest.NewRequest("POST", "/index/api/closeRtpServer", strings.NewReader(closeBody)) + req.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r, req) + + if r.Code != http.StatusOK { + t.Fatalf("close failed: %d %s", r.Code, r.Body.String()) + } + + var resp ZlmCloseRtpServerResp + if err := json.NewDecoder(r.Body).Decode(&resp); err != nil { + t.Fatal(err) + } + if resp.Code != 0 { + t.Fatalf("expected code=0, got %d", resp.Code) + } + if resp.Hit != 1 { + t.Fatalf("expected hit=1, got %d", resp.Hit) + } +} + +func TestZlmCompatCloseRtpServerNotFound(t *testing.T) { + body := `{"stream_id":"nonexistent_stream_id"}` + r := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/index/api/closeRtpServer", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r, req) + + if r.Code != http.StatusOK { + t.Fatalf("unexpected status: %d", r.Code) + } + + var resp ZlmCloseRtpServerResp + if err := json.NewDecoder(r.Body).Decode(&resp); err != nil { + t.Fatal(err) + } + if resp.Hit != 0 { + t.Fatalf("expected hit=0 for nonexistent stream, got %d", resp.Hit) + } +} + +func TestZlmCompatCloseStreams(t *testing.T) { + streamName := uniqueTestName("zlm_close_stream") + _, err := max.lalsvr.AddCustomizePubSession(streamName) + if err != nil { + t.Fatal(err) + } + + body := `{"app":"","stream":"` + streamName + `"}` + r := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/index/api/close_streams", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r, req) + + if r.Code != http.StatusOK { + t.Fatalf("unexpected status: %d %s", r.Code, r.Body.String()) + } + + var resp ZlmCloseStreamsResp + if err := json.NewDecoder(r.Body).Decode(&resp); err != nil { + t.Fatal(err) + } + if resp.Code != 0 { + t.Fatalf("expected code=0, got %d", resp.Code) + } + if resp.CountHit == 0 { + t.Fatal("expected count_hit > 0") + } +} + +func TestZlmCompatGetServerConfig(t *testing.T) { + body := `{}` + r := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/index/api/getServerConfig", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r, req) + + if r.Code != http.StatusOK { + t.Fatalf("unexpected status: %d %s", r.Code, r.Body.String()) + } + + var resp ZlmGetServerConfigResp + if err := json.NewDecoder(r.Body).Decode(&resp); err != nil { + t.Fatal(err) + } + if resp.Code != 0 { + t.Fatalf("expected code=0, got %d", resp.Code) + } + if len(resp.Data) == 0 { + t.Fatal("expected non-empty data array") + } + + // 验证返回的配置包含 ZLM 标准字段 + cfg := resp.Data[0] + requiredKeys := []string{ + "http.port", + "rtmp.port", + "rtsp.port", + "rtp_proxy.port", + "general.mediaServerId", + "hook.on_stream_changed", + } + for _, key := range requiredKeys { + if _, ok := cfg[key]; !ok { + t.Errorf("missing required config key: %s", key) + } + } +} + +func TestZlmCompatSetServerConfig(t *testing.T) { + body := `{ + "hook.on_stream_changed":"http://127.0.0.1:15123/webhook/on_stream_changed", + "hook.on_server_keepalive":"http://127.0.0.1:15123/webhook/on_server_keepalive", + "hook.on_publish":"http://127.0.0.1:15123/webhook/on_publish", + "hook.on_play":"http://127.0.0.1:15123/webhook/on_play", + "hook.on_stream_not_found":"http://127.0.0.1:15123/webhook/on_stream_not_found", + "hook.on_stream_none_reader":"http://127.0.0.1:15123/webhook/on_stream_none_reader", + "hook.on_record_mp4":"http://127.0.0.1:15123/webhook/on_record_mp4", + "hook.on_server_started":"http://127.0.0.1:15123/webhook/on_server_started", + "hook.alive_interval":"10" + }` + r := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/index/api/setServerConfig", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r, req) + + if r.Code != http.StatusOK { + t.Fatalf("unexpected status: %d %s", r.Code, r.Body.String()) + } + + var resp ZlmSetServerConfigResp + if err := json.NewDecoder(r.Body).Decode(&resp); err != nil { + t.Fatal(err) + } + if resp.Code != 0 { + t.Fatalf("expected code=0, got %d", resp.Code) + } + if resp.Changed < 8 { + t.Fatalf("expected at least 8 changed, got %d", resp.Changed) + } + + // 验证 getServerConfig 返回更新后的值 + r2 := httptest.NewRecorder() + req2 := httptest.NewRequest("POST", "/index/api/getServerConfig", strings.NewReader(`{}`)) + req2.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r2, req2) + + var getResp ZlmGetServerConfigResp + json.NewDecoder(r2.Body).Decode(&getResp) + cfg := getResp.Data[0] + + if cfg["hook.on_stream_changed"] != "http://127.0.0.1:15123/webhook/on_stream_changed" { + t.Errorf("on_stream_changed not updated: %v", cfg["hook.on_stream_changed"]) + } + if cfg["hook.on_publish"] != "http://127.0.0.1:15123/webhook/on_publish" { + t.Errorf("on_publish not updated: %v", cfg["hook.on_publish"]) + } +} + +func TestZlmCompatAddStreamProxy(t *testing.T) { + body := `{ + "vhost":"__defaultVhost__", + "app":"live", + "stream":"proxy_test", + "url":"rtmp://127.0.0.1:19350/live/test", + "retry_count":0, + "rtp_type":0, + "timeout_sec":5 + }` + r := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/index/api/addStreamProxy", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r, req) + + if r.Code != http.StatusOK { + t.Fatalf("unexpected status: %d %s", r.Code, r.Body.String()) + } + + var resp ZlmAddStreamProxyResp + if err := json.NewDecoder(r.Body).Decode(&resp); err != nil { + t.Fatal(err) + } + // 拉流可能因目标不存在而失败，但响应格式必须正确 + // code=0 表示成功，其他值表示拉流失败但格式正确 + if resp.Code == 0 && resp.Data.Key == "" { + t.Fatal("code=0 but key is empty") + } +} + +func TestZlmCompatStartStopRecord(t *testing.T) { + streamName := uniqueTestName("zlm_record") + _, err := max.lalsvr.AddCustomizePubSession(streamName) + if err != nil { + t.Fatal(err) + } + defer func() { + // 清理：尝试停止录制 + stopBody := `{"type":1,"vhost":"__defaultVhost__","app":"live","stream":"` + streamName + `"}` + sr := httptest.NewRecorder() + sreq := httptest.NewRequest("POST", "/index/api/stopRecord", strings.NewReader(stopBody)) + sreq.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(sr, sreq) + }() + + // 开始录制 + startBody := `{"type":1,"vhost":"__defaultVhost__","app":"live","stream":"` + streamName + `"}` + r := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/index/api/startRecord", strings.NewReader(startBody)) + req.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r, req) + + if r.Code != http.StatusOK { + t.Fatalf("start record unexpected status: %d %s", r.Code, r.Body.String()) + } + + var startResp ZlmStartRecordResp + if err := json.NewDecoder(r.Body).Decode(&startResp); err != nil { + t.Fatal(err) + } + if startResp.Code != 0 { + t.Fatalf("start record expected code=0, got %d msg=%s", startResp.Code, startResp.Msg) + } + if !startResp.Result { + t.Fatal("start record expected result=true") + } + + // 停止录制 + stopBody := `{"type":1,"vhost":"__defaultVhost__","app":"live","stream":"` + streamName + `"}` + r = httptest.NewRecorder() + req = httptest.NewRequest("POST", "/index/api/stopRecord", strings.NewReader(stopBody)) + req.Header.Set("Content-Type", "application/json") + max.router.ServeHTTP(r, req) + + if r.Code != http.StatusOK { + t.Fatalf("stop record unexpected status: %d %s", r.Code, r.Body.String()) + } + + var stopResp ZlmStopRecordResp + if err := json.NewDecoder(r.Body).Decode(&stopResp); err != nil { + t.Fatal(err) + } + if stopResp.Code != 0 { + t.Fatalf("stop record expected code=0, got %d msg=%s", stopResp.Code, stopResp.Msg) + } +} + +// =========================================================================== +// Hook 兼容测试 +// =========================================================================== + +// TestZlmHookOnStreamChangedFormat 验证 on_stream_changed hook 的 payload 格式与 ZLM 兼容 +func TestZlmHookOnStreamChangedFormat(t *testing.T) { + received := make(chan ZlmOnStreamChangedPayload, 2) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + var payload ZlmOnStreamChangedPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Errorf("decode on_stream_changed payload failed: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + received <- payload + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + hub := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + ZlmCompatHookConfig: config.ZlmCompatHookConfig{ZlmOnStreamChanged: ts.URL}, + }, "zlm-hook-test") + + streamName := uniqueTestName("stream_changed_test") + + // 模拟推流开始 -> 应触发 on_stream_changed(regist=true) + hub.NotifyPubStart(base.PubStartInfo{ + SessionEventCommonInfo: base.SessionEventCommonInfo{ + SessionId: "pub-session-1", + AppName: "live", + StreamName: streamName, + }, + }) + + select { + case payload := <-received: + if !payload.Regist { + t.Fatal("expected regist=true on pub_start") + } + // gb28181 优先读 app_name/stream_name（lalmax 兼容字段） + if payload.StreamName == "" && payload.Stream == "" { + t.Fatal("expected stream or stream_name to be set") + } + if payload.AppName == "" && payload.App == "" { + t.Fatal("expected app or app_name to be set") + } + case <-time.After(2 * time.Second): + t.Fatal("did not receive on_stream_changed for pub_start") + } + + // 模拟推流结束 -> 应触发 on_stream_changed(regist=false) + hub.NotifyPubStop(base.PubStopInfo{ + SessionEventCommonInfo: base.SessionEventCommonInfo{ + SessionId: "pub-session-1", + AppName: "live", + StreamName: streamName, + }, + }) + + select { + case payload := <-received: + if payload.Regist { + t.Fatal("expected regist=false on pub_stop") + } + case <-time.After(2 * time.Second): + t.Fatal("did not receive on_stream_changed for pub_stop") + } +} + +// TestZlmHookOnStreamChangedFieldCompleteness 验证 payload 包含 ZLM 必需字段 +func TestZlmHookOnStreamChangedFieldCompleteness(t *testing.T) { + received := make(chan json.RawMessage, 1) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + var raw json.RawMessage + if err := json.NewDecoder(r.Body).Decode(&raw); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + received <- raw + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + hub := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + ZlmCompatHookConfig: config.ZlmCompatHookConfig{ZlmOnStreamChanged: ts.URL}, + }, "field-test") + + hub.NotifyPubStart(base.PubStartInfo{ + SessionEventCommonInfo: base.SessionEventCommonInfo{ + SessionId: "completeness-sess", + AppName: "live", + StreamName: "completeness-stream", + }, + }) + + select { + case raw := <-received: + var m map[string]any + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + + // ZLM on_stream_changed 必须包含的字段 + requiredFields := []string{ + "regist", + "schema", + "mediaServerId", + "vhost", + } + for _, field := range requiredFields { + if _, ok := m[field]; !ok { + t.Errorf("missing required field in on_stream_changed: %s", field) + } + } + + // 必须有 app+stream 或 app_name+stream_name + hasZlmStyle := m["app"] != nil && m["stream"] != nil + hasLalmaxStyle := m["app_name"] != nil && m["stream_name"] != nil + if !hasZlmStyle && !hasLalmaxStyle { + t.Error("payload must contain (app, stream) or (app_name, stream_name)") + } + case <-time.After(2 * time.Second): + t.Fatal("did not receive on_stream_changed") + } +} + +// TestZlmHookOnServerKeepalive 验证 keepalive hook 的触发和 payload 格式 +func TestZlmHookOnServerKeepalive(t *testing.T) { + received := make(chan ZlmOnServerKeepalivePayload, 1) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + var payload ZlmOnServerKeepalivePayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Errorf("decode keepalive payload failed: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + received <- payload + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + hub := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + KeepaliveIntervalSec: 1, + ZlmCompatHookConfig: config.ZlmCompatHookConfig{ZlmOnServerKeepalive: ts.URL}, + }, "keepalive-test") + + // 手动触发 keepalive + hub.NotifyServerKeepalive() + + select { + case payload := <-received: + if payload.MediaServerID == "" { + t.Fatal("expected non-empty mediaServerId") + } + case <-time.After(2 * time.Second): + t.Fatal("did not receive on_server_keepalive") + } +} + +// TestZlmHookOnStreamNoneReader 验证无人观看 hook +func TestZlmHookOnStreamNoneReader(t *testing.T) { + received := make(chan ZlmOnStreamNoneReaderPayload, 1) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + var payload ZlmOnStreamNoneReaderPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Errorf("decode none_reader payload failed: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + received <- payload + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + hub := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + ZlmCompatHookConfig: config.ZlmCompatHookConfig{ZlmOnStreamNoneReader: ts.URL}, + }, "none-reader-test") + + hub.NotifyStreamNoneReader(ZlmOnStreamNoneReaderPayload{ + MediaServerID: "none-reader-test", + App: "live", + Schema: "rtmp", + Stream: "test-stream", + Vhost: "__defaultVhost__", + }) + + select { + case payload := <-received: + if payload.App != "live" { + t.Fatalf("expected app=live, got %s", payload.App) + } + if payload.Stream != "test-stream" { + t.Fatalf("expected stream=test-stream, got %s", payload.Stream) + } + case <-time.After(2 * time.Second): + t.Fatal("did not receive on_stream_none_reader") + } +} + +// TestZlmHookOnRtpServerTimeout 验证 RTP 超时 hook +func TestZlmHookOnRtpServerTimeout(t *testing.T) { + received := make(chan ZlmOnRtpServerTimeoutPayload, 1) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + var payload ZlmOnRtpServerTimeoutPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Errorf("decode rtp_timeout payload failed: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + received <- payload + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + hub := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + ZlmCompatHookConfig: config.ZlmCompatHookConfig{ZlmOnRtpServerTimeout: ts.URL}, + }, "rtp-timeout-test") + + hub.NotifyRtpServerTimeout(ZlmOnRtpServerTimeoutPayload{ + LocalPort: 30000, + StreamID: "timeout_stream", + TCPMode: 0, + MediaServerID: "rtp-timeout-test", + }) + + select { + case payload := <-received: + if payload.StreamID != "timeout_stream" { + t.Fatalf("expected stream_id=timeout_stream, got %s", payload.StreamID) + } + if payload.LocalPort != 30000 { + t.Fatalf("expected local_port=30000, got %d", payload.LocalPort) + } + case <-time.After(2 * time.Second): + t.Fatal("did not receive on_rtp_server_timeout") + } +} + +// TestZlmHookOnStreamChangedOrderPerStream 验证同一流的 stream_changed 事件保序 +func TestZlmHookOnStreamChangedOrderPerStream(t *testing.T) { + var order atomic.Int32 + firstDone := make(chan struct{}) + secondDone := make(chan struct{}) + allowFirst := make(chan struct{}) + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + var payload ZlmOnStreamChangedPayload + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + seq := order.Add(1) + if seq == 1 { + close(firstDone) + <-allowFirst + } else if seq == 2 { + close(secondDone) + } + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + hub := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + ZlmCompatHookConfig: config.ZlmCompatHookConfig{ZlmOnStreamChanged: ts.URL}, + }, "order-test") + + streamName := uniqueTestName("order_stream") + + hub.NotifyPubStart(base.PubStartInfo{ + SessionEventCommonInfo: base.SessionEventCommonInfo{ + SessionId: "order-1", + AppName: "live", + StreamName: streamName, + }, + }) + hub.NotifyPubStop(base.PubStopInfo{ + SessionEventCommonInfo: base.SessionEventCommonInfo{ + SessionId: "order-2", + AppName: "live", + StreamName: streamName, + }, + }) + + select { + case <-firstDone: + case <-time.After(time.Second): + t.Fatal("first on_stream_changed not received") + } + + // 第二个应被阻塞（同流保序） + select { + case <-secondDone: + t.Fatal("second on_stream_changed should be blocked") + case <-time.After(200 * time.Millisecond): + } + + close(allowFirst) + + select { + case <-secondDone: + case <-time.After(time.Second): + t.Fatal("second on_stream_changed not received after first finished") + } +} + +// ---------- on_publish ---------- + +func TestZlmHookOnPublish(t *testing.T) { + received := make(chan map[string]any, 1) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var m map[string]any + json.NewDecoder(r.Body).Decode(&m) + received <- m + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + hub := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + ZlmCompatHookConfig: config.ZlmCompatHookConfig{ZlmOnPublish: ts.URL}, + }, "pub-hook-test") + + hub.NotifyPubStart(base.PubStartInfo{ + SessionEventCommonInfo: base.SessionEventCommonInfo{ + AppName: "live", + StreamName: "test_pub", + Protocol: "rtmp", + }, + }) + + select { + case m := <-received: + if m["app"] != "live" || m["stream"] != "test_pub" || m["schema"] != "rtmp" { + t.Fatalf("unexpected on_publish payload: %+v", m) + } + if m["mediaServerId"] != "pub-hook-test" { + t.Fatalf("unexpected mediaServerId: %v", m["mediaServerId"]) + } + case <-time.After(time.Second): + t.Fatal("on_publish not received") + } +} + +// ---------- on_play ---------- + +func TestZlmHookOnPlay(t *testing.T) { + received := make(chan map[string]any, 1) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var m map[string]any + json.NewDecoder(r.Body).Decode(&m) + received <- m + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + hub := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + ZlmCompatHookConfig: config.ZlmCompatHookConfig{ZlmOnPlay: ts.URL}, + }, "play-hook-test") + + hub.NotifySubStart(base.SubStartInfo{ + SessionEventCommonInfo: base.SessionEventCommonInfo{ + AppName: "live", + StreamName: "test_play", + Protocol: "rtsp", + }, + }) + + select { + case m := <-received: + if m["app"] != "live" || m["stream"] != "test_play" || m["schema"] != "rtsp" { + t.Fatalf("unexpected on_play payload: %+v", m) + } + if m["mediaServerId"] != "play-hook-test" { + t.Fatalf("unexpected mediaServerId: %v", m["mediaServerId"]) + } + case <-time.After(time.Second): + t.Fatal("on_play not received") + } +} + +// ---------- on_stream_not_found ---------- + +func TestZlmHookOnStreamNotFound(t *testing.T) { + received := make(chan map[string]any, 1) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var m map[string]any + json.NewDecoder(r.Body).Decode(&m) + received <- m + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + hub := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + ZlmCompatHookConfig: config.ZlmCompatHookConfig{ZlmOnStreamNotFound: ts.URL}, + }, "notfound-hook-test") + + hub.NotifyStreamNotFound(ZlmOnStreamNotFoundPayload{ + App: "live", + Stream: "missing_stream", + Schema: "rtmp", + Vhost: "__defaultVhost__", + }) + + select { + case m := <-received: + if m["app"] != "live" || m["stream"] != "missing_stream" { + t.Fatalf("unexpected on_stream_not_found payload: %+v", m) + } + if m["mediaServerId"] != "notfound-hook-test" { + t.Fatalf("unexpected mediaServerId: %v", m["mediaServerId"]) + } + case <-time.After(time.Second): + t.Fatal("on_stream_not_found not received") + } +} + +// ---------- 融合兼容逻辑 ---------- + +func TestZlmHookDispatchByConfig(t *testing.T) { + // 验证：配置了 ZLM hook URL → ZLM 回调触发； + // 未配置 ZLM hook URL → ZLM 回调不触发； + // lalmax 原有回调始终按 URL 配置分发 + + zlmReceived := make(chan string, 8) + tZlm := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var m map[string]any + json.NewDecoder(r.Body).Decode(&m) + if _, ok := m["regist"]; ok { + zlmReceived <- "on_stream_changed" + } else { + zlmReceived <- "on_publish" + } + w.WriteHeader(http.StatusOK) + })) + defer tZlm.Close() + + lalReceived := make(chan string, 8) + tLal := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + lalReceived <- "on_pub_start" + w.WriteHeader(http.StatusOK) + })) + defer tLal.Close() + + // 同时配置 ZLM + lalmax → 两者都应触发 + hub := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + OnPubStart: tLal.URL, + ZlmCompatHookConfig: config.ZlmCompatHookConfig{ZlmOnStreamChanged: tZlm.URL, ZlmOnPublish: tZlm.URL}, + }, "both-mode") + + hub.NotifyPubStart(base.PubStartInfo{ + SessionEventCommonInfo: base.SessionEventCommonInfo{ + AppName: "live", + StreamName: "both_test", + Protocol: "rtmp", + }, + }) + + // ZLM 回调应触发（on_publish + on_stream_changed） + for i := 0; i < 2; i++ { + select { + case evt := <-zlmReceived: + t.Logf("both mode zlm: %s", evt) + case <-time.After(time.Second): + t.Fatal("both mode: expected zlm callback") + } + } + + // lalmax 原有回调也应触发 + select { + case evt := <-lalReceived: + t.Logf("both mode lal: %s", evt) + case <-time.After(time.Second): + t.Fatal("both mode: expected lalmax callback") + } + + // 仅配置 lalmax，不配置 ZLM → ZLM 回调不应触发 + hubLal := NewHttpNotify(config.HttpNotifyConfig{ + Enable: true, + OnPubStart: tLal.URL, + }, "lal-only") + + hubLal.NotifyPubStart(base.PubStartInfo{ + SessionEventCommonInfo: base.SessionEventCommonInfo{ + AppName: "live", + StreamName: "lal_only_test", + Protocol: "rtmp", + }, + }) + + select { + case evt := <-lalReceived: + t.Logf("lal-only mode: %s", evt) + case <-time.After(time.Second): + t.Fatal("lal-only mode: expected lalmax callback") + } + + // ZLM 回调不应触发 + select { + case <-zlmReceived: + t.Fatal("lal-only mode: should NOT receive zlm callback") + case <-time.After(200 * time.Millisecond): + } +} diff --git a/server/zlm_compat_types.go b/server/zlm_compat_types.go new file mode 100644 index 0000000..3be79a9 --- /dev/null +++ b/server/zlm_compat_types.go @@ -0,0 +1,253 @@ +package server + +// ZLM 兼容层请求/响应类型定义 +// 为什么放在 server 包：ZLM 兼容路由与现有 lalmax 路由同级，需访问 LalMaxServer 内部成员 + +// ZlmFixedHeader ZLM 标准响应头 +type ZlmFixedHeader struct { + Code int `json:"code"` + Msg string `json:"msg,omitempty"` +} + +// --- /index/api/openRtpServer --- + +type ZlmOpenRtpServerReq struct { + Port int `json:"port"` + TCPMode int8 `json:"tcp_mode"` + StreamID string `json:"stream_id"` +} + +type ZlmOpenRtpServerResp struct { + Code int `json:"code"` + Msg string `json:"msg,omitempty"` + Port int `json:"port"` +} + +// --- /index/api/closeRtpServer --- + +type ZlmCloseRtpServerReq struct { + StreamID string `json:"stream_id"` +} + +type ZlmCloseRtpServerResp struct { + Code int `json:"code"` + Hit int `json:"hit"` +} + +// --- /index/api/close_streams --- + +type ZlmCloseStreamsReq struct { + Schema string `json:"schema,omitempty"` + Vhost string `json:"vhost,omitempty"` + App string `json:"app,omitempty"` + Stream string `json:"stream,omitempty"` + Force bool `json:"force,omitempty"` +} + +type ZlmCloseStreamsResp struct { + Code int `json:"code"` + CountHit int `json:"count_hit"` + CountClosed int `json:"count_closed"` +} + +// --- /index/api/getServerConfig --- + +type ZlmGetServerConfigResp struct { + Code int `json:"code"` + Data []map[string]any `json:"data"` +} + +// --- /index/api/setServerConfig --- + +type ZlmSetServerConfigResp struct { + ZlmFixedHeader + Changed int `json:"changed"` +} + +// --- /index/api/startRecord --- + +type ZlmStartRecordReq struct { + Type int `json:"type"` + Vhost string `json:"vhost"` + App string `json:"app"` + Stream string `json:"stream"` + CustomPath string `json:"customized_path,omitempty"` + MaxSecond int `json:"max_second,omitempty"` +} + +type ZlmStartRecordResp struct { + ZlmFixedHeader + Result bool `json:"result"` +} + +// --- /index/api/stopRecord --- + +type ZlmStopRecordReq struct { + Type int `json:"type"` + Vhost string `json:"vhost"` + App string `json:"app"` + Stream string `json:"stream"` +} + +type ZlmStopRecordResp struct { + ZlmFixedHeader + Result bool `json:"result"` +} + +// --- /index/api/addStreamProxy --- + +type ZlmAddStreamProxyReq struct { + Vhost string `json:"vhost"` + App string `json:"app"` + Stream string `json:"stream"` + URL string `json:"url"` + RetryCount int `json:"retry_count"` + RTPType int `json:"rtp_type"` + TimeoutSec float32 `json:"timeout_sec"` +} + +type ZlmAddStreamProxyResp struct { + ZlmFixedHeader + Data struct { + Key string `json:"key"` + } `json:"data"` +} + +// --- /index/api/getSnap --- + +type ZlmGetSnapReq struct { + URL string `json:"url"` + TimeoutSec int `json:"timeout_sec"` + ExpireSec int `json:"expire_sec"` +} + +// --- on_stream_changed Hook Payload --- + +type ZlmOnStreamChangedPayload struct { + Regist bool `json:"regist"` + AliveSecond int `json:"aliveSecond"` + App string `json:"app"` + BytesSpeed int `json:"bytesSpeed"` + CreateStamp int64 `json:"createStamp"` + MediaServerID string `json:"mediaServerId"` + OriginSock ZlmOriginSock `json:"originSock"` + OriginType int `json:"originType"` + OriginTypeStr string `json:"originTypeStr"` + OriginURL string `json:"originUrl"` + ReaderCount int `json:"readerCount"` + Schema string `json:"schema"` + Stream string `json:"stream"` + TotalReaderCount int `json:"totalReaderCount"` + Tracks []ZlmTrack `json:"tracks"` + Vhost string `json:"vhost"` + AppName string `json:"app_name,omitempty"` + StreamName string `json:"stream_name,omitempty"` +} + +type ZlmOriginSock struct { + Identifier string `json:"identifier"` + LocalIP string `json:"local_ip"` + LocalPort int `json:"local_port"` + PeerIP string `json:"peer_ip"` + PeerPort int `json:"peer_port"` +} + +type ZlmTrack struct { + Channels int `json:"channels,omitempty"` + CodecID int `json:"codec_id"` + CodecIDName string `json:"codec_id_name"` + CodecType int `json:"codec_type"` + Ready bool `json:"ready"` + SampleBit int `json:"sample_bit,omitempty"` + SampleRate int `json:"sample_rate,omitempty"` + Fps float32 `json:"fps,omitempty"` + Height int `json:"height,omitempty"` + Width int `json:"width,omitempty"` +} + +// --- on_server_keepalive Hook Payload --- + +type ZlmOnServerKeepalivePayload struct { + MediaServerID string `json:"mediaServerId"` +} + +// --- on_stream_none_reader Hook Payload --- + +type ZlmOnStreamNoneReaderPayload struct { + MediaServerID string `json:"mediaServerId"` + App string `json:"app"` + Schema string `json:"schema"` + Stream string `json:"stream"` + Vhost string `json:"vhost"` +} + +// --- on_record_mp4 Hook Payload --- + +type ZlmOnRecordMp4Payload struct { + MediaServerID string `json:"mediaServerId"` + App string `json:"app"` + FileName string `json:"file_name"` + FilePath string `json:"file_path"` + FileSize int64 `json:"file_size"` + Folder string `json:"folder"` + StartTime int64 `json:"start_time"` + Stream string `json:"stream"` + TimeLen float64 `json:"time_len"` + URL string `json:"url"` + Vhost string `json:"vhost"` +} + +// --- on_publish Hook Payload --- + +type ZlmOnPublishPayload struct { + MediaServerID string `json:"mediaServerId"` + App string `json:"app"` + ID string `json:"id"` + IP string `json:"ip"` + Params string `json:"params"` + Port int `json:"port"` + Schema string `json:"schema"` + Stream string `json:"stream"` + Vhost string `json:"vhost"` +} + +// --- on_play Hook Payload --- + +type ZlmOnPlayPayload struct { + MediaServerID string `json:"mediaServerId"` + App string `json:"app"` + ID string `json:"id"` + IP string `json:"ip"` + Params string `json:"params"` + Port int `json:"port"` + Schema string `json:"schema"` + Stream string `json:"stream"` + Vhost string `json:"vhost"` +} + +// --- on_stream_not_found Hook Payload --- + +type ZlmOnStreamNotFoundPayload struct { + MediaServerID string `json:"mediaServerId"` + App string `json:"app"` + ID string `json:"id"` + IP string `json:"ip"` + Params string `json:"params"` + Port int `json:"port"` + Schema string `json:"schema"` + Stream string `json:"stream"` + Vhost string `json:"vhost"` + AppName string `json:"app_name,omitempty"` + StreamName string `json:"stream_name,omitempty"` +} + +// --- on_rtp_server_timeout Hook Payload --- + +type ZlmOnRtpServerTimeoutPayload struct { + LocalPort int `json:"local_port"` + ReUsePort bool `json:"re_use_port"` + SSRC uint32 `json:"ssrc"` + StreamID string `json:"stream_id"` + TCPMode int `json:"tcp_mode"` + MediaServerID string `json:"mediaServerId"` +} From 72dd671bd5569384e23b4be0a5497a49a01bbb22 Mon Sep 17 00:00:00 2001 From: xugo Date: Tue, 28 Apr 2026 19:53:15 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E6=94=AF=E6=8C=81=20webrtc,=20flv=20?= =?UTF-8?q?=E5=8D=8F=E8=AE=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/config.go | 36 ++++++++++++++ gb28181/rtppub/manager.go | 10 ++++ logic/group_manager.go | 16 ++++++ main.go | 1 + rtc/jessibucasession.go | 1 - rtc/server.go | 76 ++++++++++++++++++++++++++-- server/http_notify.go | 30 ++++++++++-- server/router.go | 2 + server/router_flv_proxy.go | 98 +++++++++++++++++++++++++++++++++++++ server/router_zlm_compat.go | 89 +++++++++++++++++++++++++++++++-- server/server.go | 9 ++++ server/zlm_compat_config.go | 22 ++++++++- 12 files changed, 376 insertions(+), 14 deletions(-) create mode 100644 server/router_flv_proxy.go diff --git a/config/config.go b/config/config.go index a97bf03..b3c78a9 100644 --- a/config/config.go +++ b/config/config.go @@ -2,7 +2,9 @@ package config import ( "encoding/json" + "fmt" "io/ioutil" + "os" ) var defaultConfig Config @@ -18,6 +20,7 @@ type Config struct { LalSvrConfigPath string `json:"lal_config_path"` // lal配置文件路径，兼容旧版配置 LogicConfig LogicConfig `json:"logic_config"` // 扩展流组配置 LalRawContent []byte `json:"-"` // lal 原始配置内容 + ConfFilePath string `json:"-"` // 配置文件路径，用于持久化 } type SrtConfig struct { @@ -116,6 +119,7 @@ type HttpNotifyConfig struct { Enable bool `json:"enable"` UpdateIntervalSec int `json:"update_interval_sec"` KeepaliveIntervalSec int `json:"keepalive_interval_sec"` + HookTimeoutSec int `json:"hook_timeout_sec"` OnServerStart string `json:"on_server_start"` OnUpdate string `json:"on_update"` OnGroupStart string `json:"on_group_start"` @@ -220,3 +224,35 @@ func unmarshalConfig(data []byte, cfg *Config) error { func GetConfig() *Config { return &defaultConfig } + +// SaveToFile 将当前配置持久化到配置文件 +// 为什么：setServerConfig 动态修改后需落盘，重启后配置仍生效 +func (c *Config) SaveToFile() error { + if c.ConfFilePath == "" { + return nil + } + + data, err := os.ReadFile(c.ConfFilePath) + if err != nil { + return fmt.Errorf("read config file: %w", err) + } + + var file map[string]json.RawMessage + if err := json.Unmarshal(data, &file); err != nil { + return fmt.Errorf("parse config file: %w", err) + } + + lalmax, err := json.MarshalIndent(c, "", " ") + if err != nil { + return fmt.Errorf("marshal lalmax config: %w", err) + } + file["lalmax"] = lalmax + + out, err := json.MarshalIndent(file, "", " ") + if err != nil { + return fmt.Errorf("marshal config file: %w", err) + } + out = append(out, '\n') + + return os.WriteFile(c.ConfFilePath, out, 0o644) +} diff --git a/gb28181/rtppub/manager.go b/gb28181/rtppub/manager.go index dcbdc4a..39b41d4 100644 --- a/gb28181/rtppub/manager.go +++ b/gb28181/rtppub/manager.go @@ -193,6 +193,16 @@ func (m *Manager) CheckSsrc(ssrc uint32) (*mediaserver.MediaInfo, bool) { func (m *Manager) NotifyClose(streamName string) { } +// UpdatePortRange 动态更新端口范围，由 setServerConfig 接口调用 +// 为什么：owl 通过 setServerConfig 下发 rtp_proxy.port_range，需运行时生效 +func (m *Manager) UpdatePortRange(portMin, portMax int) { + m.mu.Lock() + defer m.mu.Unlock() + m.portMin = portMin + m.portMax = portMax + nazalog.Infof("rtp pub port range updated. min=%d, max=%d", portMin, portMax) +} + func (m *Manager) OnRtpPacket(streamName string, mediaKey string) { m.mu.Lock() defer m.mu.Unlock() diff --git a/logic/group_manager.go b/logic/group_manager.go index 2b03988..066748d 100644 --- a/logic/group_manager.go +++ b/logic/group_manager.go @@ -2,6 +2,7 @@ package logic import ( "sync" + "time" "github.com/q191201771/lalmax/fmp4/hls" "github.com/q191201771/naza/pkg/nazalog" @@ -211,6 +212,21 @@ func (m *ComplexGroupManager) GetGroupByStreamName(streamName string) (bool, *Gr return m.GetGroup(StreamKeyFromStreamName(streamName)) } +// WaitGroup 等待流就绪，轮询 interval 间隔，总超时 timeout +// 为什么：GB28181 设备推流有延迟，播放端先于推流端到达，需短暂等待 +func (m *ComplexGroupManager) WaitGroup(key StreamKey, interval, timeout time.Duration) (bool, *Group) { + deadline := time.Now().Add(timeout) + for { + if ok, g := m.GetGroup(key); ok { + return true, g + } + if time.Now().After(deadline) { + return false, nil + } + time.Sleep(interval) + } +} + // streamName 单独查找只在匹配唯一 appName 时成功，避免跨 app 串流。 func (m *ComplexGroupManager) getGroupByOnlyStreamNameLocked(streamName string) (bool, *Group) { var found *Group diff --git a/main.go b/main.go index eb86959..86326ce 100644 --- a/main.go +++ b/main.go @@ -28,6 +28,7 @@ func main() { } maxConf := config.GetConfig() + maxConf.ConfFilePath = confFilename svr, err := server.NewLalMaxServer(maxConf) if err != nil { diff --git a/rtc/jessibucasession.go b/rtc/jessibucasession.go index 0410562..d487a65 100644 --- a/rtc/jessibucasession.go +++ b/rtc/jessibucasession.go @@ -5,7 +5,6 @@ import ( "math" "sync" "sync/atomic" - "github.com/gofrs/uuid" "github.com/pion/webrtc/v3" "github.com/q191201771/lal/pkg/base" diff --git a/rtc/server.go b/rtc/server.go index 5c91dfc..a18fe5d 100644 --- a/rtc/server.go +++ b/rtc/server.go @@ -4,8 +4,10 @@ import ( "fmt" "net" "net/http" + "time" config "github.com/q191201771/lalmax/config" + maxlogic "github.com/q191201771/lalmax/logic" "github.com/gin-gonic/gin" "github.com/pion/ice/v2" @@ -14,11 +16,37 @@ import ( "github.com/q191201771/naza/pkg/nazalog" ) +// StreamNotFoundFn 流不存在时的回调，触发 on_stream_not_found 通知上层拉流 +type StreamNotFoundFn func(app, stream, schema string) + type RtcServer struct { - config config.RtcConfig - lalServer logic.ILalServer - udpMux ice.UDPMux - tcpMux ice.TCPMux + config config.RtcConfig + lalServer logic.ILalServer + udpMux ice.UDPMux + tcpMux ice.TCPMux + streamNotFoundFn StreamNotFoundFn +} + +// SetStreamNotFoundFn 注入流不存在回调 +func (s *RtcServer) SetStreamNotFoundFn(fn StreamNotFoundFn) { + s.streamNotFoundFn = fn +} + +// waitStreamReady 触发 on_stream_not_found 后轮询等待流就绪 +// 为什么：WebRTC 播放请求先于 GB28181 设备推流到达，需通知上层拉流后等待 +func (s *RtcServer) waitStreamReady(appName, streamid, schema string) bool { + key := maxlogic.NewStreamKey(appName, streamid) + if ok, _ := maxlogic.GetGroupManagerInstance().GetGroup(key); ok { + return true + } + + if s.streamNotFoundFn != nil { + nazalog.Infof("stream not found, triggering on_stream_not_found. app=%s, stream=%s", appName, streamid) + s.streamNotFoundFn(appName, streamid, schema) + } + + ok, _ := maxlogic.GetGroupManagerInstance().WaitGroup(key, 500*time.Millisecond, 5*time.Second) + return ok } func NewRtcServer(config config.RtcConfig, lal logic.ILalServer) (*RtcServer, error) { @@ -136,6 +164,12 @@ func (s *RtcServer) HandleJessibuca(c *gin.Context) { return } + if !s.waitStreamReady(appName, streamid, "rtsp") { + nazalog.Errorf("stream not ready after waiting. app=%s, stream=%s", appName, streamid) + c.Status(http.StatusNotFound) + return + } + pc, err := newPeerConnection(s.config.ICEHostNATToIPs, s.udpMux, s.tcpMux) if err != nil { c.Status(http.StatusInternalServerError) @@ -183,6 +217,12 @@ func (s *RtcServer) HandleWHEP(c *gin.Context) { return } + if !s.waitStreamReady(appName, streamid, "rtsp") { + nazalog.Errorf("stream not ready after waiting. app=%s, stream=%s", appName, streamid) + c.Status(http.StatusNotFound) + return + } + pc, err := newPeerConnection(s.config.ICEHostNATToIPs, s.udpMux, s.tcpMux) if err != nil { c.Status(http.StatusInternalServerError) @@ -209,3 +249,31 @@ func (s *RtcServer) HandleWHEP(c *gin.Context) { c.Data(http.StatusCreated, "application/sdp", []byte(sdp)) } + +// HandleZlmWebrtcPlay ZLM 兼容 WebRTC 播放，返回 SDP answer +// 为什么独立方法：ZLM 信令格式为 JSON {"code":0,"sdp":"..."}，与 WHEP 纯 SDP 不同 +func (s *RtcServer) HandleZlmWebrtcPlay(app, stream, offer string) (string, error) { + if !s.waitStreamReady(app, stream, "rtsp") { + return "", fmt.Errorf("stream not found: %s/%s", app, stream) + } + + pc, err := newPeerConnection(s.config.ICEHostNATToIPs, s.udpMux, s.tcpMux) + if err != nil { + return "", fmt.Errorf("create peer connection: %w", err) + } + + session := NewWhepSession(app, stream, s.config.WriteChanSize, pc, s.lalServer) + if session == nil { + pc.Close() + return "", fmt.Errorf("create session failed: %s/%s", app, stream) + } + + sdp := session.GetAnswerSDP(offer) + if sdp == "" { + session.Close() + return "", fmt.Errorf("generate answer sdp failed") + } + + go session.Run() + return sdp, nil +} diff --git a/server/http_notify.go b/server/http_notify.go index b00114d..37f425b 100644 --- a/server/http_notify.go +++ b/server/http_notify.go @@ -125,14 +125,38 @@ func (h *HttpNotify) SetSubCountFn(fn SubCountFn) { // UpdateZlmHookConfig 运行时更新 ZLM 兼容 hook 配置 // 为什么：gb28181 通过 setServerConfig 动态设置 hook URL，需要立即生效 +// 为什么清零原有字段：ZLM 回调与 lalmax 原有回调互斥，避免双重触发 func (h *HttpNotify) UpdateZlmHookConfig(zlmCfg config.ZlmCompatHookConfig) { h.cfg.ZlmCompatHookConfig = zlmCfg h.cfg.Enable = true - Log.Infof("zlm compat hook config updated. on_stream_changed=%s, on_server_keepalive=%s, on_publish=%s, on_play=%s", - zlmCfg.ZlmOnStreamChanged, zlmCfg.ZlmOnServerKeepalive, zlmCfg.ZlmOnPublish, zlmCfg.ZlmOnPlay) + + if h.cfg.HookTimeoutSec > 0 { + h.client.Timeout = time.Duration(h.cfg.HookTimeoutSec) * time.Second + } + + h.cfg.OnServerStart = "" + h.cfg.OnUpdate = "" + h.cfg.OnGroupStart = "" + h.cfg.OnGroupStop = "" + h.cfg.OnStreamActive = "" + h.cfg.OnPubStart = "" + h.cfg.OnPubStop = "" + h.cfg.OnSubStart = "" + h.cfg.OnSubStop = "" + h.cfg.OnRelayPullStart = "" + h.cfg.OnRelayPullStop = "" + h.cfg.OnRtmpConnect = "" + h.cfg.OnHlsMakeTs = "" + + Log.Infof("zlm compat hook config updated. timeout=%ds, on_stream_changed=%s, on_server_keepalive=%s, on_publish=%s, on_play=%s", + h.cfg.HookTimeoutSec, zlmCfg.ZlmOnStreamChanged, zlmCfg.ZlmOnServerKeepalive, zlmCfg.ZlmOnPublish, zlmCfg.ZlmOnPlay) } func NewHttpNotify(cfg config.HttpNotifyConfig, serverId string) *HttpNotify { + timeout := notifyTimeoutSec + if cfg.HookTimeoutSec > 0 { + timeout = cfg.HookTimeoutSec + } httpNotify := &HttpNotify{ cfg: cfg, serverId: serverId, @@ -142,7 +166,7 @@ func NewHttpNotify(cfg config.HttpNotifyConfig, serverId string) *HttpNotify { plugins: make(map[string]*hookPluginEntry), httpPosts: make(map[string]*hookHTTPPostWorker), client: &http.Client{ - Timeout: time.Duration(notifyTimeoutSec) * time.Second, + Timeout: time.Duration(timeout) * time.Second, }, } httpNotify.mustRegisterBuiltinHTTPPlugin() diff --git a/server/router.go b/server/router.go index db26a7c..1cdf4b6 100644 --- a/server/router.go +++ b/server/router.go @@ -16,4 +16,6 @@ func (s *LalMaxServer) InitRouter(router *gin.Engine) { s.initStatRouter(router, auth) s.initCtrlRouter(router, auth) s.initZlmCompatRouter(router, auth) + + s.initFlvProxy(router) } diff --git a/server/router_flv_proxy.go b/server/router_flv_proxy.go new file mode 100644 index 0000000..5d2236d --- /dev/null +++ b/server/router_flv_proxy.go @@ -0,0 +1,98 @@ +package server + +import ( + "encoding/json" + "io" + "net/http" + "strings" + + "github.com/gin-gonic/gin" + "github.com/q191201771/naza/pkg/nazalog" +) + +// initFlvProxy 注册 NoRoute 兜底，将 .flv 请求代理到 lal 的 httpflv 服务 +// 为什么：ZLM 的 FLV 拉流路径是 /{app}/{stream}.live.flv，lal 的 httpflv 在独立端口， +// lalmax 不直接提供 httpflv，通过反向代理让外部只需访问 lalmax 单一端口 +func (s *LalMaxServer) initFlvProxy(router *gin.Engine) { + router.NoRoute(func(c *gin.Context) { + path := c.Request.URL.Path + if !strings.HasSuffix(path, ".flv") { + c.Status(http.StatusNotFound) + return + } + + lalHTTPAddr := s.getLalHttpflvAddr() + if lalHTTPAddr == "" { + c.Status(http.StatusBadGateway) + return + } + + targetURL := "http://" + lalHTTPAddr + path + if c.Request.URL.RawQuery != "" { + targetURL += "?" + c.Request.URL.RawQuery + } + + nazalog.Debugf("flv proxy. path=%s, target=%s", path, targetURL) + + req, err := http.NewRequestWithContext(c.Request.Context(), http.MethodGet, targetURL, nil) + if err != nil { + nazalog.Errorf("flv proxy create request failed. err=%v", err) + c.Status(http.StatusInternalServerError) + return + } + for k, vs := range c.Request.Header { + for _, v := range vs { + req.Header.Add(k, v) + } + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + nazalog.Errorf("flv proxy request failed. target=%s, err=%v", targetURL, err) + c.Status(http.StatusBadGateway) + return + } + defer resp.Body.Close() + + for k, vs := range resp.Header { + for _, v := range vs { + c.Header(k, v) + } + } + c.Status(resp.StatusCode) + + if resp.StatusCode != http.StatusOK { + return + } + + c.Header("Transfer-Encoding", "chunked") + c.Writer.Flush() + io.Copy(c.Writer, resp.Body) + }) +} + +// getLalHttpflvAddr 从 lal 原始配置中提取 httpflv 服务地址 +func (s *LalMaxServer) getLalHttpflvAddr() string { + if len(s.conf.LalRawContent) == 0 { + return "" + } + + var raw struct { + DefaultHTTP struct { + Addr string `json:"http_listen_addr"` + } `json:"default_http"` + } + + if err := json.Unmarshal(s.conf.LalRawContent, &raw); err != nil { + return "" + } + + addr := raw.DefaultHTTP.Addr + if addr == "" { + return "" + } + if addr[0] == ':' { + return "127.0.0.1" + addr + } + return addr +} diff --git a/server/router_zlm_compat.go b/server/router_zlm_compat.go index 67b7561..205b0e2 100644 --- a/server/router_zlm_compat.go +++ b/server/router_zlm_compat.go @@ -25,6 +25,7 @@ func (s *LalMaxServer) initZlmCompatRouter(router *gin.Engine, handlers ...gin.H zlm.POST("/stopRecord", s.zlmStopRecordHandler) zlm.POST("/addStreamProxy", s.zlmAddStreamProxyHandler) zlm.POST("/getSnap", s.zlmGetSnapHandler) + zlm.POST("/webrtc", s.zlmWebrtcHandler) } // ---------- openRtpServer ---------- @@ -164,11 +165,6 @@ func (s *LalMaxServer) zlmSetServerConfigHandler(c *gin.Context) { } } - if changed > 0 { - s.notifyHub.UpdateZlmHookConfig(zlmCfg) - s.conf.HttpNotifyConfig.ZlmCompatHookConfig = zlmCfg - } - // 处理 keepalive 间隔 if v, ok := params["hook.alive_interval"]; ok && v != nil { if interval, err := strconv.Atoi(*v); err == nil && interval > 0 { @@ -177,6 +173,47 @@ func (s *LalMaxServer) zlmSetServerConfigHandler(c *gin.Context) { } } + // 处理 hook 超时时间 + if v, ok := params["hook.timeoutSec"]; ok && v != nil { + if timeout, err := strconv.Atoi(*v); err == nil && timeout > 0 { + s.conf.HttpNotifyConfig.HookTimeoutSec = timeout + changed++ + } + } + + // 处理 rtp_proxy.port_range + if v, ok := params["rtp_proxy.port_range"]; ok && v != nil { + if portMin, portMax, ok := parsePortRange(*v); ok { + s.rtpPubMgr.UpdatePortRange(portMin, portMax) + changed++ + } + } + + if changed > 0 { + s.conf.HttpNotifyConfig.Enable = true + s.notifyHub.UpdateZlmHookConfig(zlmCfg) + s.conf.HttpNotifyConfig.ZlmCompatHookConfig = zlmCfg + + // 同步清零 conf 中的原有 hook URL + s.conf.HttpNotifyConfig.OnServerStart = "" + s.conf.HttpNotifyConfig.OnUpdate = "" + s.conf.HttpNotifyConfig.OnGroupStart = "" + s.conf.HttpNotifyConfig.OnGroupStop = "" + s.conf.HttpNotifyConfig.OnStreamActive = "" + s.conf.HttpNotifyConfig.OnPubStart = "" + s.conf.HttpNotifyConfig.OnPubStop = "" + s.conf.HttpNotifyConfig.OnSubStart = "" + s.conf.HttpNotifyConfig.OnSubStop = "" + s.conf.HttpNotifyConfig.OnRelayPullStart = "" + s.conf.HttpNotifyConfig.OnRelayPullStop = "" + s.conf.HttpNotifyConfig.OnRtmpConnect = "" + s.conf.HttpNotifyConfig.OnHlsMakeTs = "" + + if err := s.conf.SaveToFile(); err != nil { + Log.Errorf("zlm compat setServerConfig persist failed. err=%v", err) + } + } + Log.Infof("zlm compat setServerConfig. changed=%d", changed) c.JSON(http.StatusOK, ZlmSetServerConfigResp{ ZlmFixedHeader: ZlmFixedHeader{Code: 0}, @@ -306,6 +343,48 @@ func (s *LalMaxServer) zlmGetSnapHandler(c *gin.Context) { c.Data(http.StatusOK, "image/jpeg", data) } +// ---------- webrtc ---------- + +// zlmWebrtcHandler ZLM 兼容 WebRTC 信令接口 +// 为什么：gb28181 前端通过 /index/api/webrtc?app=xx&stream=xx&type=play 播放 +func (s *LalMaxServer) zlmWebrtcHandler(c *gin.Context) { + typ := c.Query("type") + app := c.Query("app") + stream := c.Query("stream") + + if stream == "" || typ != "play" { + c.JSON(http.StatusOK, gin.H{"code": -1, "msg": "only type=play supported"}) + return + } + + if s.rtcsvr == nil { + c.JSON(http.StatusOK, gin.H{"code": -1, "msg": "webrtc not enabled"}) + return + } + + body, err := c.GetRawData() + if err != nil || len(body) == 0 { + c.JSON(http.StatusOK, gin.H{"code": -1, "msg": "invalid sdp offer"}) + return + } + + Log.Infof("zlm compat webrtc play. app=%s, stream=%s", app, stream) + + sdp, err := s.rtcsvr.HandleZlmWebrtcPlay(app, stream, string(body)) + if err != nil { + Log.Errorf("zlm compat webrtc play failed. app=%s, stream=%s, err=%v", app, stream, err) + c.JSON(http.StatusOK, gin.H{"code": -1, "msg": err.Error()}) + return + } + + c.JSON(http.StatusOK, gin.H{ + "code": 0, + "id": s.conf.ServerId, + "sdp": sdp, + "type": "answer", + }) +} + // extractHostPort 从 lal 原始配置中提取指定协议的 host:port // 为什么有默认值：ZLM 模式下 gb28181 假设 RTMP 总在标准端口可用 func extractHostPort(conf *config.Config, protocol string) string { diff --git a/server/server.go b/server/server.go index 37d2c42..4a49d82 100644 --- a/server/server.go +++ b/server/server.go @@ -86,6 +86,15 @@ func NewLalMaxServer(conf *config.Config) (*LalMaxServer, error) { nazalog.Error("create rtc svr failed, err:", err) return nil, err } + maxsvr.rtcsvr.SetStreamNotFoundFn(func(app, stream, schema string) { + notifyHub.NotifyStreamNotFound(ZlmOnStreamNotFoundPayload{ + MediaServerID: conf.ServerId, + App: app, + Stream: stream, + Schema: schema, + Vhost: "__defaultVhost__", + }) + }) } if conf.Fmp4Config.Http.Enable { diff --git a/server/zlm_compat_config.go b/server/zlm_compat_config.go index d92cf93..b5206f9 100644 --- a/server/zlm_compat_config.go +++ b/server/zlm_compat_config.go @@ -5,6 +5,7 @@ import ( "fmt" "net" "strconv" + "strings" config "github.com/q191201771/lalmax/config" ) @@ -79,7 +80,11 @@ func buildZlmServerConfig(conf *config.Config) map[string]any { cfg["hook.on_server_exited"] = "" cfg["hook.on_stream_not_found"] = conf.HttpNotifyConfig.ZlmOnStreamNotFound cfg["hook.on_record_ts"] = "" - cfg["hook.timeoutSec"] = "10" + hookTimeout := conf.HttpNotifyConfig.HookTimeoutSec + if hookTimeout <= 0 { + hookTimeout = 10 + } + cfg["hook.timeoutSec"] = strconv.Itoa(hookTimeout) cfg["hook.retry"] = "1" cfg["hook.retry_delay"] = "3" cfg["hook.stream_changed_schemas"] = "" @@ -103,6 +108,21 @@ func extractPort(addr string) string { return portStr } +// parsePortRange 解析 "30000-35000" 格式的端口范围 +// 为什么：owl 通过 setServerConfig 下发端口范围字符串，需转换为 min/max int +func parsePortRange(s string) (int, int, bool) { + idx := strings.Index(s, "-") + if idx <= 0 || idx == len(s)-1 { + return 0, 0, false + } + minPort, err1 := strconv.Atoi(strings.TrimSpace(s[:idx])) + maxPort, err2 := strconv.Atoi(strings.TrimSpace(s[idx+1:])) + if err1 != nil || err2 != nil || minPort <= 0 || maxPort <= minPort { + return 0, 0, false + } + return minPort, maxPort, true +} + func boolStr(v bool) string { if v { return "1"