重构国标播放与修复快照

.gitignore

@@ -35,3 +35,4 @@ tables/
 *.zip
 .idea/
 data/
+cover/

changelog

@@ -1,3 +1,8 @@
+# 2025-06-14
+重构国标播放逻辑, play 接口仅返回播放地址，通过 webhook 来通知拉流
+可能会导致首播慢一些
+重构 main 函数，将文件移动到项目根目录下
+
 # 2025-02-15
 
 国标流停止，实现 sip bye 请求

configs/config.toml

@@ -1,64 +1,64 @@
 [Server]
-Debug = false
-# rtmp 推流秘钥
-RTMPSecret = '123'
+  Debug = false
+  # rtmp 推流秘钥
+  RTMPSecret = '123'
 
-# 对外提供的服务，建议由 nginx 代理
-[Server.HTTP]
-# http 端口
-Port = 15123
-# 请求超时时间
-Timeout = '1m0s'
-# jwt 秘钥，空串时，每次启动程序将随机赋值
-JwtSecret = ''
+  # 对外提供的服务，建议由 nginx 代理
+  [Server.HTTP]
+    # http 端口
+    Port = 15123
+    # 请求超时时间
+    Timeout = '1m0s'
+    # jwt 秘钥，空串时，每次启动程序将随机赋值
+    JwtSecret = ''
 
-[Server.HTTP.PProf]
-# 是否启用 pprof, 建议设置为 true
-Enabled = true
-# 访问白名单
-AccessIps = ['::1', '127.0.0.1']
+    [Server.HTTP.PProf]
+      # 是否启用 pprof, 建议设置为 true
+      Enabled = true
+      # 访问白名单
+      AccessIps = ['::1', '127.0.0.1']
 
 [Data]
-# 数据库支持 sqlite 和 postgres 两种，使用 sqlite 时 dsn 应当填写文件存储路径
-[Data.Database]
-Dsn = './configs/data.db'
-MaxIdleConns = 1
-MaxOpenConns = 1
-ConnMaxLifetime = '6h0m0s'
-SlowThreshold = '200ms'
+  # 数据库支持 sqlite 和 postgres 两种，使用 sqlite 时 dsn 应当填写文件存储路径
+  [Data.Database]
+    Dsn = './configs/data.db'
+    MaxIdleConns = 1
+    MaxOpenConns = 1
+    ConnMaxLifetime = '6h0m0s'
+    SlowThreshold = '200ms'
 
 [Log]
-# 日志存储目录，不能使用特殊符号
-Dir = './logs'
-# 记录级别 debug/info/warn/error
-Level = 'debug'
-# 保留日志多久，超过时间自动删除
-MaxAge = '744h0m0s'
-# 多久时间，分割一个新的日志文件
-RotationTime = '12h0m0s'
-# 多大文件，分割一个新的日志文件(MB)
-RotationSize = 50
+  # 日志存储目录，不能使用特殊符号
+  Dir = './logs'
+  # 记录级别 debug/info/warn/error
+  Level = 'debug'
+  # 保留日志多久，超过时间自动删除
+  MaxAge = '744h0m0s'
+  # 多久时间，分割一个新的日志文件
+  RotationTime = '12h0m0s'
+  # 多大文件，分割一个新的日志文件(MB)
+  RotationSize = 50
 
 [Sip]
-# 服务监听的 tcp/udp 端口号
-Port = 15060
-# gb/t28181 20 位国标 ID
-ID = '3402000000200000001'
-# 域
-Domain = '3402000000'
-# 注册密码
-Password = ''
+  # 服务监听的 tcp/udp 端口号
+  Port = 15060
+  # gb/t28181 20 位国标 ID
+  ID = '3402000000200000001'
+  # 域
+  Domain = '3402000000'
+  # 注册密码
+  Password = ''
 
 [Media]
-# 媒体服务器 IP
-IP = '127.0.0.1'
-# 媒体服务器 HTTP 端口
-HTTPPort = 8080
-# 媒体服务器密钥
-Secret = 'jvRqCAzEg7AszBi4gm1cfhwXpmnVmJMG'
-# 用于流媒体 webhook 回调
-WebHookIP = '192.168.10.37'
-# 媒体服务器 RTP 端口范围
-RTPPortRange = '20000-20100'
-# 媒体服务器 SDP IP
-SDPIP = '192.168.10.37'
+  # 媒体服务器 IP
+  IP = '127.0.0.1'
+  # 媒体服务器 HTTP 端口
+  HTTPPort = 8080
+  # 媒体服务器密钥
+  Secret = 'jvRqCAzEg7AszBi4gm1cfhwXpmnVmJMG'
+  # 用于流媒体 webhook 回调
+  WebHookIP = '192.168.10.10'
+  # 媒体服务器 RTP 端口范围
+  RTPPortRange = '20000-20100'
+  # 媒体服务器 SDP IP
+  SDPIP = '192.168.10.10'

internal/core/sms/node_manager.go

@@ -168,6 +168,7 @@ func (n *NodeManager) connection(server *MediaServer, serverPort int) error {
 		// HookOnHTTPAccess:     zlm.NewString(""),
 		HookOnPublish:          zlm.NewString(fmt.Sprintf("%s/on_publish", hookPrefix)),
 		HookOnStreamNoneReader: zlm.NewString(fmt.Sprintf("%s/on_stream_none_reader", hookPrefix)),
+		HookOnStreamNotFound:   zlm.NewString(fmt.Sprintf("%s/on_stream_not_found", hookPrefix)),
 		HookOnRecordTs:         zlm.NewString(""),
 		HookOnRtspAuth:         zlm.NewString(""),
 		HookOnRtspRealm:        zlm.NewString(""),

internal/web/api/api.go

@@ -28,6 +28,8 @@ var startRuntime = time.Now()
 func setupRouter(r *gin.Engine, uc *Usecase) {
 	uc.GB28181API.uc = uc
 	uc.SMSAPI.uc = uc
+	uc.WebHookAPI.uc = uc
+
 	go stat.LoadTop(system.Getwd(), func(m map[string]any) {
 		_ = m
 	})
@@ -175,6 +177,8 @@ func sortExpvarMap(data *expvar.Map, top int) []KV {
 }
 
 func (uc *Usecase) proxySMS(c *gin.Context) {
+	defer recover()
+
 	rc := http.NewResponseController(c.Writer)
 	exp := time.Now().AddDate(99, 0, 0)
 	_ = rc.SetReadDeadline(exp)

internal/web/api/gb28181.go

@@ -15,12 +15,10 @@ import (
 	"github.com/gowvp/gb28181/internal/core/gb28181"
 	"github.com/gowvp/gb28181/internal/core/media"
 	"github.com/gowvp/gb28181/internal/core/sms"
-	"github.com/gowvp/gb28181/pkg/gbs"
 	"github.com/gowvp/gb28181/pkg/zlm"
 	"github.com/ixugo/goddd/domain/uniqueid"
 	"github.com/ixugo/goddd/pkg/orm"
 	"github.com/ixugo/goddd/pkg/reason"
-	"github.com/ixugo/goddd/pkg/system"
 	"github.com/ixugo/goddd/pkg/web"
 )
 
@@ -32,13 +30,13 @@ const (
 
 // TODO: 快照不会删除，只会覆盖，设备删除时也不会删除快照，待实现
 func writeCover(dataDir, channelID string, body []byte) error {
-	coverPath := filepath.Join(system.Getwd(), dataDir, coverDir)
+	coverPath := filepath.Join(dataDir, coverDir)
 	os.MkdirAll(coverPath, 0o755)
 	return os.WriteFile(filepath.Join(coverPath, channelID+".jpg"), body, 0o644)
 }
 
 func readCoverPath(dataDir, channelID string) string {
-	coverPath := filepath.Join(system.Getwd(), dataDir, coverDir)
+	coverPath := filepath.Join(dataDir, coverDir)
 	return filepath.Join(coverPath, channelID+".jpg")
 }
 
@@ -179,18 +177,6 @@ func (a GB28181API) play(c *gin.Context, _ *struct{}) (*playOutput, error) {
 			return nil, err
 		}
 
-		dev, err := a.gb28181Core.GetDeviceByDeviceID(c.Request.Context(), ch.DeviceID)
-		if err != nil {
-			return nil, err
-		}
-
-		if err := a.uc.SipServer.Play(&gbs.PlayInput{
-			Channel:    ch,
-			StreamMode: dev.StreamMode,
-			SMS:        svr,
-		}); err != nil {
-			return nil, ErrDevice.SetMsg(err.Error())
-		}
 	} else if strings.HasPrefix(channelID, bz.IDPrefixRTMP) {
 		push, err := a.uc.MediaAPI.mediaCore.GetStreamPush(c.Request.Context(), channelID)
 		if err != nil {
@@ -335,7 +321,7 @@ func (a GB28181API) refreshSnapshot(c *gin.Context, in *refreshSnapshotInput) (a
 		}
 	}
 
-	return gin.H{"link": fmt.Sprintf("/api/channels/%s/snapshot", channelID)}, nil
+	return gin.H{"link": fmt.Sprintf("/channels/%s/snapshot", channelID)}, nil
 }
 
 func (a GB28181API) getSnapshot(c *gin.Context) {

internal/web/api/zlm_webhook.go

@@ -20,6 +20,7 @@ type WebHookAPI struct {
 	conf        *conf.Bootstrap
 	log         *slog.Logger
 	gbs         *gbs.Server
+	uc          *Usecase
 }
 
 func NewWebHookAPI(core sms.Core, mediaCore media.Core, conf *conf.Bootstrap, gbs *gbs.Server, gb28181 gb28181.Core) WebHookAPI {
@@ -42,6 +43,7 @@ func registerZLMWebhookAPI(r gin.IRouter, api WebHookAPI, handler ...gin.Handler
 		group.POST("/on_play", web.WarpH(api.onPlay))
 		group.POST("/on_stream_none_reader", web.WarpH(api.onStreamNoneReader))
 		group.POST("/on_rtp_server_timeout", web.WarpH(api.onRTPServerTimeout))
+		group.POST("/on_stream_not_found", web.WarpH(api.onStreamNotFound))
 	}
 }
 
@@ -159,3 +161,38 @@ func (w WebHookAPI) onRTPServerTimeout(c *gin.Context, in *onRTPServerTimeoutInp
 	w.log.Info("rtp 收流超时", "local_port", in.LocalPort, "ssrc", in.SSRC, "stream_id", in.StreamID, "mediaServerID", in.MediaServerID)
 	return newDefaultOutputOK(), nil
 }
+
+func (w WebHookAPI) onStreamNotFound(c *gin.Context, in *onStreamNotFoundInput) (DefaultOutput, error) {
+	w.log.Info("流不存在", "app", in.App, "stream", in.Stream, "schema", in.Schema, "mediaServerID", in.MediaServerID)
+
+	// 国标流处理
+	if in.App == "rtp" {
+		ch, err := w.gb28181Core.GetChannel(c.Request.Context(), in.Stream)
+		if err != nil {
+			// slog.Error("获取通道失败", "err", err)
+			return newDefaultOutputOK(), nil
+		}
+
+		dev, err := w.gb28181Core.GetDeviceByDeviceID(c.Request.Context(), ch.DeviceID)
+		if err != nil {
+			// slog.Error("获取设备失败", "err", err)
+			return newDefaultOutputOK(), nil
+		}
+
+		svr, err := w.uc.SMSAPI.smsCore.GetMediaServer(c.Request.Context(), sms.DefaultMediaServerID)
+		if err != nil {
+			// slog.Error("GetMediaServer", "err", err)
+			return newDefaultOutputOK(), nil
+		}
+
+		if err := w.gbs.Play(&gbs.PlayInput{
+			Channel:    ch,
+			StreamMode: dev.StreamMode,
+			SMS:        svr,
+		}); err != nil {
+			slog.Error("play", "err", err, "channel", ch.ID)
+			return newDefaultOutputOK(), nil
+		}
+	}
+	return newDefaultOutputOK(), nil
+}

internal/web/api/zlm_webhook_param.go

@@ -201,3 +201,15 @@ type onRTPServerTimeoutInput struct {
 	TCPMode       int    `json:"tcp_mode"`      // openRtpServer 输入的参数
 	MediaServerID string `json:"mediaServerId"` // 服务器 id,通过配置文件设置
 }
+
+type onStreamNotFoundInput struct {
+	MediaServerID string `json:"mediaServerId"` // 服务器 id,通过配置文件设置
+	App           string `json:"app"`           // 流应用名
+	ID            string `json:"id"`            // TCP链接唯一ID
+	IP            string `json:"ip"`            // 播放器ip
+	Params        string `json:"params"`        // 播放url参数
+	Port          int    `json:"port"`          // 播放器端口号
+	Schema        string `json:"schema"`        // 播放的协议，可能是rtsp、rtmp、http
+	Stream        string `json:"stream"`        // 流 ID
+	Vhost         string `json:"vhost"`         // 流虚拟主机
+}

main.go

@@ -44,6 +44,8 @@ func main() {
 	}
 	bc.Debug = !getBuildRelease()
 	bc.BuildVersion = buildVersion
+	bc.ConfigDir = filedir
+	bc.ConfigPath = filePath
 
 	{
 		expvar.NewString("version").Set(buildVersion)

pkg/gbs/play.go

@@ -24,23 +24,18 @@ type StopPlayInput struct {
 	Channel *gb28181.Channel
 }
 
-func (g *GB28181API) StopPlay(in *StopPlayInput) error {
-	ch, ok := g.svr.memoryStorer.GetChannel(in.Channel.DeviceID, in.Channel.ChannelID)
-	if !ok {
-		return ErrDeviceNotExist
-	}
-
-	ch.device.playMutex.Lock()
-	defer ch.device.playMutex.Unlock()
-
+// stopPlay 不加锁的
+func (g *GB28181API) stopPlay(ch *Channel, in *StopPlayInput) error {
 	key := "play:" + in.Channel.DeviceID + ":" + in.Channel.ChannelID
 	stream, ok := g.streams.LoadAndDelete(key)
 	if !ok {
 		return nil
 	}
+
 	if stream.Resp == nil {
 		return nil
 	}
+
 	req := sip.NewRequestFromResponse(sip.MethodBYE, stream.Resp)
 	req.SetDestination(ch.Source())
 	req.SetConnection(ch.Conn())
@@ -53,6 +48,18 @@ func (g *GB28181API) StopPlay(in *StopPlayInput) error {
 	return err
 }
 
+// StopPlay 加锁的停止播放
+func (g *GB28181API) StopPlay(in *StopPlayInput) error {
+	ch, ok := g.svr.memoryStorer.GetChannel(in.Channel.DeviceID, in.Channel.ChannelID)
+	if !ok {
+		return ErrDeviceNotExist
+	}
+
+	ch.device.playMutex.Lock()
+	defer ch.device.playMutex.Unlock()
+	return g.stopPlay(ch, in)
+}
+
 func (g *GB28181API) Play(in *PlayInput) error {
 	ch, ok := g.svr.memoryStorer.GetChannel(in.Channel.DeviceID, in.Channel.ChannelID)
 	if !ok {
@@ -68,7 +75,7 @@ func (g *GB28181API) Play(in *PlayInput) error {
 	if ok {
 		// TODO: 临时解决方案，每次播放，先停止再播放
 		// https://github.com/gowvp/gb28181/issues/16
-		if err := g.StopPlay(&StopPlayInput{
+		if err := g.stopPlay(ch, &StopPlayInput{
 			Channel: in.Channel,
 		}); err != nil {
 			slog.Error("stop play failed", "err", err)