diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..09f7a31 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +.codex-cache/ diff --git a/README.md b/README.md index ff1c497..7ecb6fc 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,7 @@ docker run -it -p 1935:1935 -p 8080:8080 -p 4433:4433 -p 5544:5544 -p 8083:8083 # 架构 -![图片](image/init.png) +![图片](document/images/init.png) # 支持的协议 ## 推流 diff --git a/conf/config_test.go b/conf/config_test.go deleted file mode 100644 index 7de1c67..0000000 --- a/conf/config_test.go +++ /dev/null @@ -1,64 +0,0 @@ -package config - -import ( - "strings" - "testing" -) - -func TestUnmarshalStructuredConfig(t *testing.T) { - raw := []byte(`{ - "lalmax": { - "srt_config": { - "enable": true, - "addr": ":6001" - }, - "server_id": "lalmax-1" - }, - "lal": { - "rtmp": { - "enable": true, - "addr": ":1935" - } - } - }`) - - if err := Unmarshal(raw); err != nil { - t.Fatalf("unmarshal structured config: %v", err) - } - - cfg := GetConfig() - if !cfg.SrtConfig.Enable || cfg.SrtConfig.Addr != ":6001" { - t.Fatalf("unexpected srt config: %+v", cfg.SrtConfig) - } - if cfg.ServerId != "lalmax-1" { - t.Fatalf("unexpected server id: %s", cfg.ServerId) - } - if !strings.Contains(string(cfg.LalRawContent), `"rtmp"`) { - t.Fatalf("lal raw content not preserved: %s", string(cfg.LalRawContent)) - } -} - -func TestUnmarshalLegacyConfig(t *testing.T) { - raw := []byte(`{ - "srt_config": { - "enable": true, - "addr": ":6001" - }, - "lal_config_path:": "./conf/lalserver.conf.json" - }`) - - if err := Unmarshal(raw); err != nil { - t.Fatalf("unmarshal legacy config: %v", err) - } - - cfg := GetConfig() - if !cfg.SrtConfig.Enable || cfg.SrtConfig.Addr != ":6001" { - t.Fatalf("unexpected srt config: %+v", cfg.SrtConfig) - } - if cfg.LalSvrConfigPath != "./conf/lalserver.conf.json" { - t.Fatalf("unexpected lal config path: %s", cfg.LalSvrConfigPath) - } - if len(cfg.LalRawContent) != 0 { - t.Fatalf("legacy config should not set lal raw content: %s", string(cfg.LalRawContent)) - } -} diff --git a/conf/lalmax.conf.json b/conf/lalmax.conf.json index a348ea8..30f793e 100644 --- a/conf/lalmax.conf.json +++ b/conf/lalmax.conf.json @@ -21,10 +21,19 @@ "secrets": [] } }, - "httpfmp4_config": { - "enable": true + "fmp4_config": { + "http": { + "enable": true + }, + "hls": { + "enable": true, + "segment_count": 7, + "segment_duration": 1, + "part_duration": 200, + "low_latency": false + } }, - "hook_config": { + "logic_config": { "gop_cache_num": 1, "single_gop_max_frame_num": 0 }, diff --git a/conf/config.go b/config/config.go similarity index 84% rename from conf/config.go rename to config/config.go index 14809fd..805246a 100644 --- a/conf/config.go +++ b/config/config.go @@ -11,13 +11,12 @@ type Config struct { SrtConfig SrtConfig `json:"srt_config"` // srt配置 RtcConfig RtcConfig `json:"rtc_config"` // rtc配置 HttpConfig HttpConfig `json:"http_config"` // http/https配置 - HttpFmp4Config HttpFmp4Config `json:"httpfmp4_config"` // http-fmp4配置 - HlsConfig HlsConfig `json:"hls_config"` // hls-fmp4/llhls配置 + Fmp4Config Fmp4Config `json:"fmp4_config"` // fmp4配置 GB28181Config GB28181Config `json:"gb28181_config"` // gb28181配置 ServerId string `json:"server_id"` // http 通知唯一标识 HttpNotifyConfig HttpNotifyConfig `json:"http_notify"` // http 通知配置 LalSvrConfigPath string `json:"lal_config_path"` // lal配置文件路径，兼容旧版配置 - HookConfig HookConfig `json:"hook_config"` // gop cache配置 + LogicConfig LogicConfig `json:"logic_config"` // 扩展流组配置 LalRawContent []byte `json:"-"` // lal 原始配置内容 } @@ -43,17 +42,22 @@ type HttpConfig struct { CtrlAuthWhitelist CtrlAuthWhitelist `json:"ctrl_auth_whitelist"` } -// CtrlAuthWhitelist 控制类接口鉴权 +// CtrlAuthWhitelist 控制类接口鉴权。 type CtrlAuthWhitelist struct { IPs []string // 允许访问的远程 IP，零值时不生效 Secrets []string // 认证信息，零值时不生效 } -type HttpFmp4Config struct { +type Fmp4Config struct { + Http Fmp4HttpConfig `json:"http"` + Hls Fmp4HlsConfig `json:"hls"` +} + +type Fmp4HttpConfig struct { Enable bool `json:"enable"` // http-fmp4使能标志 } -type HlsConfig struct { +type Fmp4HlsConfig struct { Enable bool `json:"enable"` // hls使能标志 SegmentCount int `json:"segment_count"` // 分片个数,llhls默认7个 SegmentDuration int `json:"segment_duration"` // hls分片时长,默认1s @@ -78,8 +82,9 @@ type GB28181Config struct { type GB28181MediaConfig struct { MediaIp string `json:"media_ip"` // 流媒体IP,用于在SDP中指定 ListenPort uint16 `json:"listen_port"` // tcp,udp监听端口 默认启动 - MultiPortMaxIncrement uint16 `json:"multi_port_max_increment"` //多端口范围 ListenPort+1至ListenPort+MultiPortMax + MultiPortMaxIncrement uint16 `json:"multi_port_max_increment"` // 多端口范围 ListenPort+1至ListenPort+MultiPortMax } + type HttpNotifyConfig struct { Enable bool `json:"enable"` UpdateIntervalSec int `json:"update_interval_sec"` @@ -95,7 +100,7 @@ type HttpNotifyConfig struct { OnHlsMakeTs string `json:"on_hls_make_ts"` } -type HookConfig struct { +type LogicConfig struct { GopCacheNum int `json:"gop_cache_num"` SingleGopMaxFrameNum int `json:"single_gop_max_frame_num"` } @@ -142,6 +147,10 @@ func unmarshalConfig(data []byte, cfg *Config) error { if err := json.Unmarshal(data, cfg); err != nil { return err } + var raw map[string]json.RawMessage + if err := json.Unmarshal(data, &raw); err != nil { + return err + } if cfg.LalSvrConfigPath == "" { var legacy struct { LalSvrConfigPath string `json:"lal_config_path:"` @@ -151,6 +160,26 @@ func unmarshalConfig(data []byte, cfg *Config) error { } cfg.LalSvrConfigPath = legacy.LalSvrConfigPath } + if _, ok := raw["logic_config"]; !ok { + var legacy struct { + LogicConfig LogicConfig `json:"hook_config"` + } + if err := json.Unmarshal(data, &legacy); err != nil { + return err + } + cfg.LogicConfig = legacy.LogicConfig + } + if _, ok := raw["fmp4_config"]; !ok { + var legacy struct { + Http Fmp4HttpConfig `json:"httpfmp4_config"` + Hls Fmp4HlsConfig `json:"hls_config"` + } + if err := json.Unmarshal(data, &legacy); err != nil { + return err + } + cfg.Fmp4Config.Http = legacy.Http + cfg.Fmp4Config.Hls = legacy.Hls + } return nil } diff --git a/config/config_test.go b/config/config_test.go new file mode 100644 index 0000000..8ac73a5 --- /dev/null +++ b/config/config_test.go @@ -0,0 +1,152 @@ +package config + +import ( + "strings" + "testing" +) + +func TestUnmarshalStructuredConfig(t *testing.T) { + raw := []byte(`{ + "lalmax": { + "srt_config": { + "enable": true, + "addr": ":6001" + }, + "server_id": "lalmax-1" + }, + "lal": { + "rtmp": { + "enable": true, + "addr": ":1935" + } + } + }`) + + if err := Unmarshal(raw); err != nil { + t.Fatalf("unmarshal structured config: %v", err) + } + + cfg := GetConfig() + if !cfg.SrtConfig.Enable || cfg.SrtConfig.Addr != ":6001" { + t.Fatalf("unexpected srt config: %+v", cfg.SrtConfig) + } + if cfg.ServerId != "lalmax-1" { + t.Fatalf("unexpected server id: %s", cfg.ServerId) + } + if !strings.Contains(string(cfg.LalRawContent), `"rtmp"`) { + t.Fatalf("lal raw content not preserved: %s", string(cfg.LalRawContent)) + } +} + +func TestUnmarshalLegacyConfig(t *testing.T) { + raw := []byte(`{ + "srt_config": { + "enable": true, + "addr": ":6001" + }, + "httpfmp4_config": { + "enable": true + }, + "hls_config": { + "enable": true, + "segment_count": 3, + "segment_duration": 2, + "part_duration": 100, + "low_latency": true + }, + "hook_config": { + "gop_cache_num": 3, + "single_gop_max_frame_num": 120 + }, + "lal_config_path:": "./conf/lalserver.conf.json" + }`) + + if err := Unmarshal(raw); err != nil { + t.Fatalf("unmarshal legacy config: %v", err) + } + + cfg := GetConfig() + if !cfg.SrtConfig.Enable || cfg.SrtConfig.Addr != ":6001" { + t.Fatalf("unexpected srt config: %+v", cfg.SrtConfig) + } + if cfg.LalSvrConfigPath != "./conf/lalserver.conf.json" { + t.Fatalf("unexpected lal config path: %s", cfg.LalSvrConfigPath) + } + if cfg.LogicConfig.GopCacheNum != 3 || cfg.LogicConfig.SingleGopMaxFrameNum != 120 { + t.Fatalf("unexpected legacy logic config: %+v", cfg.LogicConfig) + } + if !cfg.Fmp4Config.Http.Enable { + t.Fatalf("unexpected legacy fmp4 http config: %+v", cfg.Fmp4Config.Http) + } + if !cfg.Fmp4Config.Hls.Enable || cfg.Fmp4Config.Hls.SegmentCount != 3 || cfg.Fmp4Config.Hls.SegmentDuration != 2 || cfg.Fmp4Config.Hls.PartDuration != 100 || !cfg.Fmp4Config.Hls.LowLatency { + t.Fatalf("unexpected legacy fmp4 hls config: %+v", cfg.Fmp4Config.Hls) + } + if len(cfg.LalRawContent) != 0 { + t.Fatalf("legacy config should not set lal raw content: %s", string(cfg.LalRawContent)) + } +} + +func TestUnmarshalStructuredFmp4ConfigKeepsExplicitZero(t *testing.T) { + raw := []byte(`{ + "lalmax": { + "fmp4_config": { + "http": { + "enable": false + }, + "hls": { + "enable": true, + "segment_count": 0, + "segment_duration": 0, + "part_duration": 0, + "low_latency": false + } + }, + "httpfmp4_config": { + "enable": true + }, + "hls_config": { + "enable": true, + "segment_count": 3, + "segment_duration": 2, + "part_duration": 100, + "low_latency": true + } + } + }`) + + if err := Unmarshal(raw); err != nil { + t.Fatalf("unmarshal structured config: %v", err) + } + + cfg := GetConfig() + if cfg.Fmp4Config.Http.Enable { + t.Fatalf("explicit fmp4 http config should not be overwritten: %+v", cfg.Fmp4Config.Http) + } + if !cfg.Fmp4Config.Hls.Enable || cfg.Fmp4Config.Hls.SegmentCount != 0 || cfg.Fmp4Config.Hls.SegmentDuration != 0 || cfg.Fmp4Config.Hls.PartDuration != 0 || cfg.Fmp4Config.Hls.LowLatency { + t.Fatalf("explicit fmp4 hls config should not be overwritten: %+v", cfg.Fmp4Config.Hls) + } +} + +func TestUnmarshalStructuredLogicConfigKeepsExplicitZero(t *testing.T) { + raw := []byte(`{ + "lalmax": { + "logic_config": { + "gop_cache_num": 0, + "single_gop_max_frame_num": 0 + }, + "hook_config": { + "gop_cache_num": 3, + "single_gop_max_frame_num": 120 + } + } + }`) + + if err := Unmarshal(raw); err != nil { + t.Fatalf("unmarshal structured config: %v", err) + } + + cfg := GetConfig() + if cfg.LogicConfig.GopCacheNum != 0 || cfg.LogicConfig.SingleGopMaxFrameNum != 0 { + t.Fatalf("explicit logic config should not be overwritten: %+v", cfg.LogicConfig) + } +} diff --git a/document/config.md b/document/config.md index 456a59a..594b156 100644 --- a/document/config.md +++ b/document/config.md @@ -87,48 +87,53 @@ *值举例*: ["192.168.1.2","192.168.1.3"] -# http-fmp4配置 -主要用于设置http-fmp4相关的配置,需要配合http_config一起使用 -- enable: http-fmp4服务使能配置 +# fmp4_config +主要用于设置 lalmax fMP4 相关能力，需要配合 `http_config` 一起使用。 + +## http +主要用于设置 HTTP-FMP4 相关的配置。 + +- enable: HTTP-FMP4 服务使能配置 *类型*: bool *值举例*: true -# hls_config -主要用于设置hls-fmp4/llhls相关的配置,需要配合http_config一起使用,hls-ts的能力请使用lal,这里不做过多描述 -- enable: hls-fmp4/llhls服务使能配置 +## hls +主要用于设置 HLS-FMP4/LLHLS 相关的配置。HLS-TS 能力请使用 lal 的 `hls` 配置。 + +- enable: HLS-FMP4/LLHLS 服务使能配置 *类型*: bool *值举例*: true -- segmentCount: hls-fmp4 m3u8返回的切片个数,默认为7, llhls默认设置为7个(gohlslib要求) +- segment_count: HLS-FMP4 m3u8 返回的切片个数，默认为 7。LLHLS 默认设置为 7 个。 *类型*: int *值举例*: 3 -- segmentDuration: hls-fmp4 切片时长,默认为1s +- segment_duration: HLS-FMP4 切片时长，默认为 1s *类型*: int *值举例*: 3 -- partDuration:llhls part部分的时长,默认为200ms +- part_duration: LLHLS part 部分的时长，默认为 200ms *类型*: int *值举例*: 100 -- lowLatency: llhls使能配置,开启此配置后则都走llhls +- low_latency: LLHLS 使能配置，开启此配置后使用低延迟 HLS *类型*: bool *值举例*: true -# hook_config -主要用于 hook 相关的配置。 +# logic_config +主要用于 lalmax 扩展流组相关的配置。 - gop_cache_num: gop 缓存的数量，默认为 1 diff --git a/document/gb28181.md b/document/gb28181.md index ec354d1..d00dfc5 100644 --- a/document/gb28181.md +++ b/document/gb28181.md @@ -287,4 +287,4 @@ Method: POST ``` # 海康设备接入 -![图片](../image/gb-hk.png) +![图片](images/gb-hk.png) diff --git a/document/hook_logic_management_analysis.md b/document/hook_logic_management_analysis.md new file mode 100644 index 0000000..0e16271 --- /dev/null +++ b/document/hook_logic_management_analysis.md @@ -0,0 +1,393 @@ +# Hook 与流管理改造分析 + +本文档分析两个问题： + +- `hook` 是否适合改成类似 lal `logic` 的管理方式。 +- lal 中所有流信息是否适合都由 lalmax 来管理。 + +结论先行： + +- `hook` 适合改造成“lalmax 扩展流管理层”，但不适合照搬 lal `logic.Group` 的完整职责。 +- lal 原生流生命周期、原生协议会话、remux、GOP 缓存、统计计算，仍应由 lal `logic` 作为事实源。 +- lalmax 更适合作为“扩展能力管理者 + 统一查询门面”，聚合 lal 原生状态和 lalmax 扩展订阅状态，而不是接管 lal 的全部流状态。 + +## 当前职责边界 + +### lal `logic` + +lal `logic` 当前承担流媒体内核职责： + +- 管理 `Group` 生命周期。 +- 管理输入流：RTMP pub、RTSP pub、GB28181/RTP PS pub、自定义 pub、relay pull。 +- 管理输出流：RTMP sub、RTSP sub、HTTP-FLV sub、HTTP-TS sub、HLS sub、relay push。 +- 管理协议转换链路：RTMP 到 RTSP、RTMP 到 MPEG-TS、RTSP/PS 到 RTMP 等。 +- 管理 GOP 缓存、HLS、录制、转推、回源。 +- 维护 `StatGroup`、`StatSession`、码率、在线状态、HTTP API 状态。 +- 处理输入流生命周期中的 hook 回调：`WithOnHookSession`。 + +lal 的 `Group` 是强状态对象，核心特征是： + +- 单流强一致生命周期。 +- 所有核心协议会话在同一个锁内增删。 +- 统计信息从真实 session 对象计算。 +- 输入流和输出流互相关联，任何重复管理都会产生状态一致性问题。 + +### lalmax `hook` + +lalmax 当前 `hook` 是 lal `WithOnHookSession` 的业务扩展层： + +- lal 输入流出现时创建 `HookSession`。 +- 接收 lal 分发出来的 RTMP 消息。 +- 自己维护一份轻量 GOP 缓存。 +- 给 WHEP、Jessibuca、HTTP-FMP4、HLS-FMP4/LLHLS 等 lalmax 扩展消费者分发数据。 +- 给新消费者回放缓存 GOP。 +- 在 lalmax HTTP API 里补充扩展消费者统计。 + +当前 `hook` 的特点： + +- 管理粒度只有 `streamName`，没有 `appName`。 +- `HookSessionMangaer` 是全局 singleton + `sync.Map`。 +- `consumerInfo` 的 `StatSession` 很轻，协议、远端地址、读写字节、码率等大多未完整填充。 +- 扩展消费者生命周期主要由各模块自行调用 `AddConsumer` / `RemoveConsumer`。 +- `hook` 不管理 lal 原生协议 session。 + +这说明 `hook` 当前不是流内核，而是 lalmax 扩展订阅层。 + +## 是否适合改成 lal `logic` 那样的管理 + +适合借鉴，不适合照搬。 + +### 适合借鉴的部分 + +`hook` 可以借鉴 lal `logic` 的管理模型，形成更清晰的 lalmax 扩展流管理层： + +- 抽象 `HookManager` 接口，避免全局 singleton 直接散落使用。 +- 抽象 `HookGroup` 或 `ExtGroup`，按流管理扩展消费者。 +- 支持 `GetOrCreateGroup`、`GetGroup`、`Iterate`、`Len` 等方法。 +- 支持 `appName + streamName` 的流标识，为未来复杂路径做准备。 +- 将扩展消费者按类型管理，例如 WHEP、Jessibuca、HTTP-FMP4、HLS-FMP4。 +- 统一扩展消费者的 `StatSession` 填充、码率更新、远端地址、协议名和关闭逻辑。 +- 把 `GetAllConsumer()` 做成可信统计，而不是临时拼接。 +- 增加统一的 `Tick` 或定时统计，避免扩展订阅没有码率。 +- 用明确的生命周期事件：`OnLalInputStart`、`OnRtmpMsg`、`OnLalInputStop`、`AddExtSub`、`DelExtSub`。 + +这些改造能解决当前 `hook` 的几个问题： + +- 统计不准。 +- 生命周期分散。 +- 扩展订阅者和 lal 原生订阅者没有统一视图。 +- 未来新增扩展协议时容易继续堆在 `HookSession` 里。 +- `streamName` 作为唯一 key 对多 appName 场景不友好。 + +### 不适合照搬的部分 + +不建议把 lal `logic.Group` 的职责完整搬到 `hook`： + +- lal `Group` 已经负责协议转发、remux、缓存、录制、推拉流生命周期。 +- hook 只能拿到 lal 分发后的 RTMP 消息，不拥有原生输入 session。 +- 如果 hook 也管理原生 pub/sub/pull，会和 lal 内部 `Group` 形成双事实源。 +- 双事实源会带来状态不一致，例如 lal 内部 session 已关闭但 lalmax 仍认为在线。 +- lal 升级后内部 session 行为变化，lalmax 复制逻辑很容易漂移。 +- hook 回调在 lal 内部处理链路上，同步阻塞会影响 lal 核心转发性能。 + +因此，合理目标不是“把 hook 改成另一个 lal logic”，而是“把 hook 改成 lalmax 扩展订阅的 logic”。 + +## lal 所有流信息是否应由 lalmax 管理 + +不建议。 + +### 不适合全部由 lalmax 管理的原因 + +lal 原生流信息应该继续由 lal 管理，原因是： + +- lal 才拥有真实的原生 session 对象。 +- lal 才知道 RTMP、RTSP、HTTP-FLV、HTTP-TS、HLS-TS、pull、push 的真实生命周期。 +- lal 统计来自真实连接读写字节，lalmax 外层无法无损重建。 +- lal 内部 group 还管理 remuxer、GOP 缓存、HLS muxer、录制、回源、转推。 +- lalmax 如果接管这些状态，要么侵入 lal 内部，要么复制大量逻辑。 +- 复制后会产生“lal 状态”和“lalmax 状态”不一致的问题。 + +典型风险： + +- 拉流 session 已经断开，lalmax 未收到对应事件。 +- lal 内部因错误关闭 group，lalmax 仍保留扩展状态。 +- 同名 `streamName` 在不同 `appName` 下冲突。 +- API 返回的 pub/sub 数量与真实 lal 内部不一致。 +- 后续升级 lal 版本时，lalmax 需要同步内部结构变化。 + +### 适合由 lalmax 管理的内容 + +lalmax 适合管理以下内容： + +- lalmax 扩展协议服务：SRT、WHIP、WHEP、Jessibuca、HTTP-FMP4、HLS-FMP4/LLHLS、GB28181 控制面。 +- 扩展消费者生命周期和统计。 +- 扩展协议相关的缓存、队列、写超时、回放策略。 +- 对外统一 HTTP API 聚合视图。 +- lal 原生状态的只读镜像或快照。 + +也就是说，lalmax 应该做： + +- `lal.StatAllGroup()` 的聚合增强。 +- lalmax 扩展订阅状态补充。 +- 统一 API 输出。 +- 统一控制入口转发给 lal 或 lalmax 对应模块。 + +lalmax 不应该做： + +- 替代 lal 管理 RTMP/RTSP/HTTP-FLV/HTTP-TS/HLS-TS session。 +- 复制 lal `Group` 内部 remux 和协议状态机。 +- 自己维护一份“原生 pub/sub/pull 是否存在”的权威状态。 + +## 推荐架构 + +推荐采用“双层管理，单一事实源”的架构。 + +> 落地说明：第一阶段已将原 `hook` 包迁移为 `lalmax/logic` 扩展管理包。为了避免和上游 `github.com/q191201771/lal/pkg/logic` 导入名冲突，调用侧通常使用 `maxlogic` 作为别名。 + +### 第一层：lal 原生事实源 + +lal `logic` 继续管理： + +- 原生输入输出 session。 +- 原生 group。 +- 原生统计。 +- 原生控制 API。 +- 原生协议转发。 + +lalmax 通过 `ILalServer` 查询： + +- `StatLalInfo` +- `StatAllGroup` +- `StatGroup` +- `CtrlStartRelayPull` +- `CtrlStopRelayPull` +- `CtrlKickSession` + +### 第二层：lalmax 扩展管理层 + +新增或重构 `hook` 为扩展流管理层，例如： + +```text +LalMaxStreamManager + HookGroup(streamKey) + input snapshot + GOP cache for extension protocols + extension subscribers + WHEP + Jessibuca + HTTP-FMP4 + HLS-FMP4/LLHLS + extension stats +``` + +这里的 `HookGroup` 只管理 lalmax 扩展状态，不接管 lal 原生状态。 + +### 第三层：统一 API 门面 + +lalmax HTTP API 输出时： + +```text +lal stat group + + lalmax extension subscribers + + lalmax extension module status + = unified stat response +``` + +这样对外看起来是 lalmax 管理了全部视图，但内部事实源仍然清晰。 + +## Hook 管理层建议设计 + +### 流标识 + +建议引入显式 `StreamKey`： + +```text +StreamKey { + AppName + StreamName +} +``` + +短期可继续兼容只有 `streamName` 的模式： + +- `AppName` 为空时，按 `streamName` 匹配。 +- 未来如果启用 lal `ComplexGroupManager`，可以平滑过渡。 + +### Manager 接口 + +建议从全局 singleton 收敛为可注入对象： + +```text +type Manager interface { + OnInputStart(key, uniqueKey) + OnInputMsg(key, msg) + OnInputStop(key) + AddSubscriber(key, sub) + RemoveSubscriber(key, subscriberID) + GetGroup(key) + StatGroups() +} +``` + +第一阶段已经改为 `GetGroupManagerInstance()`，不再保留旧的 `GetHookSessionManagerInstance()` 入口；后续如果要进一步降低全局状态依赖，可再将 manager 挂到 `LalMaxServer` 实例上。 + +### Group 职责 + +`HookGroup` 建议只做： + +- 保存输入流元信息。 +- 保存音视频头。 +- 保存扩展协议所需 GOP cache。 +- 管理扩展订阅者。 +- 管理扩展订阅者回放顺序。 +- 统计扩展订阅者信息。 +- 在输入流停止时通知扩展订阅者。 + +不建议做： + +- 管理 lal 原生 pub/sub/pull session。 +- 参与 lal 原生 remux 链路。 +- 替代 lal 原生 GOP cache。 + +### Subscriber 接口 + +当前 `IHookSessionSubscriber` 过于简单： + +```text +OnMsg(msg) +OnStop() +``` + +建议扩展为更完整的订阅者描述： + +```text +Subscriber { + ID + Protocol + RemoteAddr + StartTime + ReplayPolicy + OnMsg + OnStop + UpdateStat + Stat +} +``` + +这样 `GetAllConsumer()` 才能返回可靠统计，而不是只返回 `SessionId` 和 `StartTime`。 + +## 迁移方案 + +### 阶段一：保持事实源不变，整理 hook + +目标： + +- 不改 lal 交互方式。 +- 不改扩展协议行为。 +- 只重构 hook 内部管理模型。 + +内容： + +- 新增 `lalmax/logic` 扩展管理包，避免继续使用 `hook` 作为业务层命名。 +- 使用 `Group`、`IGroupManager`、`ComplexGroupManager`、`Subscriber`、`ReplaySubscriber` 等更接近 lal `logic` 的命名。 +- 引入 `StreamKey{AppName, StreamName}`，管理器支持 `appName + streamName` 精确匹配。 +- 保留 `streamName` 单键兼容查找；当空 `appName` 查到多个同名不同 appName 的流时返回未命中，避免随机匹配。 +- 将 `HookSession` 职责迁移到扩展 `Group`，只管理扩展 GOP 缓存和扩展订阅者。 +- 完善扩展 subscriber stat。 +- 保留现有 `AddConsumer` 兼容方法。 +- WHEP、Jessibuca、HTTP-FMP4、HLS-FMP4、stat group 支持可选 `app_name` 参数。 +- lal 当前 `WithOnHookSession` 只传 `uniqueKey` 和 `streamName`，所以 lal 原生回调创建的扩展 `Group` 仍默认 `AppName` 为空；后续如果 lal 回调能提供 appName，可直接改为 `NewGroup(uniqueKey, StreamKey{AppName, StreamName}, ...)`。 + +收益： + +- 风险低。 +- 统计更准确。 +- 后续新增扩展协议更清晰。 + +### 阶段二：lalmax API 做聚合视图 + +目标： + +- lal 仍是原生事实源。 +- lalmax 输出统一完整状态。 + +内容： + +- `statGroupHandler` 从 lal 获取原生 `StatGroup`。 +- 从 lalmax manager 获取扩展订阅者。 +- 合并到响应中。 +- 为扩展订阅者设置明确协议名，例如 `WHEP`、`JESSIBUCA`、`FMP4`、`LLHLS`。 + +收益： + +- 对外统一。 +- 内部边界清晰。 + +### 阶段三：事件化和异步化 + +目标： + +- 降低 hook 对 lal 核心链路的阻塞风险。 + +内容： + +- `OnMsg` 内部尽量只做轻量复制和入队。 +- 扩展分发在独立 goroutine 中完成。 +- 每个订阅者有明确队列上限和丢弃/断开策略。 +- 慢消费者不会拖住 lal 原生转发。 + +收益： + +- 更适合多扩展订阅者和慢客户端。 +- 避免 WHEP/FMP4 等写阻塞影响原始流。 + +### 阶段四：按需考虑 appName 复杂管理 + +目标： + +- 支持多 appName 同 streamName。 + +内容： + +- 跟随 lal `ComplexGroupManager` 的语义。 +- 统一流 key 解析规则。 +- API 支持 `app_name` 可选参数。 + +收益： + +- 适合更复杂的多租户或多应用路径。 + +## 不推荐方案 + +### 方案：lalmax 接管 lal 全部流状态 + +不推荐，除非准备 fork lal 并长期维护。 + +问题： + +- 会复制 lal 内部逻辑。 +- 会破坏 lal 作为内核库的边界。 +- 会增加升级 lal 的成本。 +- 会引入双事实源一致性问题。 + +### 方案：hook 直接改成 lal `Group` 等价物 + +不推荐。 + +问题： + +- hook 没有原生 session 对象。 +- hook 只接收已经 remux 成 RTMP 的消息。 +- 管理范围和 lal `Group` 不对等。 +- 容易形成“看起来像 logic，但不具备 logic 真实能力”的半成品。 + +## 建议结论 + +推荐方向： + +- lal 保持流媒体内核和原生状态事实源。 +- lalmax 增加自己的扩展流管理层，借鉴 lal `logic` 的 manager/group/session 模型。 +- lalmax API 做统一聚合视图，而不是把 lal 的状态搬出来重新管理。 +- hook 改造重点放在扩展订阅者生命周期、统计、队列、回放和慢消费者隔离。 + +一句话：`hook` 可以“logic 化”，但应该是 lalmax 扩展层的 logic，而不是替代 lal 的 logic。 diff --git a/image/gb-hk.png b/document/images/gb-hk.png similarity index 100% rename from image/gb-hk.png rename to document/images/gb-hk.png diff --git a/image/init.png b/document/images/init.png similarity index 100% rename from image/init.png rename to document/images/init.png diff --git a/image/rtc_01.jpeg b/document/images/rtc_01.jpeg similarity index 100% rename from image/rtc_01.jpeg rename to document/images/rtc_01.jpeg diff --git a/image/rtc_02.png b/document/images/rtc_02.png similarity index 100% rename from image/rtc_02.png rename to document/images/rtc_02.png diff --git a/image/srt_0.png b/document/images/srt_0.png similarity index 100% rename from image/srt_0.png rename to document/images/srt_0.png diff --git a/image/srt_1.png b/document/images/srt_1.png similarity index 100% rename from image/srt_1.png rename to document/images/srt_1.png diff --git a/image/srt_2.png b/document/images/srt_2.png similarity index 100% rename from image/srt_2.png rename to document/images/srt_2.png diff --git a/image/srt_3.png b/document/images/srt_3.png similarity index 100% rename from image/srt_3.png rename to document/images/srt_3.png diff --git a/document/lal_api.md b/document/lal_api.md index 90b12d9..33b281f 100644 --- a/document/lal_api.md +++ b/document/lal_api.md @@ -17,7 +17,7 @@ http://127.0.0.1:8083 ``` -lalmax 自身也在 `lalmax.http_config.http_listen_addr` 上提供 `/api/stat` 和 `/api/ctrl` 兼容接口，并会补充 lalmax hook 订阅信息。只需要管理 lal 原生流状态时，可以直接使用本文档中的 lal 原生 API。 +lalmax 自身也在 `lalmax.http_config.http_listen_addr` 上提供 `/api/stat` 和 `/api/ctrl` 兼容接口，并会补充 lalmax 扩展订阅信息。只需要管理 lal 原生流状态时，可以直接使用本文档中的 lal 原生 API。 ## 通用响应 @@ -104,6 +104,7 @@ curl "http://127.0.0.1:8083/api/stat/group?stream_name=test110" | 参数 | 必填 | 说明 | | --- | --- | --- | | `stream_name` | 是 | 流名称 | +| `app_name` | 否 | lalmax 兼容 API 可用于精确匹配扩展订阅者；lal 原生 API 当前仍主要按 `stream_name` 查询 | 响应示例： @@ -311,7 +312,7 @@ http://127.0.0.1:1290/api/ctrl/kick_session http://127.0.0.1:1290/api/ctrl/start_rtp_pub ``` -lalmax 兼容 API 的请求和响应结构与 lal 原生 API 基本一致，但会在统计结果中补充 lalmax hook 订阅者信息。控制类接口还可能受 `lalmax.http_config.ctrl_auth_whitelist` 限制。 +lalmax 兼容 API 的请求和响应结构与 lal 原生 API 基本一致，但会在统计结果中补充 lalmax 扩展订阅者信息。控制类接口还可能受 `lalmax.http_config.ctrl_auth_whitelist` 限制。 ## 鉴权说明 diff --git a/document/rtc.md b/document/rtc.md index 485e6b1..e6876e0 100644 --- a/document/rtc.md +++ b/document/rtc.md @@ -92,9 +92,9 @@ WHEP拉流可以使用[vue-wish](https://github.com/zllovesuki/vue-wish)测试 OBS推流配置 -![图片](../image/rtc_01.jpeg) +![图片](images/rtc_01.jpeg) vue-wish拉流效果 -![图片](../image/rtc_02.png) +![图片](images/rtc_02.png) diff --git a/document/srt.md b/document/srt.md index c5e8728..04a3070 100644 --- a/document/srt.md +++ b/document/srt.md @@ -35,14 +35,14 @@ SRT(Secure Reliable Transport)的简称,主要优化在不可靠网络(非阻塞 (1) 启动LalMax服务 (2) 使用OBS进行推流，在"直播"中输入srt的推流地址 -![图片](../image/srt_0.png) +![图片](images/srt_0.png) (3) VLC进行播放 在VLC中设置streamid,这部分填streamid后面的所有信息 -![图片](../image/srt_1.png) +![图片](images/srt_1.png) 输入streamid前面的部分进行拉流 -![图片](../image/srt_2.png) +![图片](images/srt_2.png) -![图片](../image/srt_3.png) \ No newline at end of file +![图片](images/srt_3.png) diff --git a/document/stream_url.md b/document/stream_url.md index 1f3d866..fb1d180 100644 --- a/document/stream_url.md +++ b/document/stream_url.md @@ -7,6 +7,7 @@ - `lal` 原生能力使用 `lal` 配置段中的端口，例如 RTMP、RTSP、HTTP-FLV、HLS-TS、HTTP-TS。 - `lalmax` 扩展能力使用 `lalmax` 配置段中的端口，例如 SRT、WHIP/WHEP、HTTP-FMP4、HLS-FMP4/LLHLS。 - 当前 lal 使用简单流管理时主要按 `streamName` 匹配。示例中的 `/live/test110` 里，`test110` 是流名，`live` 可作为常用路径前缀。 +- lalmax 扩展拉流接口支持可选 `app_name` 参数，用于未来多 appName 同 streamName 的精确匹配；不传时仍按历史 `streamName` 兼容查找。 - HTTP-FLV、HTTP-TS、HLS-TS 的路径还受 `lal.httpflv.url_pattern`、`lal.httpts.url_pattern`、`lal.hls.url_pattern` 影响。示例配置中 HTTP-FLV 的 `url_pattern` 为 `/`，因此 `/live/test110.flv` 可用。 - HTTPS、RTMPS、RTSPS 依赖配置中的证书文件，浏览器或播放器可能需要信任测试证书。 @@ -148,6 +149,12 @@ http://127.0.0.1:1290/webrtc/whep?streamid=test110 https://127.0.0.1:1233/webrtc/whep?streamid=test110 ``` +如果需要指定 appName： + +```text +http://127.0.0.1:1290/webrtc/whep?streamid=test110&app_name=live +``` + WHEP 使用 HTTP POST 传输 SDP offer，通常由 WHEP 播放器或 WebRTC 工具调用。 ### Jessibuca DataChannel @@ -156,6 +163,12 @@ WHEP 使用 HTTP POST 传输 SDP offer，通常由 WHEP 播放器或 WebRTC 工 webrtc://127.0.0.1:1290/webrtc/play/live/test110 ``` +如果需要指定 appName： + +```text +webrtc://127.0.0.1:1290/webrtc/play/live/test110?app_name=live +``` + ### HTTP-FMP4 ```text @@ -163,13 +176,25 @@ http://127.0.0.1:1290/live/m4s/test110.mp4 https://127.0.0.1:1233/live/m4s/test110.mp4 ``` +如果需要指定 appName： + +```text +http://127.0.0.1:1290/live/m4s/test110.mp4?app_name=live +``` + ### HLS-FMP4/LLHLS -需要启用 `lalmax.hls_config.enable`。 +需要启用 `lalmax.fmp4_config.hls.enable`。 ```text http://127.0.0.1:1290/live/hls/test110/index.m3u8 https://127.0.0.1:1233/live/hls/test110/index.m3u8 ``` -如果需要低延迟 HLS，设置 `lalmax.hls_config.low_latency` 为 `true`。 +如果需要指定 appName： + +```text +http://127.0.0.1:1290/live/hls/test110/index.m3u8?app_name=live +``` + +如果需要低延迟 HLS，设置 `lalmax.fmp4_config.hls.low_latency` 为 `true`。 diff --git a/fmp4/hls/server.go b/fmp4/hls/server.go index 2263f56..a7bb0dc 100644 --- a/fmp4/hls/server.go +++ b/fmp4/hls/server.go @@ -4,7 +4,7 @@ import ( "sync" "time" - config "github.com/q191201771/lalmax/conf" + config "github.com/q191201771/lalmax/config" "github.com/gin-gonic/gin" "github.com/q191201771/lal/pkg/base" @@ -13,11 +13,11 @@ import ( type HlsServer struct { sessions sync.Map - conf config.HlsConfig + conf config.Fmp4HlsConfig invalidSessions sync.Map } -func NewHlsServer(conf config.HlsConfig) *HlsServer { +func NewHlsServer(conf config.Fmp4HlsConfig) *HlsServer { svr := &HlsServer{ conf: conf, } @@ -28,13 +28,21 @@ func NewHlsServer(conf config.HlsConfig) *HlsServer { } func (s *HlsServer) NewHlsSession(streamName string) { - nazalog.Info("new hls session, streamName:", streamName) - session := NewHlsSession(streamName, s.conf) - s.sessions.Store(streamName, session) + s.NewHlsSessionWithAppName("", streamName) +} + +func (s *HlsServer) NewHlsSessionWithAppName(appName, streamName string) { + nazalog.Infof("new hls session, appName:%s, streamName:%s", appName, streamName) + session := NewHlsSessionWithAppName(appName, streamName, s.conf) + s.sessions.Store(hlsSessionKey(appName, streamName), session) } func (s *HlsServer) OnMsg(streamName string, msg base.RtmpMsg) { - value, ok := s.sessions.Load(streamName) + s.OnMsgWithAppName("", streamName, msg) +} + +func (s *HlsServer) OnMsgWithAppName(appName, streamName string, msg base.RtmpMsg) { + value, ok := s.sessions.Load(hlsSessionKey(appName, streamName)) if ok { session := value.(*HlsSession) session.OnMsg(msg) @@ -42,20 +50,63 @@ func (s *HlsServer) OnMsg(streamName string, msg base.RtmpMsg) { } func (s *HlsServer) OnStop(streamName string) { - value, ok := s.sessions.Load(streamName) + s.OnStopWithAppName("", streamName) +} + +func (s *HlsServer) OnStopWithAppName(appName, streamName string) { + key := hlsSessionKey(appName, streamName) + value, ok := s.sessions.Load(key) if ok { session := value.(*HlsSession) s.invalidSessions.Store(session.SessionId, session) - s.sessions.Delete(streamName) + s.sessions.Delete(key) } } func (s *HlsServer) HandleRequest(ctx *gin.Context) { streamName := ctx.Param("streamid") - value, ok := s.sessions.Load(streamName) + appName := ctx.Query("app_name") + if session, ok := s.getSession(appName, streamName); ok { + session.HandleRequest(ctx) + } +} + +func (s *HlsServer) getSession(appName, streamName string) (*HlsSession, bool) { + value, ok := s.sessions.Load(hlsSessionKey(appName, streamName)) if ok { + return value.(*HlsSession), true + } + + if appName != "" { + return nil, false + } + + var found *HlsSession + matchCount := 0 + s.sessions.Range(func(_, value interface{}) bool { session := value.(*HlsSession) - session.HandleRequest(ctx) + if session.streamName != streamName { + return true + } + found = session + matchCount++ + return matchCount <= 1 + }) + if matchCount != 1 { + return nil, false + } + return found, true +} + +type sessionKey struct { + appName string + streamName string +} + +func hlsSessionKey(appName, streamName string) sessionKey { + return sessionKey{ + appName: appName, + streamName: streamName, } } diff --git a/fmp4/hls/session.go b/fmp4/hls/session.go index 2f16af1..c1777b4 100644 --- a/fmp4/hls/session.go +++ b/fmp4/hls/session.go @@ -3,7 +3,7 @@ package hls import ( "time" - config "github.com/q191201771/lalmax/conf" + config "github.com/q191201771/lalmax/config" "github.com/bluenviron/gohlslib" "github.com/bluenviron/gohlslib/pkg/codecs" @@ -23,6 +23,7 @@ type HlsSession struct { audioCodecId int videoCodecId int maxMsgSize int + appName string streamName string sps []byte pps []byte @@ -35,7 +36,11 @@ type HlsSession struct { SessionId string } -func NewHlsSession(streamName string, conf config.HlsConfig) *HlsSession { +func NewHlsSession(streamName string, conf config.Fmp4HlsConfig) *HlsSession { + return NewHlsSessionWithAppName("", streamName, conf) +} + +func NewHlsSessionWithAppName(appName, streamName string, conf config.Fmp4HlsConfig) *HlsSession { variant := gohlslib.MuxerVariantFMP4 if conf.LowLatency { variant = gohlslib.MuxerVariantLowLatency @@ -51,6 +56,7 @@ func NewHlsSession(streamName string, conf config.HlsConfig) *HlsSession { videoCodecId: -1, maxMsgSize: 128, data: make([]Frame, 10)[0:0], + appName: appName, streamName: streamName, SessionId: u.String(), } @@ -320,7 +326,7 @@ func (session *HlsSession) OnStop() { } func (session *HlsSession) HandleRequest(ctx *gin.Context) { - nazalog.Info("handle hls request, streamName:", session.streamName, " path:", ctx.Request.URL.Path) + nazalog.Info("handle hls request, appName:", session.appName, " streamName:", session.streamName, " path:", ctx.Request.URL.Path) session.muxer.Handle(ctx.Writer, ctx.Request) } diff --git a/fmp4/http-fmp4/server.go b/fmp4/http-fmp4/server.go index 02e27c4..aff51aa 100644 --- a/fmp4/http-fmp4/server.go +++ b/fmp4/http-fmp4/server.go @@ -15,7 +15,8 @@ func NewHttpFmp4Server() *HttpFmp4Server { func (s *HttpFmp4Server) HandleRequest(c *gin.Context) { streamid := c.Param("streamid") + appName := c.Query("app_name") - session := NewHttpFmp4Session(streamid) + session := NewHttpFmp4Session(appName, streamid) session.handleSession(c) } diff --git a/fmp4/http-fmp4/session.go b/fmp4/http-fmp4/session.go index 5393a95..3be5c39 100644 --- a/fmp4/http-fmp4/session.go +++ b/fmp4/http-fmp4/session.go @@ -8,7 +8,7 @@ import ( "time" "github.com/q191201771/lalmax/fmp4/muxer" - "github.com/q191201771/lalmax/hook" + maxlogic "github.com/q191201771/lalmax/logic" "github.com/gofrs/uuid" "github.com/q191201771/naza/pkg/connection" @@ -26,8 +26,9 @@ var ( ) type HttpFmp4Session struct { + appName string streamid string - hooks *hook.HookSession + group *maxlogic.Group subscriberId string rtmp2Fmp4Remuxer *muxer.Rtmp2Fmp4Remuxer @@ -37,12 +38,13 @@ type HttpFmp4Session struct { log nazalog.Logger } -func NewHttpFmp4Session(streamid string) *HttpFmp4Session { +func NewHttpFmp4Session(appName, streamid string) *HttpFmp4Session { streamid = strings.TrimSuffix(streamid, ".mp4") u, _ := uuid.NewV4() session := &HttpFmp4Session{ + appName: appName, streamid: streamid, subscriberId: u.String(), log: nazalog.WithPrefix(u.String()), @@ -50,7 +52,7 @@ func NewHttpFmp4Session(streamid string) *HttpFmp4Session { session.rtmp2Fmp4Remuxer = muxer.NewRtmp2Fmp4Remuxer(session).WithLog(session.log) - session.log.Info("create http fmp4 seesion, streamid:", streamid) + session.log.Infof("create http fmp4 session, appName:%s, streamid:%s", appName, streamid) return session } @@ -82,14 +84,14 @@ func (session *HttpFmp4Session) dispose() error { return retErr } func (session *HttpFmp4Session) handleSession(c *gin.Context) { - ok, hooksession := hook.GetHookSessionManagerInstance().GetHookSession(session.streamid) + ok, group := maxlogic.GetGroupManagerInstance().GetGroup(maxlogic.NewStreamKey(session.appName, session.streamid)) if !ok { - nazalog.Error("stream is not found, streamid:", session.streamid) + nazalog.Errorf("stream is not found, appName:%s, streamid:%s", session.appName, session.streamid) c.Status(http.StatusNotFound) return } - session.hooks = hooksession + session.group = group session.w = c.Writer c.Header("Content-Type", "video/mp4") @@ -117,7 +119,10 @@ func (session *HttpFmp4Session) handleSession(c *gin.Context) { nazalog.Errorf("session writeHttpHeader. err=%+v", err) return } - session.hooks.AddConsumer(session.subscriberId, session) + session.group.AddSubscriber(maxlogic.SubscriberInfo{ + SubscriberID: session.subscriberId, + Protocol: maxlogic.SubscriberProtocolHTTPFMP4, + }, session) go func() { readBuf := make([]byte, 1024) @@ -162,5 +167,7 @@ func (session *HttpFmp4Session) OnMsg(msg base.RtmpMsg) { } func (session *HttpFmp4Session) OnStop() { - session.hooks.RemoveConsumer(session.subscriberId) + if session.group != nil { + session.group.RemoveSubscriber(session.subscriberId) + } } diff --git a/gb28181/channel.go b/gb28181/channel.go index f606fc4..b715646 100644 --- a/gb28181/channel.go +++ b/gb28181/channel.go @@ -10,7 +10,7 @@ import ( "github.com/q191201771/naza/pkg/nazaatomic" - config "github.com/q191201771/lalmax/conf" + config "github.com/q191201771/lalmax/config" "github.com/q191201771/lalmax/gb28181/mediaserver" "github.com/ghettovoice/gosip/sip" diff --git a/gb28181/device.go b/gb28181/device.go index cd29e27..17ffe3f 100644 --- a/gb28181/device.go +++ b/gb28181/device.go @@ -8,7 +8,7 @@ import ( "sync" "time" - config "github.com/q191201771/lalmax/conf" + config "github.com/q191201771/lalmax/config" "github.com/ghettovoice/gosip/sip" "github.com/q191201771/naza/pkg/nazalog" diff --git a/gb28181/server.go b/gb28181/server.go index 9d93f60..72ac3db 100644 --- a/gb28181/server.go +++ b/gb28181/server.go @@ -12,7 +12,7 @@ import ( "time" udpTransport "github.com/pion/transport/v3/udp" - config "github.com/q191201771/lalmax/conf" + config "github.com/q191201771/lalmax/config" "github.com/q191201771/lalmax/gb28181/mediaserver" "github.com/ghettovoice/gosip" diff --git a/hook/hookmanager.go b/hook/hookmanager.go deleted file mode 100644 index 6ccdf4c..0000000 --- a/hook/hookmanager.go +++ /dev/null @@ -1,46 +0,0 @@ -package hook - -import ( - "sync" - - "github.com/q191201771/naza/pkg/nazalog" -) - -type HookSessionMangaer struct { - sessionMap sync.Map -} - -var ( - manager *HookSessionMangaer - once sync.Once -) - -func GetHookSessionManagerInstance() *HookSessionMangaer { - once.Do(func() { - manager = &HookSessionMangaer{} - }) - - return manager -} - -func (m *HookSessionMangaer) SetHookSession(streamName string, session *HookSession) { - nazalog.Info("SetHookSession, streamName:", streamName) - m.sessionMap.Store(streamName, session) -} - -func (m *HookSessionMangaer) RemoveHookSession(streamName string) { - nazalog.Info("RemoveHookSession, streamName:", streamName) - // s, ok := m.sessionMap.Load(streamName) - // if ok { - m.sessionMap.Delete(streamName) - // } -} - -func (m *HookSessionMangaer) GetHookSession(streamName string) (bool, *HookSession) { - s, ok := m.sessionMap.Load(streamName) - if ok { - return true, s.(*HookSession) - } - - return false, nil -} diff --git a/hook/hooksession.go b/hook/hooksession.go deleted file mode 100644 index 6e5d3cd..0000000 --- a/hook/hooksession.go +++ /dev/null @@ -1,300 +0,0 @@ -package hook - -import ( - "sync" - "time" - - "github.com/q191201771/lalmax/fmp4/hls" - - "github.com/q191201771/lal/pkg/base" - "github.com/q191201771/naza/pkg/nazalog" -) - -var _ base.ISession = (*consumerInfo)(nil) - -type IHookSessionSubscriber interface { - OnMsg(msg base.RtmpMsg) - OnStop() -} - -type IHookSessionReplaySubscriber interface { - OnReplayStart() - OnReplayStop() -} - -type HookSession struct { - uniqueKey string - streamName string - consumers sync.Map - hlssvr *hls.HlsServer - gopCache *GopCache - gopCacheMux sync.RWMutex - msgMux sync.Mutex - hasVideo bool -} - -type consumerInfo struct { - subscriber IHookSessionSubscriber - hasSendVideo bool - replayCache bool - writeMux sync.Mutex - - base.StatSession -} - -// AppName implements base.ISession. -func (c *consumerInfo) AppName() string { - return c.SessionId -} - -// GetStat implements base.ISession. -func (c *consumerInfo) GetStat() base.StatSession { - return c.StatSession -} - -// IsAlive implements base.ISession. -func (c *consumerInfo) IsAlive() (readAlive bool, writeAlive bool) { - return true, true -} - -// RawQuery implements base.ISession. -func (c *consumerInfo) RawQuery() string { - return "" -} - -// StreamName implements base.ISession. -func (c *consumerInfo) StreamName() string { - return c.SessionId -} - -// UniqueKey implements base.ISession. -func (c *consumerInfo) UniqueKey() string { - return c.SessionId -} - -// UpdateStat implements base.ISession. -func (c *consumerInfo) UpdateStat(intervalSec uint32) { -} - -// Url implements base.ISession. -func (*consumerInfo) Url() string { - return "" -} - -func NewHookSession(uniqueKey, streamName string, hlssvr *hls.HlsServer, gopNum, singleGopMaxFrameNum int) *HookSession { - s := &HookSession{ - uniqueKey: uniqueKey, - streamName: streamName, - hlssvr: hlssvr, - gopCache: NewGopCache(gopNum, singleGopMaxFrameNum), - } - - if s.hlssvr != nil { - s.hlssvr.NewHlsSession(streamName) - } - - nazalog.Infof("create hook session, uniqueKey:%s, streamName:%s", uniqueKey, streamName) - - GetHookSessionManagerInstance().SetHookSession(streamName, s) - return s -} - -func (session *HookSession) OnMsg(msg base.RtmpMsg) { - if session.hlssvr != nil { - session.hlssvr.OnMsg(session.streamName, msg) - } - - session.msgMux.Lock() - hasVideo := session.hasVideo - consumers := make([]*consumerInfo, 0) - session.consumers.Range(func(key, value interface{}) bool { - if c, ok := value.(*consumerInfo); ok { - consumers = append(consumers, c) - } - return true - }) - - if !session.hasVideo && msg.IsVideoKeyNalu() { - session.hasVideo = true - } - - session.gopCacheMux.Lock() - session.gopCache.Feed(msg) - session.gopCacheMux.Unlock() - session.msgMux.Unlock() - - for _, c := range consumers { - session.handleConsumerMsg(c, msg, hasVideo) - } -} - -func (session *HookSession) OnStop() { - if session.hlssvr != nil { - session.hlssvr.OnStop(session.streamName) - } - - nazalog.Debugf("OnStop, uniqueKey:%s, streamName:%s", session.uniqueKey, session.streamName) - session.consumers.Range(func(key, value interface{}) bool { - c := value.(*consumerInfo) - if c.subscriber != nil { - c.subscriber.OnStop() - } - return true - }) - - GetHookSessionManagerInstance().RemoveHookSession(session.streamName) -} - -func (session *HookSession) AddConsumer(consumerId string, subscriber IHookSessionSubscriber) { - session.AddConsumerWithReplay(consumerId, subscriber, true) -} - -func (session *HookSession) AddConsumerWithReplay(consumerId string, subscriber IHookSessionSubscriber, replayCache bool) { - info := &consumerInfo{ - subscriber: subscriber, - replayCache: replayCache, - StatSession: base.StatSession{ - SessionId: consumerId, - StartTime: time.Now().Format(time.DateTime), - // Protocol: , TODO: (xugo)需要传递更多的参数来填充数据 - }, - } - - nazalog.Info("AddConsumer, consumerId:", consumerId) - if replayCache { - info.writeMux.Lock() - } - var replayMsgs []base.RtmpMsg - - session.msgMux.Lock() - session.consumers.Store(consumerId, info) - if replayCache { - replayMsgs = session.getGopReplayMessages() - } - session.msgMux.Unlock() - - if replayCache { - session.replayGopMessagesLocked(info, replayMsgs) - info.writeMux.Unlock() - } -} - -func (session *HookSession) GetAllConsumer() []base.StatSub { - out := make([]base.StatSub, 0, 10) - session.consumers.Range(func(key, value any) bool { - v, ok := value.(*consumerInfo) - if ok { - // TODO: (xugo)先简单实现，此处需要优化数据准确性 - out = append(out, base.Session2StatSub(v)) - } - return true - }) - return out -} - -func (session *HookSession) RemoveConsumer(consumerId string) { - _, ok := session.consumers.Load(consumerId) - if ok { - nazalog.Info("RemoveConsumer, consumerId:", consumerId) - session.consumers.Delete(consumerId) - } -} - -func (session *HookSession) GetVideoSeqHeaderMsg() *base.RtmpMsg { - session.gopCacheMux.RLock() - defer session.gopCacheMux.RUnlock() - if session.gopCache.videoheader == nil { - return nil - } - m := session.gopCache.videoheader.Clone() - return &m -} - -func (session *HookSession) GetAudioSeqHeaderMsg() *base.RtmpMsg { - session.gopCacheMux.RLock() - defer session.gopCacheMux.RUnlock() - if session.gopCache.audioheader == nil { - return nil - } - m := session.gopCache.audioheader.Clone() - return &m -} - -func (session *HookSession) handleConsumerMsg(c *consumerInfo, msg base.RtmpMsg, hasVideo bool) { - if c == nil { - return - } - - c.writeMux.Lock() - defer c.writeMux.Unlock() - - if c.subscriber == nil { - return - } - - if msg.Header.MsgTypeId == base.RtmpTypeIdVideo { - if !c.hasSendVideo { - if !msg.IsVideoKeyNalu() { - return - } - if v := session.GetVideoSeqHeaderMsg(); v != nil { - c.subscriber.OnMsg(*v) - } - if v := session.GetAudioSeqHeaderMsg(); v != nil && v.IsAacSeqHeader() { - c.subscriber.OnMsg(*v) - } - c.hasSendVideo = true - } - - c.subscriber.OnMsg(msg) - } else if msg.Header.MsgTypeId == base.RtmpTypeIdAudio { - if !hasVideo || c.hasSendVideo { - c.subscriber.OnMsg(msg) - } - } -} - -func (session *HookSession) replayGopMessagesLocked(c *consumerInfo, msgs []base.RtmpMsg) { - if c == nil || c.subscriber == nil || c.hasSendVideo || !c.replayCache { - return - } - - if len(msgs) == 0 { - return - } - - if replaySubscriber, ok := c.subscriber.(IHookSessionReplaySubscriber); ok { - replaySubscriber.OnReplayStart() - defer replaySubscriber.OnReplayStop() - } - - for _, msg := range msgs { - c.subscriber.OnMsg(msg) - } - c.hasSendVideo = true -} - -func (session *HookSession) getGopReplayMessages() []base.RtmpMsg { - session.gopCacheMux.RLock() - defer session.gopCacheMux.RUnlock() - - gopCount := session.gopCache.GetGopCount() - if gopCount == 0 { - return nil - } - - msgs := make([]base.RtmpMsg, 0, gopCount) - if v := session.gopCache.videoheader; v != nil { - msgs = append(msgs, v.Clone()) - } - if v := session.gopCache.audioheader; v != nil && v.IsAacSeqHeader() { - msgs = append(msgs, v.Clone()) - } - for i := 0; i < gopCount; i++ { - for _, item := range session.gopCache.GetGopDataAt(i) { - msgs = append(msgs, item.Clone()) - } - } - - return msgs -} diff --git a/hook/hooksession_test.go b/hook/hooksession_test.go deleted file mode 100644 index 6cbfe84..0000000 --- a/hook/hooksession_test.go +++ /dev/null @@ -1,295 +0,0 @@ -package hook - -import ( - "sync" - "testing" - "time" - - "github.com/q191201771/lal/pkg/base" -) - -type recordSubscriber struct { - msgs []base.RtmpMsg -} - -func (s *recordSubscriber) OnMsg(msg base.RtmpMsg) { - s.msgs = append(s.msgs, msg.Clone()) -} - -func (s *recordSubscriber) OnStop() {} - -type blockingSubscriber struct { - mu sync.Mutex - msgs []base.RtmpMsg - blocked chan struct{} - release chan struct{} - replaying bool - blockOnce sync.Once -} - -func newBlockingSubscriber() *blockingSubscriber { - return &blockingSubscriber{ - blocked: make(chan struct{}), - release: make(chan struct{}), - } -} - -func (s *blockingSubscriber) OnMsg(msg base.RtmpMsg) { - s.mu.Lock() - s.msgs = append(s.msgs, msg.Clone()) - shouldBlock := s.replaying - s.mu.Unlock() - - if shouldBlock { - s.blockOnce.Do(func() { - close(s.blocked) - <-s.release - }) - } -} - -func (s *blockingSubscriber) OnStop() {} - -func (s *blockingSubscriber) OnReplayStart() { - s.mu.Lock() - s.replaying = true - s.mu.Unlock() -} - -func (s *blockingSubscriber) OnReplayStop() { - s.mu.Lock() - s.replaying = false - s.mu.Unlock() -} - -func (s *blockingSubscriber) markers() []byte { - s.mu.Lock() - defer s.mu.Unlock() - - out := make([]byte, 0, len(s.msgs)) - for _, msg := range s.msgs { - out = append(out, payloadMarker(msg)) - } - return out -} - -func videoSeqHeader(marker byte) base.RtmpMsg { - return base.RtmpMsg{ - Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdVideo}, - Payload: []byte{ - base.RtmpAvcKeyFrame, - base.RtmpAvcPacketTypeSeqHeader, - 0, 0, 0, - marker, - }, - } -} - -func videoKeyNalu(marker byte) base.RtmpMsg { - return base.RtmpMsg{ - Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdVideo}, - Payload: []byte{ - base.RtmpAvcKeyFrame, - base.RtmpAvcPacketTypeNalu, - 0, 0, 0, - marker, - }, - } -} - -func videoInterNalu(marker byte) base.RtmpMsg { - return base.RtmpMsg{ - Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdVideo}, - Payload: []byte{ - base.RtmpAvcInterFrame, - base.RtmpAvcPacketTypeNalu, - 0, 0, 0, - marker, - }, - } -} - -func aacSeqHeader(marker byte) base.RtmpMsg { - return base.RtmpMsg{ - Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdAudio}, - Payload: []byte{ - base.RtmpSoundFormatAac << 4, - base.RtmpAacPacketTypeSeqHeader, - marker, - }, - } -} - -func aacRaw(marker byte) base.RtmpMsg { - return base.RtmpMsg{ - Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdAudio}, - Payload: []byte{ - base.RtmpSoundFormatAac << 4, - base.RtmpAacPacketTypeRaw, - marker, - }, - } -} - -func g711aAudio(marker byte) base.RtmpMsg { - return base.RtmpMsg{ - Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdAudio}, - Payload: []byte{base.RtmpSoundFormatG711A<<4 | marker}, - } -} - -func payloadMarker(msg base.RtmpMsg) byte { - return msg.Payload[len(msg.Payload)-1] -} - -func TestAddConsumerReplaysCachedGopImmediately(t *testing.T) { - session := NewHookSession("test-replay", "test-replay", nil, 1, 0) - defer GetHookSessionManagerInstance().RemoveHookSession("test-replay") - - session.OnMsg(videoSeqHeader(1)) - session.OnMsg(aacSeqHeader(2)) - session.OnMsg(videoKeyNalu(3)) - session.OnMsg(aacRaw(4)) - session.OnMsg(videoInterNalu(5)) - - sub := &recordSubscriber{} - session.AddConsumer("consumer", sub) - - if len(sub.msgs) != 5 { - t.Fatalf("expected 5 replay messages, got %d", len(sub.msgs)) - } - - wantMarkers := []byte{1, 2, 3, 4, 5} - for i, want := range wantMarkers { - if got := payloadMarker(sub.msgs[i]); got != want { - t.Fatalf("message %d marker = %d, want %d", i, got, want) - } - } -} - -func TestVideoSeqHeaderChangeClearsStaleGop(t *testing.T) { - session := NewHookSession("test-clear", "test-clear", nil, 1, 0) - defer GetHookSessionManagerInstance().RemoveHookSession("test-clear") - - session.OnMsg(videoSeqHeader(1)) - session.OnMsg(videoKeyNalu(2)) - session.OnMsg(videoInterNalu(3)) - session.OnMsg(videoSeqHeader(4)) - - sub := &recordSubscriber{} - session.AddConsumer("consumer", sub) - if len(sub.msgs) != 0 { - t.Fatalf("expected no stale GOP replay after sequence header change, got %d messages", len(sub.msgs)) - } - - session.OnMsg(videoKeyNalu(5)) - if len(sub.msgs) != 2 { - t.Fatalf("expected new header and current key frame, got %d messages", len(sub.msgs)) - } - if got := payloadMarker(sub.msgs[0]); got != 4 { - t.Fatalf("header marker = %d, want 4", got) - } - if got := payloadMarker(sub.msgs[1]); got != 5 { - t.Fatalf("key frame marker = %d, want 5", got) - } -} - -func TestNonAacAudioIsNotReplayedAsHeader(t *testing.T) { - session := NewHookSession("test-g711", "test-g711", nil, 1, 0) - defer GetHookSessionManagerInstance().RemoveHookSession("test-g711") - - session.OnMsg(videoSeqHeader(1)) - session.OnMsg(videoKeyNalu(2)) - session.OnMsg(g711aAudio(3)) - - sub := &recordSubscriber{} - session.AddConsumer("consumer", sub) - - if len(sub.msgs) != 3 { - t.Fatalf("expected video header, key frame and one G711 packet, got %d messages", len(sub.msgs)) - } - - wantMarkers := []byte{1, 2, base.RtmpSoundFormatG711A<<4 | 3} - for i, want := range wantMarkers { - if got := payloadMarker(sub.msgs[i]); got != want { - t.Fatalf("message %d marker = %d, want %d", i, got, want) - } - } -} - -func TestAddConsumerWithReplayDisabledDoesNotReplayCachedGop(t *testing.T) { - session := NewHookSession("test-no-replay", "test-no-replay", nil, 1, 0) - defer GetHookSessionManagerInstance().RemoveHookSession("test-no-replay") - - session.OnMsg(videoSeqHeader(1)) - session.OnMsg(videoKeyNalu(2)) - session.OnMsg(videoInterNalu(3)) - - sub := &recordSubscriber{} - session.AddConsumerWithReplay("consumer", sub, false) - - if len(sub.msgs) != 0 { - t.Fatalf("expected no cached messages when replay is disabled, got %d messages", len(sub.msgs)) - } - - session.OnMsg(videoInterNalu(4)) - if len(sub.msgs) != 0 { - t.Fatalf("expected to wait for next key frame, got %d messages", len(sub.msgs)) - } - - session.OnMsg(videoKeyNalu(5)) - if len(sub.msgs) != 2 { - t.Fatalf("expected header and current key frame, got %d messages", len(sub.msgs)) - } - if got := payloadMarker(sub.msgs[0]); got != 1 { - t.Fatalf("header marker = %d, want 1", got) - } - if got := payloadMarker(sub.msgs[1]); got != 5 { - t.Fatalf("key frame marker = %d, want 5", got) - } -} - -func TestAddConsumerReplayDoesNotInterleaveWithLiveKeyFrame(t *testing.T) { - session := NewHookSession("test-replay-order", "test-replay-order", nil, 1, 0) - defer GetHookSessionManagerInstance().RemoveHookSession("test-replay-order") - - session.OnMsg(videoSeqHeader(1)) - session.OnMsg(videoKeyNalu(2)) - session.OnMsg(videoInterNalu(3)) - - sub := newBlockingSubscriber() - addDone := make(chan struct{}) - go func() { - session.AddConsumer("consumer", sub) - close(addDone) - }() - - <-sub.blocked - - liveDone := make(chan struct{}) - go func() { - session.OnMsg(videoKeyNalu(4)) - close(liveDone) - }() - - select { - case <-liveDone: - t.Fatal("live key frame should not be delivered before cached GOP replay finishes") - case <-time.After(50 * time.Millisecond): - } - - close(sub.release) - <-addDone - <-liveDone - - wantMarkers := []byte{1, 2, 3, 4} - gotMarkers := sub.markers() - if len(gotMarkers) != len(wantMarkers) { - t.Fatalf("markers = %v, want %v", gotMarkers, wantMarkers) - } - for i, want := range wantMarkers { - if got := gotMarkers[i]; got != want { - t.Fatalf("message %d marker = %d, want %d, all=%v", i, got, want, gotMarkers) - } - } -} diff --git a/hook/gop_cache.go b/logic/gop_cache.go similarity index 87% rename from hook/gop_cache.go rename to logic/gop_cache.go index 266e788..c2f0031 100644 --- a/hook/gop_cache.go +++ b/logic/gop_cache.go @@ -1,4 +1,4 @@ -package hook +package logic import ( "bytes" @@ -6,7 +6,6 @@ import ( "github.com/q191201771/lal/pkg/base" ) -// GopCache gop cache type GopCache struct { videoheader *base.RtmpMsg audioheader *base.RtmpMsg @@ -19,11 +18,14 @@ type GopCache struct { last int } -// NewGopCache 创建 gop 缓存 +// gopSize 为 0 时只保存音视频头，不缓存 GOP。 func NewGopCache(gopSize, singleGopMaxFrameNum int) *GopCache { if gopSize < 0 { gopSize = 0 } + if singleGopMaxFrameNum < 0 { + singleGopMaxFrameNum = 0 + } num := gopSize + 1 return &GopCache{ data: make([]Gop, num), @@ -32,7 +34,6 @@ func NewGopCache(gopSize, singleGopMaxFrameNum int) *GopCache { } } -// Feed 写入缓存 func (c *GopCache) Feed(msg base.RtmpMsg) { switch msg.Header.MsgTypeId { case base.RtmpTypeIdMetadata: @@ -95,13 +96,15 @@ func (c *GopCache) feedLastGop(msg base.RtmpMsg) { func (c *GopCache) isGopRingFull() bool { return (c.last+1)%c.gopSize == c.first } + func (c *GopCache) isGopRingEmpty() bool { return c.first == c.last } func (c *GopCache) Clear() { - // c.audioheader = nil - // c.videoheader = nil + for i := range c.data { + c.data[i].release() + } c.last = 0 c.first = 0 } @@ -117,6 +120,7 @@ func (c *GopCache) GetGopDataAt(pos int) []base.RtmpMsg { return c.data[(c.first+pos)%c.gopSize].data } +// clear 保留底层容量用于复用；release 用于码流头变化时释放旧 payload。 type Gop struct { data []base.RtmpMsg } @@ -129,8 +133,16 @@ func (g *Gop) clear() { if len(g.data) == 0 { return } + for i := range g.data { + g.data[i] = base.RtmpMsg{} + } g.data = g.data[:0] } + +func (g *Gop) release() { + g.data = nil +} + func (g *Gop) size() int { return len(g.data) } diff --git a/logic/group.go b/logic/group.go new file mode 100644 index 0000000..38422e4 --- /dev/null +++ b/logic/group.go @@ -0,0 +1,417 @@ +package logic + +import ( + "sync" + "sync/atomic" + "time" + + "github.com/q191201771/lalmax/fmp4/hls" + + "github.com/q191201771/lal/pkg/base" + "github.com/q191201771/naza/pkg/nazalog" +) + +var _ base.ISession = (*subscriberState)(nil) + +const ( + SubscriberProtocolLalmax = "LALMAX" + SubscriberProtocolWHEP = "WHEP" + SubscriberProtocolJessibuca = "JESSIBUCA" + SubscriberProtocolHTTPFMP4 = "HTTP-FMP4" + SubscriberProtocolSRT = "SRT" +) + +type Subscriber interface { + OnMsg(msg base.RtmpMsg) + OnStop() +} + +// 可选接口：订阅者需要区分 GOP 回放和实时帧时实现。 +type ReplaySubscriber interface { + OnReplayStart() + OnReplayStop() +} + +type SubscriberInfo struct { + SubscriberID string + Protocol string + RemoteAddr string +} + +// Group 只维护 lalmax 侧订阅者和回放缓存，推流状态仍以 lal 为准。 +type Group struct { + uniqueKey string + key StreamKey + consumers sync.Map + hlssvr *hls.HlsServer + gopCache *GopCache + gopCacheMux sync.RWMutex + lifecycleMux sync.RWMutex + stopOnce sync.Once + msgMux sync.Mutex + hasVideo bool + closed bool +} + +type subscriberState struct { + key StreamKey + subscriber Subscriber + hasSendVideo bool + replayCache bool + writeMux sync.Mutex + stopped atomic.Bool + + base.StatSession +} + +func (s *subscriberState) AppName() string { + return s.key.AppName +} + +func (s *subscriberState) GetStat() base.StatSession { + return s.StatSession +} + +func (s *subscriberState) IsAlive() (readAlive bool, writeAlive bool) { + return true, true +} + +func (s *subscriberState) RawQuery() string { + return "" +} + +func (s *subscriberState) StreamName() string { + return s.key.StreamName +} + +func (s *subscriberState) UniqueKey() string { + return s.SessionId +} + +func (s *subscriberState) UpdateStat(intervalSec uint32) { +} + +func (s *subscriberState) Url() string { + return s.key.String() +} + +func NewGroup(uniqueKey string, key StreamKey, hlssvr *hls.HlsServer, gopNum, singleGopMaxFrameNum int) *Group { + group := &Group{ + uniqueKey: uniqueKey, + key: key, + hlssvr: hlssvr, + gopCache: NewGopCache(gopNum, singleGopMaxFrameNum), + } + + if group.hlssvr != nil { + group.hlssvr.NewHlsSessionWithAppName(key.AppName, key.StreamName) + } + + nazalog.Infof("create group, uniqueKey:%s, streamKey:%s", uniqueKey, key.String()) + + GetGroupManagerInstance().SetGroup(key, group) + return group +} + +func NewGroupByStreamName(uniqueKey, streamName string, hlssvr *hls.HlsServer, gopNum, singleGopMaxFrameNum int) *Group { + return NewGroup(uniqueKey, StreamKeyFromStreamName(streamName), hlssvr, gopNum, singleGopMaxFrameNum) +} + +func (group *Group) Key() StreamKey { + return group.key +} + +func (group *Group) UniqueKey() string { + return group.uniqueKey +} + +func (group *Group) OnMsg(msg base.RtmpMsg) { + group.lifecycleMux.RLock() + if group.closed { + group.lifecycleMux.RUnlock() + return + } + defer group.lifecycleMux.RUnlock() + + if group.hlssvr != nil { + group.hlssvr.OnMsgWithAppName(group.key.AppName, group.key.StreamName, msg) + } + + group.msgMux.Lock() + hasVideo := group.hasVideo + consumers := make([]*subscriberState, 0) + group.consumers.Range(func(key, value interface{}) bool { + if c, ok := value.(*subscriberState); ok { + consumers = append(consumers, c) + } + return true + }) + + if !group.hasVideo && msg.IsVideoKeyNalu() { + group.hasVideo = true + } + + group.gopCacheMux.Lock() + group.gopCache.Feed(msg) + group.gopCacheMux.Unlock() + group.msgMux.Unlock() + + for _, c := range consumers { + group.handleSubscriberMsg(c, msg, hasVideo) + } +} + +func (group *Group) OnStop() { + group.stopOnce.Do(func() { + group.lifecycleMux.Lock() + group.closed = true + + if group.hlssvr != nil { + group.hlssvr.OnStopWithAppName(group.key.AppName, group.key.StreamName) + } + + consumers := make([]*subscriberState, 0) + group.consumers.Range(func(key, value interface{}) bool { + c, ok := value.(*subscriberState) + if ok { + consumers = append(consumers, c) + } + group.consumers.Delete(key) + return true + }) + group.lifecycleMux.Unlock() + + nazalog.Debugf("OnStop, uniqueKey:%s, streamKey:%s", group.uniqueKey, group.key.String()) + for _, c := range consumers { + c.stopWithNotify() + } + + GetGroupManagerInstance().RemoveGroupIfMatch(group.key, group) + }) +} + +func (group *Group) AddSubscriber(info SubscriberInfo, subscriber Subscriber) { + group.AddSubscriberWithReplay(info, subscriber, true) +} + +func (group *Group) AddSubscriberWithReplay(info SubscriberInfo, subscriber Subscriber, replayCache bool) { + if info.SubscriberID == "" { + nazalog.Warn("AddSubscriber skipped, subscriber id is empty") + return + } + if info.Protocol == "" { + info.Protocol = SubscriberProtocolLalmax + } + + group.lifecycleMux.RLock() + if group.closed { + group.lifecycleMux.RUnlock() + nazalog.Warnf("AddSubscriber skipped, group is closed, streamKey:%s, subscriberId:%s", group.key.String(), info.SubscriberID) + return + } + defer group.lifecycleMux.RUnlock() + + state := &subscriberState{ + key: group.key, + subscriber: subscriber, + replayCache: replayCache, + StatSession: base.StatSession{ + SessionId: info.SubscriberID, + Protocol: info.Protocol, + BaseType: base.SessionBaseTypeSubStr, + RemoteAddr: info.RemoteAddr, + StartTime: time.Now().Format(time.DateTime), + }, + } + + nazalog.Infof("AddSubscriber, streamKey:%s, subscriberId:%s, protocol:%s", group.key.String(), info.SubscriberID, info.Protocol) + if replayCache { + // 保证该订阅者先收到缓存 GOP，再收到实时帧。 + state.writeMux.Lock() + } + var replayMsgs []base.RtmpMsg + + group.msgMux.Lock() + if _, loaded := group.consumers.Load(info.SubscriberID); loaded { + group.msgMux.Unlock() + if replayCache { + state.writeMux.Unlock() + } + nazalog.Warnf("AddSubscriber skipped, subscriber already exists, streamKey:%s, subscriberId:%s", group.key.String(), info.SubscriberID) + return + } + group.consumers.Store(info.SubscriberID, state) + if replayCache { + replayMsgs = group.getGopReplayMessages() + } + group.msgMux.Unlock() + + if replayCache { + group.replayGopMessagesLocked(state, replayMsgs) + state.writeMux.Unlock() + } +} + +func (group *Group) AddConsumer(consumerID string, subscriber Subscriber) { + group.AddSubscriber(SubscriberInfo{SubscriberID: consumerID}, subscriber) +} + +func (group *Group) AddConsumerWithReplay(consumerID string, subscriber Subscriber, replayCache bool) { + group.AddSubscriberWithReplay(SubscriberInfo{SubscriberID: consumerID}, subscriber, replayCache) +} + +func (group *Group) StatSubscribers() []base.StatSub { + out := make([]base.StatSub, 0, 10) + group.consumers.Range(func(key, value any) bool { + v, ok := value.(*subscriberState) + if ok { + out = append(out, base.Session2StatSub(v)) + } + return true + }) + return out +} + +func (group *Group) GetAllConsumer() []base.StatSub { + return group.StatSubscribers() +} + +func (group *Group) RemoveSubscriber(subscriberID string) { + value, ok := group.consumers.LoadAndDelete(subscriberID) + if ok { + nazalog.Infof("RemoveSubscriber, streamKey:%s, subscriberId:%s", group.key.String(), subscriberID) + if c, ok := value.(*subscriberState); ok { + c.stopWithoutNotify() + } + } +} + +func (group *Group) RemoveConsumer(consumerID string) { + group.RemoveSubscriber(consumerID) +} + +func (group *Group) GetVideoSeqHeaderMsg() *base.RtmpMsg { + group.gopCacheMux.RLock() + defer group.gopCacheMux.RUnlock() + if group.gopCache.videoheader == nil { + return nil + } + m := group.gopCache.videoheader.Clone() + return &m +} + +func (group *Group) GetAudioSeqHeaderMsg() *base.RtmpMsg { + group.gopCacheMux.RLock() + defer group.gopCacheMux.RUnlock() + if group.gopCache.audioheader == nil { + return nil + } + m := group.gopCache.audioheader.Clone() + return &m +} + +func (group *Group) handleSubscriberMsg(c *subscriberState, msg base.RtmpMsg, hasVideo bool) { + if c == nil { + return + } + + c.writeMux.Lock() + defer c.writeMux.Unlock() + + if c.stopped.Load() || c.subscriber == nil { + return + } + + if msg.Header.MsgTypeId == base.RtmpTypeIdVideo { + if !c.hasSendVideo { + if !msg.IsVideoKeyNalu() { + return + } + if v := group.GetVideoSeqHeaderMsg(); v != nil { + c.subscriber.OnMsg(*v) + } + if v := group.GetAudioSeqHeaderMsg(); v != nil && v.IsAacSeqHeader() { + c.subscriber.OnMsg(*v) + } + c.hasSendVideo = true + } + + c.subscriber.OnMsg(msg) + } else if msg.Header.MsgTypeId == base.RtmpTypeIdAudio { + if !hasVideo || c.hasSendVideo { + c.subscriber.OnMsg(msg) + } + } +} + +func (group *Group) replayGopMessagesLocked(c *subscriberState, msgs []base.RtmpMsg) { + if c == nil || c.subscriber == nil || c.stopped.Load() || c.hasSendVideo || !c.replayCache { + return + } + + if len(msgs) == 0 { + return + } + + if replaySubscriber, ok := c.subscriber.(ReplaySubscriber); ok { + replaySubscriber.OnReplayStart() + defer replaySubscriber.OnReplayStop() + } + + for _, msg := range msgs { + c.subscriber.OnMsg(msg) + } + c.hasSendVideo = true +} + +func (s *subscriberState) stopWithNotify() { + if s == nil { + return + } + + s.writeMux.Lock() + defer s.writeMux.Unlock() + + if s.stopped.Swap(true) { + return + } + if s.subscriber != nil { + s.subscriber.OnStop() + s.subscriber = nil + } +} + +func (s *subscriberState) stopWithoutNotify() { + if s == nil { + return + } + + // 不能在这里获取 writeMux：部分订阅者会在 OnMsg 调用栈内主动移除自己。 + // 只标记停止，避免后续投递；订阅者对象随 state 一起释放。 + s.stopped.Store(true) +} + +func (group *Group) getGopReplayMessages() []base.RtmpMsg { + group.gopCacheMux.RLock() + defer group.gopCacheMux.RUnlock() + + gopCount := group.gopCache.GetGopCount() + if gopCount == 0 { + return nil + } + + msgs := make([]base.RtmpMsg, 0, gopCount) + if v := group.gopCache.videoheader; v != nil { + msgs = append(msgs, v.Clone()) + } + if v := group.gopCache.audioheader; v != nil && v.IsAacSeqHeader() { + msgs = append(msgs, v.Clone()) + } + for i := 0; i < gopCount; i++ { + for _, item := range group.gopCache.GetGopDataAt(i) { + msgs = append(msgs, item.Clone()) + } + } + + return msgs +} diff --git a/logic/group_manager.go b/logic/group_manager.go new file mode 100644 index 0000000..2212105 --- /dev/null +++ b/logic/group_manager.go @@ -0,0 +1,216 @@ +package logic + +import ( + "sync" + + "github.com/q191201771/naza/pkg/nazalog" +) + +type IGroupManager interface { + SetGroup(key StreamKey, group *Group) + RemoveGroup(key StreamKey) + RemoveGroupIfMatch(key StreamKey, group *Group) + GetGroup(key StreamKey) (bool, *Group) + Iterate(onIterateGroup func(key StreamKey, group *Group) bool) + Len() int +} + +type ComplexGroupManager struct { + mutex sync.RWMutex + + onlyStreamNameGroups map[string]*Group + appNameStreamNameGroups map[string]map[string]*Group +} + +// 同时支持新路径 app/stream 和旧路径 stream 的查找方式。 +func NewComplexGroupManager() *ComplexGroupManager { + return &ComplexGroupManager{ + onlyStreamNameGroups: make(map[string]*Group), + appNameStreamNameGroups: make(map[string]map[string]*Group), + } +} + +var ( + defaultGroupManager *ComplexGroupManager + groupManagerOnce sync.Once +) + +func GetGroupManagerInstance() *ComplexGroupManager { + groupManagerOnce.Do(func() { + defaultGroupManager = NewComplexGroupManager() + }) + return defaultGroupManager +} + +func (m *ComplexGroupManager) SetGroup(key StreamKey, group *Group) { + if m == nil || !key.Valid() || group == nil { + return + } + + nazalog.Info("SetGroup, streamKey:", key.String()) + + m.mutex.Lock() + defer m.mutex.Unlock() + + if key.AppName == "" { + m.onlyStreamNameGroups[key.StreamName] = group + return + } + + groups, ok := m.appNameStreamNameGroups[key.AppName] + if !ok { + groups = make(map[string]*Group) + m.appNameStreamNameGroups[key.AppName] = groups + } + groups[key.StreamName] = group +} + +func (m *ComplexGroupManager) SetGroupByStreamName(streamName string, group *Group) { + m.SetGroup(StreamKeyFromStreamName(streamName), group) +} + +func (m *ComplexGroupManager) RemoveGroup(key StreamKey) { + m.removeGroup(key, nil, false) +} + +// 避免旧流晚到的 OnStop 或遍历删除误删同 key 的新流。 +func (m *ComplexGroupManager) RemoveGroupIfMatch(key StreamKey, group *Group) { + m.removeGroup(key, group, true) +} + +func (m *ComplexGroupManager) removeGroup(key StreamKey, group *Group, shouldMatch bool) { + if m == nil || !key.Valid() { + return + } + + nazalog.Info("RemoveGroup, streamKey:", key.String()) + + m.mutex.Lock() + defer m.mutex.Unlock() + + if key.AppName == "" { + if shouldMatch && m.onlyStreamNameGroups[key.StreamName] != group { + return + } + delete(m.onlyStreamNameGroups, key.StreamName) + return + } + + deleted := false + if groups, ok := m.appNameStreamNameGroups[key.AppName]; ok { + if current, ok := groups[key.StreamName]; ok { + if shouldMatch && current != group { + return + } + delete(groups, key.StreamName) + deleted = true + } + if len(groups) == 0 { + delete(m.appNameStreamNameGroups, key.AppName) + } + } + + if !deleted { + if shouldMatch && m.onlyStreamNameGroups[key.StreamName] != group { + return + } + delete(m.onlyStreamNameGroups, key.StreamName) + } +} + +func (m *ComplexGroupManager) RemoveGroupByStreamName(streamName string) { + m.RemoveGroup(StreamKeyFromStreamName(streamName)) +} + +func (m *ComplexGroupManager) GetGroup(key StreamKey) (bool, *Group) { + if m == nil || !key.Valid() { + return false, nil + } + + m.mutex.RLock() + defer m.mutex.RUnlock() + + if key.AppName == "" { + if group, ok := m.onlyStreamNameGroups[key.StreamName]; ok { + return true, group + } + return m.getGroupByOnlyStreamName(key.StreamName) + } + + if groups, ok := m.appNameStreamNameGroups[key.AppName]; ok { + if group, ok := groups[key.StreamName]; ok { + return true, group + } + } + + if group, ok := m.onlyStreamNameGroups[key.StreamName]; ok { + return true, group + } + + return false, nil +} + +func (m *ComplexGroupManager) GetGroupByStreamName(streamName string) (bool, *Group) { + return m.GetGroup(StreamKeyFromStreamName(streamName)) +} + +// streamName 单独查找只在匹配唯一 appName 时成功，避免跨 app 串流。 +func (m *ComplexGroupManager) getGroupByOnlyStreamName(streamName string) (bool, *Group) { + var found *Group + matchCount := 0 + for _, groups := range m.appNameStreamNameGroups { + if group, ok := groups[streamName]; ok { + found = group + matchCount++ + if matchCount > 1 { + nazalog.Warn("streamName matched multiple appName groups, streamName:", streamName) + return false, nil + } + } + } + return matchCount == 1, found +} + +func (m *ComplexGroupManager) Iterate(onIterateGroup func(key StreamKey, group *Group) bool) { + if m == nil || onIterateGroup == nil { + return + } + + type entry struct { + key StreamKey + group *Group + } + entries := make([]entry, 0, m.Len()) + + m.mutex.RLock() + for streamName, group := range m.onlyStreamNameGroups { + entries = append(entries, entry{key: StreamKeyFromStreamName(streamName), group: group}) + } + for appName, groups := range m.appNameStreamNameGroups { + for streamName, group := range groups { + entries = append(entries, entry{key: NewStreamKey(appName, streamName), group: group}) + } + } + m.mutex.RUnlock() + + for _, item := range entries { + if !onIterateGroup(item.key, item.group) { + m.RemoveGroupIfMatch(item.key, item.group) + } + } +} + +func (m *ComplexGroupManager) Len() int { + if m == nil { + return 0 + } + + m.mutex.RLock() + defer m.mutex.RUnlock() + + count := len(m.onlyStreamNameGroups) + for _, groups := range m.appNameStreamNameGroups { + count += len(groups) + } + return count +} diff --git a/logic/group_test.go b/logic/group_test.go new file mode 100644 index 0000000..b8e70fd --- /dev/null +++ b/logic/group_test.go @@ -0,0 +1,472 @@ +package logic + +import ( + "sync" + "testing" + "time" + + "github.com/q191201771/lal/pkg/base" +) + +type recordSubscriber struct { + mu sync.Mutex + msgs []base.RtmpMsg + stopCount int +} + +func (s *recordSubscriber) OnMsg(msg base.RtmpMsg) { + s.mu.Lock() + defer s.mu.Unlock() + s.msgs = append(s.msgs, msg.Clone()) +} + +func (s *recordSubscriber) OnStop() { + s.mu.Lock() + defer s.mu.Unlock() + s.stopCount++ +} + +func (s *recordSubscriber) len() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.msgs) +} + +func (s *recordSubscriber) markerAt(idx int) byte { + s.mu.Lock() + defer s.mu.Unlock() + return payloadMarker(s.msgs[idx]) +} + +func (s *recordSubscriber) stopCountValue() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.stopCount +} + +type blockingSubscriber struct { + mu sync.Mutex + msgs []base.RtmpMsg + blocked chan struct{} + release chan struct{} + replaying bool + blockOnce sync.Once +} + +func newBlockingSubscriber() *blockingSubscriber { + return &blockingSubscriber{ + blocked: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (s *blockingSubscriber) OnMsg(msg base.RtmpMsg) { + s.mu.Lock() + s.msgs = append(s.msgs, msg.Clone()) + shouldBlock := s.replaying + s.mu.Unlock() + + if shouldBlock { + s.blockOnce.Do(func() { + close(s.blocked) + <-s.release + }) + } +} + +func (s *blockingSubscriber) OnStop() {} + +func (s *blockingSubscriber) OnReplayStart() { + s.mu.Lock() + s.replaying = true + s.mu.Unlock() +} + +func (s *blockingSubscriber) OnReplayStop() { + s.mu.Lock() + s.replaying = false + s.mu.Unlock() +} + +func (s *blockingSubscriber) markers() []byte { + s.mu.Lock() + defer s.mu.Unlock() + + out := make([]byte, 0, len(s.msgs)) + for _, msg := range s.msgs { + out = append(out, payloadMarker(msg)) + } + return out +} + +func videoSeqHeader(marker byte) base.RtmpMsg { + return base.RtmpMsg{ + Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdVideo}, + Payload: []byte{ + base.RtmpAvcKeyFrame, + base.RtmpAvcPacketTypeSeqHeader, + 0, 0, 0, + marker, + }, + } +} + +func videoKeyNalu(marker byte) base.RtmpMsg { + return base.RtmpMsg{ + Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdVideo}, + Payload: []byte{ + base.RtmpAvcKeyFrame, + base.RtmpAvcPacketTypeNalu, + 0, 0, 0, + marker, + }, + } +} + +func videoInterNalu(marker byte) base.RtmpMsg { + return base.RtmpMsg{ + Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdVideo}, + Payload: []byte{ + base.RtmpAvcInterFrame, + base.RtmpAvcPacketTypeNalu, + 0, 0, 0, + marker, + }, + } +} + +func aacSeqHeader(marker byte) base.RtmpMsg { + return base.RtmpMsg{ + Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdAudio}, + Payload: []byte{ + base.RtmpSoundFormatAac << 4, + base.RtmpAacPacketTypeSeqHeader, + marker, + }, + } +} + +func aacRaw(marker byte) base.RtmpMsg { + return base.RtmpMsg{ + Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdAudio}, + Payload: []byte{ + base.RtmpSoundFormatAac << 4, + base.RtmpAacPacketTypeRaw, + marker, + }, + } +} + +func g711aAudio(marker byte) base.RtmpMsg { + return base.RtmpMsg{ + Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdAudio}, + Payload: []byte{base.RtmpSoundFormatG711A<<4 | marker}, + } +} + +func payloadMarker(msg base.RtmpMsg) byte { + return msg.Payload[len(msg.Payload)-1] +} + +func TestAddConsumerReplaysCachedGopImmediately(t *testing.T) { + group := NewGroupByStreamName("test-replay", "test-replay", nil, 1, 0) + defer GetGroupManagerInstance().RemoveGroupByStreamName("test-replay") + + group.OnMsg(videoSeqHeader(1)) + group.OnMsg(aacSeqHeader(2)) + group.OnMsg(videoKeyNalu(3)) + group.OnMsg(aacRaw(4)) + group.OnMsg(videoInterNalu(5)) + + sub := &recordSubscriber{} + group.AddConsumer("consumer", sub) + + if sub.len() != 5 { + t.Fatalf("expected 5 replay messages, got %d", sub.len()) + } + + wantMarkers := []byte{1, 2, 3, 4, 5} + for i, want := range wantMarkers { + if got := sub.markerAt(i); got != want { + t.Fatalf("message %d marker = %d, want %d", i, got, want) + } + } +} + +func TestVideoSeqHeaderChangeClearsStaleGop(t *testing.T) { + group := NewGroupByStreamName("test-clear", "test-clear", nil, 1, 0) + defer GetGroupManagerInstance().RemoveGroupByStreamName("test-clear") + + group.OnMsg(videoSeqHeader(1)) + group.OnMsg(videoKeyNalu(2)) + group.OnMsg(videoInterNalu(3)) + group.OnMsg(videoSeqHeader(4)) + + sub := &recordSubscriber{} + group.AddConsumer("consumer", sub) + if sub.len() != 0 { + t.Fatalf("expected no stale GOP replay after sequence header change, got %d messages", sub.len()) + } + + group.OnMsg(videoKeyNalu(5)) + if sub.len() != 2 { + t.Fatalf("expected new header and current key frame, got %d messages", sub.len()) + } + if got := sub.markerAt(0); got != 4 { + t.Fatalf("header marker = %d, want 4", got) + } + if got := sub.markerAt(1); got != 5 { + t.Fatalf("key frame marker = %d, want 5", got) + } +} + +func TestNonAacAudioIsNotReplayedAsHeader(t *testing.T) { + group := NewGroupByStreamName("test-g711", "test-g711", nil, 1, 0) + defer GetGroupManagerInstance().RemoveGroupByStreamName("test-g711") + + group.OnMsg(videoSeqHeader(1)) + group.OnMsg(videoKeyNalu(2)) + group.OnMsg(g711aAudio(3)) + + sub := &recordSubscriber{} + group.AddConsumer("consumer", sub) + + if sub.len() != 3 { + t.Fatalf("expected video header, key frame and one G711 packet, got %d messages", sub.len()) + } + + wantMarkers := []byte{1, 2, base.RtmpSoundFormatG711A<<4 | 3} + for i, want := range wantMarkers { + if got := sub.markerAt(i); got != want { + t.Fatalf("message %d marker = %d, want %d", i, got, want) + } + } +} + +func TestAddConsumerWithReplayDisabledDoesNotReplayCachedGop(t *testing.T) { + group := NewGroupByStreamName("test-no-replay", "test-no-replay", nil, 1, 0) + defer GetGroupManagerInstance().RemoveGroupByStreamName("test-no-replay") + + group.OnMsg(videoSeqHeader(1)) + group.OnMsg(videoKeyNalu(2)) + group.OnMsg(videoInterNalu(3)) + + sub := &recordSubscriber{} + group.AddConsumerWithReplay("consumer", sub, false) + + if sub.len() != 0 { + t.Fatalf("expected no cached messages when replay is disabled, got %d messages", sub.len()) + } + + group.OnMsg(videoInterNalu(4)) + if sub.len() != 0 { + t.Fatalf("expected to wait for next key frame, got %d messages", sub.len()) + } + + group.OnMsg(videoKeyNalu(5)) + if sub.len() != 2 { + t.Fatalf("expected header and current key frame, got %d messages", sub.len()) + } + if got := sub.markerAt(0); got != 1 { + t.Fatalf("header marker = %d, want 1", got) + } + if got := sub.markerAt(1); got != 5 { + t.Fatalf("key frame marker = %d, want 5", got) + } +} + +func TestAddConsumerReplayDoesNotInterleaveWithLiveKeyFrame(t *testing.T) { + group := NewGroupByStreamName("test-replay-order", "test-replay-order", nil, 1, 0) + defer GetGroupManagerInstance().RemoveGroupByStreamName("test-replay-order") + + group.OnMsg(videoSeqHeader(1)) + group.OnMsg(videoKeyNalu(2)) + group.OnMsg(videoInterNalu(3)) + + sub := newBlockingSubscriber() + addDone := make(chan struct{}) + go func() { + group.AddConsumer("consumer", sub) + close(addDone) + }() + + <-sub.blocked + + liveDone := make(chan struct{}) + go func() { + group.OnMsg(videoKeyNalu(4)) + close(liveDone) + }() + + select { + case <-liveDone: + t.Fatal("live key frame should not be delivered before cached GOP replay finishes") + case <-time.After(50 * time.Millisecond): + } + + close(sub.release) + <-addDone + <-liveDone + + wantMarkers := []byte{1, 2, 3, 4} + gotMarkers := sub.markers() + if len(gotMarkers) != len(wantMarkers) { + t.Fatalf("markers = %v, want %v", gotMarkers, wantMarkers) + } + for i, want := range wantMarkers { + if got := gotMarkers[i]; got != want { + t.Fatalf("message %d marker = %d, want %d, all=%v", i, got, want, gotMarkers) + } + } +} + +func TestGroupManagerSupportsAppNameAndStreamName(t *testing.T) { + manager := NewComplexGroupManager() + group := &Group{key: NewStreamKey("live", "camera")} + + manager.SetGroup(group.Key(), group) + + ok, got := manager.GetGroup(NewStreamKey("live", "camera")) + if !ok || got != group { + t.Fatal("expected exact appName and streamName lookup") + } + + ok, got = manager.GetGroup(StreamKeyFromStreamName("camera")) + if !ok || got != group { + t.Fatal("expected streamName-only lookup to find the unique appName group") + } +} + +func TestGroupManagerStreamNameFallbackRejectsAmbiguousAppName(t *testing.T) { + manager := NewComplexGroupManager() + manager.SetGroup(NewStreamKey("app1", "camera"), &Group{key: NewStreamKey("app1", "camera")}) + manager.SetGroup(NewStreamKey("app2", "camera"), &Group{key: NewStreamKey("app2", "camera")}) + + ok, got := manager.GetGroup(StreamKeyFromStreamName("camera")) + if ok || got != nil { + t.Fatal("expected ambiguous streamName-only lookup to fail") + } +} + +func TestGroupManagerRemoveGroupIfMatchDoesNotRemoveNewGroup(t *testing.T) { + manager := NewComplexGroupManager() + key := StreamKeyFromStreamName("camera") + oldGroup := &Group{key: key} + newGroup := &Group{key: key} + + manager.SetGroup(key, oldGroup) + manager.SetGroup(key, newGroup) + manager.RemoveGroupIfMatch(key, oldGroup) + + ok, got := manager.GetGroup(key) + if !ok || got != newGroup { + t.Fatal("old group stop should not remove new group") + } +} + +func TestGroupManagerIterateRemoveDoesNotRemoveReplacement(t *testing.T) { + manager := NewComplexGroupManager() + key := StreamKeyFromStreamName("camera") + oldGroup := &Group{key: key} + newGroup := &Group{key: key} + + manager.SetGroup(key, oldGroup) + manager.Iterate(func(iterKey StreamKey, group *Group) bool { + if iterKey != key || group != oldGroup { + t.Fatalf("unexpected iterate entry, key=%v group=%p", iterKey, group) + } + manager.SetGroup(key, newGroup) + return false + }) + + ok, got := manager.GetGroup(key) + if !ok || got != newGroup { + t.Fatal("iterate removal should not remove a replacement group") + } +} + +func TestGopCacheClearReleasesStaleGopPayloads(t *testing.T) { + cache := NewGopCache(1, 0) + + cache.Feed(videoKeyNalu(1)) + cache.Feed(videoInterNalu(2)) + cache.Clear() + + if cache.GetGopCount() != 0 { + t.Fatalf("gop count = %d, want 0", cache.GetGopCount()) + } + for i, gop := range cache.data { + if gop.data != nil { + t.Fatalf("gop %d data was not released", i) + } + } +} + +func TestGopCacheNegativeFrameLimitMeansUnlimited(t *testing.T) { + cache := NewGopCache(1, -1) + + cache.Feed(videoKeyNalu(1)) + cache.Feed(videoInterNalu(2)) + + msgs := cache.GetGopDataAt(0) + if len(msgs) != 2 { + t.Fatalf("cached messages = %d, want 2", len(msgs)) + } +} + +func TestOnStopIsIdempotentAndClosesSubscribers(t *testing.T) { + group := NewGroupByStreamName("test-stop", "test-stop", nil, 1, 0) + defer GetGroupManagerInstance().RemoveGroupByStreamName("test-stop") + + sub := &recordSubscriber{} + group.AddConsumer("consumer", sub) + + group.OnStop() + group.OnStop() + + if sub.stopCountValue() != 1 { + t.Fatalf("stop count = %d, want 1", sub.stopCountValue()) + } + + group.OnMsg(videoKeyNalu(1)) + if sub.len() != 0 { + t.Fatalf("expected no messages after stop, got %d", sub.len()) + } +} + +func TestAddSubscriberAfterStopIsIgnored(t *testing.T) { + group := NewGroupByStreamName("test-add-after-stop", "test-add-after-stop", nil, 1, 0) + defer GetGroupManagerInstance().RemoveGroupByStreamName("test-add-after-stop") + + group.OnStop() + + sub := &recordSubscriber{} + group.AddConsumer("consumer", sub) + group.OnMsg(videoKeyNalu(1)) + + if sub.len() != 0 { + t.Fatalf("expected no messages after adding to stopped group, got %d", sub.len()) + } + if len(group.StatSubscribers()) != 0 { + t.Fatalf("expected no subscribers after adding to stopped group, got %d", len(group.StatSubscribers())) + } +} + +func TestDuplicateSubscriberIDIsIgnored(t *testing.T) { + group := NewGroupByStreamName("test-duplicate", "test-duplicate", nil, 1, 0) + defer GetGroupManagerInstance().RemoveGroupByStreamName("test-duplicate") + + first := &recordSubscriber{} + second := &recordSubscriber{} + group.AddConsumer("consumer", first) + group.AddConsumer("consumer", second) + + group.OnMsg(videoKeyNalu(1)) + + if first.len() != 1 { + t.Fatalf("first subscriber messages = %d, want 1", first.len()) + } + if second.len() != 0 { + t.Fatalf("duplicate subscriber messages = %d, want 0", second.len()) + } +} diff --git a/logic/stream_key.go b/logic/stream_key.go new file mode 100644 index 0000000..2e4dc3a --- /dev/null +++ b/logic/stream_key.go @@ -0,0 +1,29 @@ +package logic + +type StreamKey struct { + // AppName 为空表示兼容历史的 streamName 单键查找。 + AppName string + StreamName string +} + +func NewStreamKey(appName, streamName string) StreamKey { + return StreamKey{ + AppName: appName, + StreamName: streamName, + } +} + +func StreamKeyFromStreamName(streamName string) StreamKey { + return NewStreamKey("", streamName) +} + +func (key StreamKey) Valid() bool { + return key.StreamName != "" +} + +func (key StreamKey) String() string { + if key.AppName == "" { + return key.StreamName + } + return key.AppName + "/" + key.StreamName +} diff --git a/main.go b/main.go index e6067cc..eb86959 100644 --- a/main.go +++ b/main.go @@ -12,7 +12,7 @@ import ( "github.com/q191201771/lal/pkg/base" - config "github.com/q191201771/lalmax/conf" + config "github.com/q191201771/lalmax/config" "github.com/q191201771/naza/pkg/bininfo" ) diff --git a/rtc/jessibucasession.go b/rtc/jessibucasession.go index 21de478..0e959f5 100644 --- a/rtc/jessibucasession.go +++ b/rtc/jessibucasession.go @@ -11,13 +11,13 @@ import ( "github.com/q191201771/lal/pkg/httpflv" "github.com/q191201771/lal/pkg/logic" "github.com/q191201771/lal/pkg/remux" - "github.com/q191201771/lalmax/hook" + maxlogic "github.com/q191201771/lalmax/logic" "github.com/q191201771/naza/pkg/nazalog" "github.com/smallnest/chanx" ) type jessibucaSession struct { - hooks *hook.HookSession + group *maxlogic.Group pc *peerConnection subscriberId string lalServer logic.ILalServer @@ -34,17 +34,17 @@ type jessibucaSession struct { stopOne sync.Once } -func NewJessibucaSession(streamid string, writeChanSize int, pc *peerConnection, lalServer logic.ILalServer) *jessibucaSession { - ok, session := hook.GetHookSessionManagerInstance().GetHookSession(streamid) +func NewJessibucaSession(appName, streamid string, writeChanSize int, pc *peerConnection, lalServer logic.ILalServer) *jessibucaSession { + ok, group := maxlogic.GetGroupManagerInstance().GetGroup(maxlogic.NewStreamKey(appName, streamid)) if !ok { - nazalog.Error("not found streamid:", streamid) + nazalog.Errorf("not found stream, appName:%s, streamid:%s", appName, streamid) return nil } u, _ := uuid.NewV4() ctx, cancel := context.WithCancel(context.Background()) return &jessibucaSession{ - hooks: session, + group: group, pc: pc, lalServer: lalServer, subscriberId: u.String(), @@ -95,9 +95,12 @@ func (conn *jessibucaSession) GetAnswerSDP(offer string) (sdp string) { } func (conn *jessibucaSession) Run() { - ok, _ := hook.GetHookSessionManagerInstance().GetHookSession(conn.streamId) + ok, _ := maxlogic.GetGroupManagerInstance().GetGroup(conn.group.Key()) if ok { - conn.hooks.AddConsumer(conn.subscriberId, conn) + conn.group.AddSubscriber(maxlogic.SubscriberInfo{ + SubscriberID: conn.subscriberId, + Protocol: maxlogic.SubscriberProtocolJessibuca, + }, conn) conn.pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { nazalog.Info("peer connection state: ", state.String()) @@ -121,7 +124,7 @@ func (conn *jessibucaSession) Run() { defer func() { nazalog.Info("RemoveConsumer, connid:", conn.subscriberId) - conn.hooks.RemoveConsumer(conn.subscriberId) + conn.group.RemoveSubscriber(conn.subscriberId) conn.DC.Close() conn.pc.Close() conn.DC = nil diff --git a/rtc/server.go b/rtc/server.go index dbfaec3..5c91dfc 100644 --- a/rtc/server.go +++ b/rtc/server.go @@ -5,7 +5,7 @@ import ( "net" "net/http" - config "github.com/q191201771/lalmax/conf" + config "github.com/q191201771/lalmax/config" "github.com/gin-gonic/gin" "github.com/pion/ice/v2" @@ -121,6 +121,7 @@ func (s *RtcServer) HandleJessibuca(c *gin.Context) { c.Status(http.StatusMethodNotAllowed) return } + appName := c.Query("app_name") body, err := c.GetRawData() if err != nil { @@ -141,7 +142,7 @@ func (s *RtcServer) HandleJessibuca(c *gin.Context) { return } - jessibucaSession := NewJessibucaSession(streamid, s.config.WriteChanSize, pc, s.lalServer) + jessibucaSession := NewJessibucaSession(appName, streamid, s.config.WriteChanSize, pc, s.lalServer) if jessibucaSession == nil { c.Status(http.StatusInternalServerError) pc.Close() @@ -167,6 +168,7 @@ func (s *RtcServer) HandleWHEP(c *gin.Context) { c.Status(http.StatusMethodNotAllowed) return } + appName := c.Request.URL.Query().Get("app_name") body, err := c.GetRawData() if err != nil { @@ -187,7 +189,7 @@ func (s *RtcServer) HandleWHEP(c *gin.Context) { return } - whepsession := NewWhepSession(streamid, s.config.WriteChanSize, pc, s.lalServer) + whepsession := NewWhepSession(appName, streamid, s.config.WriteChanSize, pc, s.lalServer) if whepsession == nil { c.Status(http.StatusInternalServerError) pc.Close() diff --git a/rtc/whepsession.go b/rtc/whepsession.go index 3b1917c..a6e5dbb 100644 --- a/rtc/whepsession.go +++ b/rtc/whepsession.go @@ -6,7 +6,7 @@ import ( "sync" "time" - "github.com/q191201771/lalmax/hook" + maxlogic "github.com/q191201771/lalmax/logic" "github.com/smallnest/chanx" "github.com/gofrs/uuid" @@ -21,7 +21,7 @@ import ( const whepMaxReplayPaceDelay = 5 * time.Millisecond type whepSession struct { - hooks *hook.HookSession + group *maxlogic.Group pc *peerConnection subscriberId string lalServer logic.ILalServer @@ -39,16 +39,16 @@ type whepSession struct { replayingCache bool } -func NewWhepSession(streamid string, writeChanSize int, pc *peerConnection, lalServer logic.ILalServer) *whepSession { - ok, session := hook.GetHookSessionManagerInstance().GetHookSession(streamid) +func NewWhepSession(appName, streamid string, writeChanSize int, pc *peerConnection, lalServer logic.ILalServer) *whepSession { + ok, group := maxlogic.GetGroupManagerInstance().GetGroup(maxlogic.NewStreamKey(appName, streamid)) if !ok { - nazalog.Error("not found streamid:", streamid) + nazalog.Errorf("not found stream, appName:%s, streamid:%s", appName, streamid) return nil } u, _ := uuid.NewV4() return &whepSession{ - hooks: session, + group: group, pc: pc, lalServer: lalServer, subscriberId: u.String(), @@ -61,7 +61,7 @@ func NewWhepSession(streamid string, writeChanSize int, pc *peerConnection, lalS func (conn *whepSession) GetAnswerSDP(offer string) (sdp string) { var err error - videoHeader := conn.hooks.GetVideoSeqHeaderMsg() + videoHeader := conn.group.GetVideoSeqHeaderMsg() if videoHeader != nil { if videoHeader.IsAvcKeySeqHeader() { conn.videoTrack, err = webrtc.NewTrackLocalStaticRTP(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, "video", "lalmax") @@ -94,7 +94,7 @@ func (conn *whepSession) GetAnswerSDP(offer string) (sdp string) { } } - audioHeader := conn.hooks.GetAudioSeqHeaderMsg() + audioHeader := conn.group.GetAudioSeqHeaderMsg() if audioHeader != nil { var mimeType string audioId := audioHeader.AudioCodecId() @@ -186,11 +186,14 @@ func (conn *whepSession) Run() { for { select { case <-conn.connectedChan: - conn.hooks.AddConsumer(conn.subscriberId, conn) + conn.group.AddSubscriber(maxlogic.SubscriberInfo{ + SubscriberID: conn.subscriberId, + Protocol: maxlogic.SubscriberProtocolWHEP, + }, conn) goto connected case <-conn.closeChan: nazalog.Info("RemoveConsumer, connid:", conn.subscriberId) - conn.hooks.RemoveConsumer(conn.subscriberId) + conn.group.RemoveSubscriber(conn.subscriberId) return } } @@ -215,7 +218,7 @@ connected: } case <-conn.closeChan: nazalog.Info("RemoveConsumer, connid:", conn.subscriberId) - conn.hooks.RemoveConsumer(conn.subscriberId) + conn.group.RemoveSubscriber(conn.subscriberId) return } } diff --git a/server/http_notify.go b/server/http_notify.go index 73b587e..6582e70 100644 --- a/server/http_notify.go +++ b/server/http_notify.go @@ -12,9 +12,9 @@ import ( "net/http" "time" - "github.com/q191201771/lalmax/hook" + maxlogic "github.com/q191201771/lalmax/logic" - config "github.com/q191201771/lalmax/conf" + config "github.com/q191201771/lalmax/config" "github.com/q191201771/lal/pkg/base" "github.com/q191201771/naza/pkg/nazahttp" @@ -73,9 +73,9 @@ func (h *HttpNotify) NotifyServerStart(info base.LalInfo) { func (h *HttpNotify) NotifyUpdate(info base.UpdateInfo) { info.ServerId = h.serverId for i, v := range info.Groups { - exist, session := hook.GetHookSessionManagerInstance().GetHookSession(v.StreamName) + exist, session := maxlogic.GetGroupManagerInstance().GetGroup(maxlogic.NewStreamKey(v.AppName, v.StreamName)) if exist { - info.Groups[i].StatSubs = append(info.Groups[i].StatSubs, session.GetAllConsumer()...) + info.Groups[i].StatSubs = append(info.Groups[i].StatSubs, session.StatSubscribers()...) } } h.notifyUpdateAsyncPost(h.cfg.OnUpdate, info) diff --git a/server/router.go b/server/router.go index 7be076b..954182e 100644 --- a/server/router.go +++ b/server/router.go @@ -5,7 +5,7 @@ import ( "io" "net/http" - "github.com/q191201771/lalmax/hook" + maxlogic "github.com/q191201771/lalmax/logic" "github.com/q191201771/lalmax/gb28181" @@ -133,6 +133,7 @@ func (s *LalMaxServer) statGroupHandler(c *gin.Context) { c.JSON(http.StatusOK, v) return } + appName := c.Query("app_name") v.Data = s.lalsvr.StatGroup(streamName) if v.Data == nil { v.ErrorCode = base.ErrorCodeGroupNotFound @@ -140,9 +141,9 @@ func (s *LalMaxServer) statGroupHandler(c *gin.Context) { c.JSON(http.StatusOK, v) return } - exist, session := hook.GetHookSessionManagerInstance().GetHookSession(streamName) + exist, session := maxlogic.GetGroupManagerInstance().GetGroup(maxlogic.NewStreamKey(appName, streamName)) if exist { - v.Data.StatSubs = append(v.Data.StatSubs, session.GetAllConsumer()...) + v.Data.StatSubs = append(v.Data.StatSubs, session.StatSubscribers()...) } v.ErrorCode = base.ErrorCodeSucc v.Desp = base.DespSucc @@ -155,9 +156,9 @@ func (s *LalMaxServer) statAllGroupHandler(c *gin.Context) { out.Desp = base.DespSucc groups := s.lalsvr.StatAllGroup() for i, group := range groups { - exist, session := hook.GetHookSessionManagerInstance().GetHookSession(group.StreamName) + exist, session := maxlogic.GetGroupManagerInstance().GetGroup(maxlogic.NewStreamKey(group.AppName, group.StreamName)) if exist { - groups[i].StatSubs = append(groups[i].StatSubs, session.GetAllConsumer()...) + groups[i].StatSubs = append(groups[i].StatSubs, session.StatSubscribers()...) } } out.Data.Groups = groups diff --git a/server/router_test.go b/server/router_test.go index 68e3fcd..9c19d22 100644 --- a/server/router_test.go +++ b/server/router_test.go @@ -9,9 +9,9 @@ import ( "testing" "time" - "github.com/q191201771/lalmax/hook" + maxlogic "github.com/q191201771/lalmax/logic" - config "github.com/q191201771/lalmax/conf" + config "github.com/q191201771/lalmax/config" "github.com/q191201771/lal/pkg/base" ) @@ -23,8 +23,10 @@ const httpNotifyAddr = ":55559" func TestMain(m *testing.M) { var err error max, err = NewLalMaxServer(&config.Config{ - HttpFmp4Config: config.HttpFmp4Config{Enable: true}, - LalRawContent: []byte(`{"rtmp":{"enable":false},"rtsp":{"enable":false},"http_api":{"enable":false},"pprof":{"enable":false}}`), + Fmp4Config: config.Fmp4Config{ + Http: config.Fmp4HttpConfig{Enable: true}, + }, + LalRawContent: []byte(`{"rtmp":{"enable":false},"rtsp":{"enable":false},"http_api":{"enable":false},"pprof":{"enable":false}}`), HttpConfig: config.HttpConfig{ ListenAddr: ":52349", }, @@ -67,9 +69,9 @@ func TestAllGroup(t *testing.T) { }) t.Run("has consumer", func(t *testing.T) { - ss := hook.NewHookSession("test", "test", max.hlssvr, 1, 0) + ss := maxlogic.NewGroupByStreamName("test", "test", max.hlssvr, 1, 0) ss.AddConsumer("consumer1", nil) - hook.GetHookSessionManagerInstance().SetHookSession("test", ss) + maxlogic.GetGroupManagerInstance().SetGroupByStreamName("test", ss) r := httptest.NewRecorder() req := httptest.NewRequest("GET", "/api/stat/all_group", nil) @@ -103,9 +105,9 @@ func TestNotifyUpdate(t *testing.T) { if err != nil { t.Fatal(err) } - ss := hook.NewHookSession(streamName, streamName, max.hlssvr, 1, 0) + ss := maxlogic.NewGroupByStreamName(streamName, streamName, max.hlssvr, 1, 0) ss.AddConsumer(consumerID, nil) - hook.GetHookSessionManagerInstance().SetHookSession(streamName, ss) + maxlogic.GetGroupManagerInstance().SetGroupByStreamName(streamName, ss) http.HandleFunc("/on_update", func(w http.ResponseWriter, r *http.Request) { var out base.ApiStatAllGroupResp diff --git a/server/server.go b/server/server.go index d230db0..7a28353 100644 --- a/server/server.go +++ b/server/server.go @@ -9,7 +9,7 @@ import ( "github.com/q191201771/lalmax/rtc" - "github.com/q191201771/lalmax/hook" + maxlogic "github.com/q191201771/lalmax/logic" "github.com/q191201771/lalmax/gb28181" @@ -17,7 +17,7 @@ import ( "github.com/q191201771/lalmax/fmp4/hls" - config "github.com/q191201771/lalmax/conf" + config "github.com/q191201771/lalmax/config" "github.com/gin-gonic/gin" "github.com/q191201771/lal/pkg/logic" @@ -67,12 +67,12 @@ func NewLalMaxServer(conf *config.Config) (*LalMaxServer, error) { } } - if conf.HttpFmp4Config.Enable { + if conf.Fmp4Config.Http.Enable { maxsvr.httpfmp4svr = httpfmp4.NewHttpFmp4Server() } - if conf.HlsConfig.Enable { - maxsvr.hlssvr = hls.NewHlsServer(conf.HlsConfig) + if conf.Fmp4Config.Hls.Enable { + maxsvr.hlssvr = hls.NewHlsServer(conf.Fmp4Config.Hls) } if conf.GB28181Config.Enable { @@ -91,8 +91,8 @@ func NewLalMaxServer(conf *config.Config) (*LalMaxServer, error) { func (s *LalMaxServer) Run() (err error) { s.lalsvr.WithOnHookSession(func(uniqueKey string, streamName string) logic.ICustomizeHookSessionContext { - // 有新的流了，创建业务层的对象，用于hook这个流 - return hook.NewHookSession(uniqueKey, streamName, s.hlssvr, s.conf.HookConfig.GopCacheNum, s.conf.HookConfig.SingleGopMaxFrameNum) + // lal 有新的输入流时，创建 lalmax 扩展流组用于分发扩展协议。 + return maxlogic.NewGroupByStreamName(uniqueKey, streamName, s.hlssvr, s.conf.LogicConfig.GopCacheNum, s.conf.LogicConfig.SingleGopMaxFrameNum) }) ctx, cancel := context.WithCancel(context.Background()) diff --git a/srt/sub.go b/srt/sub.go index 45ba91a..311fea4 100644 --- a/srt/sub.go +++ b/srt/sub.go @@ -3,7 +3,7 @@ package srt import ( "context" - "github.com/q191201771/lalmax/hook" + maxlogic "github.com/q191201771/lalmax/logic" srt "github.com/datarhei/gosrt" "github.com/gofrs/uuid" @@ -47,7 +47,7 @@ func NewSubscriber(ctx context.Context, conn srt.Conn, streamName string, maxSen } func (s *Subscriber) Run() { - ok, session := hook.GetHookSessionManagerInstance().GetHookSession(s.streamName) + ok, group := maxlogic.GetGroupManagerInstance().GetGroupByStreamName(s.streamName) if ok { var err error sendBuf := make([]byte, 0, s.maxSendPacketSize*ts.TS_PAKCET_SIZE) @@ -67,7 +67,7 @@ func (s *Subscriber) Run() { } if len(sendBuf) > (s.maxSendPacketSize-1)*ts.TS_PAKCET_SIZE { if _, err = s.conn.Write(sendBuf); err != nil { - session.RemoveConsumer(s.subscriberId) + group.RemoveSubscriber(s.subscriberId) return } sendBuf = sendBuf[0:0] @@ -75,9 +75,12 @@ func (s *Subscriber) Run() { sendBuf = append(sendBuf, tsPacket...) } - session.AddConsumer(s.subscriberId, s) + group.AddSubscriber(maxlogic.SubscriberInfo{ + SubscriberID: s.subscriberId, + Protocol: maxlogic.SubscriberProtocolSRT, + }, s) } else { - nazalog.Warnf("not found hook session, streamName:%s", s.streamName) + nazalog.Warnf("not found stream group, streamName:%s", s.streamName) s.conn.Close() } } @@ -85,9 +88,9 @@ func (s *Subscriber) Run() { func (s *Subscriber) OnMsg(msg base.RtmpMsg) { var err error if !s.hasInit { - ok, session := hook.GetHookSessionManagerInstance().GetHookSession(s.streamName) + ok, group := maxlogic.GetGroupManagerInstance().GetGroupByStreamName(s.streamName) if ok { - videoheader := session.GetVideoSeqHeaderMsg() + videoheader := group.GetVideoSeqHeaderMsg() if videoheader != nil { if videoheader.IsAvcKeySeqHeader() { s.videoPid = s.muxer.AddStream(ts.TS_STREAM_H264) @@ -107,7 +110,7 @@ func (s *Subscriber) OnMsg(msg base.RtmpMsg) { } } - audioheader := session.GetAudioSeqHeaderMsg() + audioheader := group.GetAudioSeqHeaderMsg() if audioheader != nil { if audioheader.IsAacSeqHeader() { s.audioPid = s.muxer.AddStream(ts.TS_STREAM_AAC) diff --git a/thirdparty/srt-1.5.1.tar.gz b/thirdparty/srt-1.5.1.tar.gz deleted file mode 100644 index 683eb52..0000000 Binary files a/thirdparty/srt-1.5.1.tar.gz and /dev/null differ