From ddc4447db3592ecc33d7b5339afa9ecbd2ddb902 Mon Sep 17 00:00:00 2001 From: betashepherd Date: Thu, 9 Apr 2026 16:05:00 +0800 Subject: [PATCH] fix: improve error handling and logging across multiple plugins - HLS: reset writer state on retry to prevent timeout issues, add nil checks for audio/video tracks, remove unused task import - RTSP: enhance unsupported media warnings with codec details - Snap: support configurable ffmpeg path, add debug logging - Server: remove shutdown script creation on startup --- plugin/hls/pkg/writer.go | 16 +++++++++++----- plugin/rtsp/pkg/client.go | 16 ++++++++++++++-- plugin/rtsp/pkg/transceiver.go | 2 +- plugin/snap/pkg/transform.go | 16 ++++++++++++++-- plugin/snap/pkg/util.go | 4 ++++ server.go | 3 --- 6 files changed, 44 insertions(+), 13 deletions(-) diff --git a/plugin/hls/pkg/writer.go b/plugin/hls/pkg/writer.go index 5ecba489..e89a0d20 100644 --- a/plugin/hls/pkg/writer.go +++ b/plugin/hls/pkg/writer.go @@ -10,7 +10,6 @@ import ( "sync" "time" - task "github.com/langhuihui/gotask" "m7s.live/v5" "m7s.live/v5/pkg/codec" "m7s.live/v5/pkg/format" @@ -63,6 +62,13 @@ func (w *HLSWriter) checkNoBodyRead() bool { } func (w *HLSWriter) Run() (err error) { + // 每次（重）启动时重置状态，避免 retry 后因残留 lastReadTime 立刻超时退出 + w.lastReadTime = time.Time{} + w.hls_segment_count = 0 + w.hls_playlist_count = 0 + w.write_time = 0 + w.memoryTs = sync.Map{} + w.M3u8.Reset() if conf, ok := w.TransformJob.Config.Input.(string); ok { ss := strings.Split(conf, "x") if len(ss) != 2 { @@ -88,10 +94,10 @@ func (w *HLSWriter) Run() (err error) { } MemoryTs.Store(w.TransformJob.StreamPath, w) var audioCodec, videoCodec codec.FourCC - if subscriber.Publisher.HasAudioTrack() { + if subscriber.Publisher.HasAudioTrack() && subscriber.Publisher.AudioTrack.AVTrack != nil && subscriber.Publisher.AudioTrack.ICodecCtx != nil { audioCodec = subscriber.Publisher.AudioTrack.FourCC() } - if subscriber.Publisher.HasVideoTrack() { + if subscriber.Publisher.HasVideoTrack() && subscriber.Publisher.VideoTrack.AVTrack != nil && subscriber.Publisher.VideoTrack.ICodecCtx != nil { videoCodec = subscriber.Publisher.VideoTrack.FourCC() } w.ts = &TsInMemory{} @@ -100,12 +106,12 @@ func (w *HLSWriter) Run() (err error) { return m7s.PlayBlock(subscriber, func(audio *format.Mpeg2Audio) error { pesAudio.Pts = uint64(subscriber.AudioReader.AbsTime) * 90 if w.checkNoBodyRead() { - return errors.Join(ErrNoBodyRead, task.ErrStopByUser) + return ErrNoBodyRead } return pesAudio.WritePESPacket(audio.Memory, &w.ts.RecyclableMemory) }, func(video *mpegts.VideoFrame) (err error) { if w.checkNoBodyRead() { - return errors.Join(ErrNoBodyRead, task.ErrStopByUser) + return ErrNoBodyRead } vr := w.TransformJob.Subscriber.VideoReader if vr.Value.IDR { diff --git a/plugin/rtsp/pkg/client.go b/plugin/rtsp/pkg/client.go index 176caa9c..61ad9df9 100644 --- a/plugin/rtsp/pkg/client.go +++ b/plugin/rtsp/pkg/client.go @@ -106,7 +106,13 @@ func (c *Client) Run() (err error) { return } default: - c.Warn("media kind not support", "kind", media.Kind) + codecNames := make([]string, len(media.Codecs)) + for i, codec := range media.Codecs { + if codec != nil { + codecNames[i] = codec.Name + } + } + c.Warn("media kind not supported, will be skipped", "kind", media.Kind, "direction", media.Direction, "id", media.ID, "codecs", codecNames) } } @@ -139,7 +145,13 @@ func (c *Client) Run() (err error) { return } default: - c.Warn("media kind not support", "kind", media.Kind) + codecNames := make([]string, len(media.Codecs)) + for i, codec := range media.Codecs { + if codec != nil { + codecNames[i] = codec.Name + } + } + c.Warn("media kind not supported, will be skipped", "kind", media.Kind, "direction", media.Direction, "id", media.ID, "codecs", codecNames) } } if err = c.Record(); err != nil { diff --git a/plugin/rtsp/pkg/transceiver.go b/plugin/rtsp/pkg/transceiver.go index e6a0c482..1c011839 100644 --- a/plugin/rtsp/pkg/transceiver.go +++ b/plugin/rtsp/pkg/transceiver.go @@ -415,7 +415,7 @@ func (r *Receiver) SetMedia(medias []*Media) (err error) { } hasVideo = true // 标记找到视频 } else { - r.Stream.Warn("media kind not support", "kind", codec.Kind()) + r.Stream.Warn("media codec not supported, will be skipped", "kind", codec.Kind(), "codecName", codec.Name, "payloadType", codec.PayloadType, "clockRate", codec.ClockRate) } } diff --git a/plugin/snap/pkg/transform.go b/plugin/snap/pkg/transform.go index 0a2f0c18..83068363 100644 --- a/plugin/snap/pkg/transform.go +++ b/plugin/snap/pkg/transform.go @@ -122,6 +122,7 @@ type SnapTask struct { config SnapConfig job *m7s.TransformJob watermarkConfig *WatermarkConfig + ffmpegPath string } // saveSnap 保存截图 @@ -134,7 +135,11 @@ func (t *SnapTask) saveSnap(annexb []*format.AnnexB, snapMode int) error { // 处理视频帧 var buf bytes.Buffer - if err := ProcessWithFFmpeg(annexb, &buf, ""); err != nil { + ffmpegPath := t.ffmpegPath + if ffmpegPath == "" { + ffmpegPath = "ffmpeg" + } + if err := ProcessWithFFmpeg(annexb, &buf, ffmpegPath); err != nil { return fmt.Errorf("process with ffmpeg error: %w", err) } @@ -242,16 +247,23 @@ func NewTransform() m7s.ITransformer { func (t *Transformer) Start() (err error) { // 为每个输出配置创建一个截图任务 - for _, output := range t.TransformJob.Config.Output { + t.Info("snap transform starting", "output_count", len(t.TransformJob.Config.Output)) + for i, output := range t.TransformJob.Config.Output { var task task.ITask var snapConfig SnapConfig if output.Conf != nil { + t.Info("output conf found", "index", i, "type", fmt.Sprintf("%T", output.Conf)) switch v := output.Conf.(type) { case SnapConfig: snapConfig = v + t.Info("parsed as SnapConfig", "timeinterval", snapConfig.TimeInterval, "iframeinterval", snapConfig.IFrameInterval, "savepath", snapConfig.SavePath) case map[string]any: + t.Info("parsing as map[string]any", "content", v) config.Parse(&snapConfig, v) + t.Info("parsed from map", "timeinterval", snapConfig.TimeInterval, "iframeinterval", snapConfig.IFrameInterval, "savepath", snapConfig.SavePath) } + } else { + t.Warn("output.Conf is nil", "index", i) } // 初始化水印配置 diff --git a/plugin/snap/pkg/util.go b/plugin/snap/pkg/util.go index fdcd220d..cbc4e16f 100644 --- a/plugin/snap/pkg/util.go +++ b/plugin/snap/pkg/util.go @@ -44,6 +44,10 @@ func GetVideoFrame(publisher *m7s.Publisher, server *m7s.Server) ([]*format.Anne // ProcessWithFFmpeg 使用 FFmpeg 处理视频帧并生成截图 func ProcessWithFFmpeg(annexb []*format.AnnexB, output io.Writer, FFMPEGPath string) error { + // 如果 FFMPEGPath 为空，使用默认值 + if FFMPEGPath == "" { + FFMPEGPath = "ffmpeg" + } // 创建ffmpeg命令，使用select过滤器选择最后一帧 cmd := exec.Command(FFMPEGPath, "-hide_banner", "-i", "pipe:0", "-vf", fmt.Sprintf("select='eq(n,%d)'", len(annexb)-1), "-vframes", "1", "-f", "mjpeg", "pipe:1") diff --git a/server.go b/server.go index d44d77d5..57a576d7 100644 --- a/server.go +++ b/server.go @@ -210,9 +210,6 @@ func (l errLogger) Println(v ...interface{}) { } func (s *Server) Start() (err error) { - if err = util.CreateShutdownScript(); err != nil { - s.Error("create shutdown script error:", err) - } s.Server = s s.handler = s httpConf, tcpConf := &s.config.HTTP, &s.config.TCP