读取写入增加sleep保持速度

common/frame.go

@@ -102,13 +102,13 @@ func (rtp *RTPFrame) Unmarshal(raw []byte) *RTPFrame {
 
 type BaseFrame struct {
 	DeltaTime  uint32    // 相对上一帧时间戳，毫秒
+	AbsTime    uint32    // 绝对时间戳，毫秒
 	Timestamp  time.Time // 写入时间,可用于比较两个帧的先后
 	SeqInTrack uint32    // 在一个Track中的序号
 	BytesIn    int       // 输入字节数用于计算BPS
 }
 
 type DataFrame[T any] struct {
-	Timestamp time.Time // 写入时间
 	BaseFrame
 	Value T
 }

common/ring_av.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"runtime"
 	"time"
-
 )
 
 type AVRing[T RawSlice] struct {
@@ -36,9 +35,9 @@ func (r *AVRing[T]) Read(ctx context.Context) (item *AVFrame[T]) {
 	return
 }
 
-func (r *AVRing[T]) TryRead(ctx context.Context) (item *AVFrame[T]) {
+// func (r *AVRing[T]) TryRead(ctx context.Context) (item *AVFrame[T]) {
-	if item = &r.Value; ctx.Err() == nil && !item.canRead {
+// 	if item = &r.Value; ctx.Err() == nil && !item.canRead {
-		return nil
+// 		return nil
-	}
+// 	}
-	return
+// 	return
-}
+// }

config/config.go

@@ -1,12 +1,14 @@
 package config
 
 import (
-	"github.com/Monibuca/engine/v4/log"
 	"net"
 	"net/http"
 	"reflect"
 	"strings"
 	"time"
+
+	"github.com/Monibuca/engine/v4/log"
+	"go.uber.org/zap"
 )
 
 type Config map[string]any
@@ -110,8 +112,11 @@ func (config Config) Merge(source Config) {
 			case Config:
 				m.Merge(v.(Config))
 			default:
+				log.Debug("merge", zap.String("k", k), zap.Any("v", v))
 				config[k] = v
 			}
+		} else {
+			log.Debug("exist", zap.String("k", k))
 		}
 	}
 }

http.go

@@ -14,8 +14,6 @@ type GlobalConfig struct {
 }
 
 func (cfg *GlobalConfig) Update(override config.Config) {
-	// 使得RawConfig具备全量配置信息，用于合并到插件配置中
-	Engine.RawConfig = config.Struct2Config(cfg.Engine)
 	log.Info(Green("api server start at"), BrightBlue(cfg.ListenAddr), BrightBlue(cfg.ListenAddrTLS))
 	cfg.Listen(Engine, cfg)
 }

io.go

@@ -15,6 +15,9 @@ import (
 type IOConfig interface {
 	config.Publish | config.Subscribe
 }
+type ClientConfig interface {
+	config.Pull | config.Push
+}
 
 type IO[C IOConfig] struct {
 	ID   string
@@ -40,7 +43,6 @@ func (io *IO[C]) OnEvent(event any) any {
 		io.Context, io.CancelFunc = context.WithCancel(v)
 	case *Stream:
 		io.StartTime = time.Now()
-		io.Stream = v
 		io.Logger = v.With(zap.String("type", io.Type))
 		if io.ID != "" {
 			io.Logger = io.Logger.With(zap.String("ID", io.ID))
@@ -78,16 +80,18 @@ func (io *IO[C]) bye(specific any) {
 	}
 }
 
+// receive 用于接收发布或者订阅
 func (io *IO[C]) receive(streamPath string, specific any, conf *C) bool {
 	Streams.Lock()
 	defer Streams.Unlock()
 	streamPath = strings.Trim(streamPath, "/")
 	u, err := url.Parse(streamPath)
 	if err != nil {
+		io.Error("receive streamPath wrong format", zap.String("streamPath", streamPath), zap.Error(err))
 		return false
 	}
 	io.Args = u.Query()
-	wt := time.Second
+	wt := time.Second*5
 	var c any = conf
 	if v, ok := c.(*config.Subscribe); ok {
 		wt = v.WaitTimeout.Duration()
@@ -99,14 +103,16 @@ func (io *IO[C]) receive(streamPath string, specific any, conf *C) bool {
 	if s.IsClosed() {
 		return false
 	}
+	io.Config = conf
+	io.Stream = s
 	if v, ok := c.(*config.Publish); ok {
 		if s.Publisher != nil && !s.Publisher.IsClosed() {
 			// 根据配置是否剔出原来的发布者
 			if v.KickExist {
 				s.Warn("kick", zap.Any("publisher", s.Publisher))
-				s.Publisher.OnEvent(SEKick{})
+				s.Publisher.OnEvent(SEKick{specific.(IPublisher)})
 			} else {
-				s.Warn("publisher exist", zap.Any("publisher", s.Publisher))
+				s.Warn("badName", zap.Any("publisher", s.Publisher))
 				return false
 			}
 		}
@@ -120,9 +126,13 @@ func (io *IO[C]) receive(streamPath string, specific any, conf *C) bool {
 	if io.Type == "" {
 		io.Type = reflect.TypeOf(specific).Elem().Name()
 	}
-	if s.Receive(specific); io.Stream != nil {
+	s.Receive(specific)
-		io.Config = conf
+	return true
-		return true
+}
-	}
+
-	return false
+type Client[C ClientConfig] struct {
+	Config         *C
+	StreamPath     string // 本地流标识
+	RemoteURL      string // 远程服务器地址（用于推拉）
+	ReConnectCount int    //重连次数
 }

log/log.go

@@ -2,21 +2,32 @@ package log
 
 import (
 	// . "github.com/logrusorgru/aurora"
-	"github.com/mattn/go-colorable"
+	"io"
+
+	// "github.com/mattn/go-colorable"
+	"gopkg.in/yaml.v3"
 
 	// log "github.com/sirupsen/logrus"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 )
 
 var logger *zap.SugaredLogger
+
 // var levelColors = []func(any) Value{Red, Red, Red, Yellow, Blue, Green, White}
 
 // type LogWriter func(*log.Entry) string
 
-var colorableStdout = colorable.NewColorableStdout()
+// var colorableStdout = colorable.NewColorableStdout()
 
 func init() {
-	l, _ := zap.NewDevelopment()
+	config := zap.NewDevelopmentConfig()
+	config.EncoderConfig.NewReflectedEncoder = func(w io.Writer) zapcore.ReflectedEncoder {
+		return yaml.NewEncoder(w)
+	}
+	config.EncoderConfig.EncodeLevel = zapcore.CapitalColorLevelEncoder
+	config.EncoderConfig.EncodeTime = zapcore.TimeEncoderOfLayout("15:04:05")
+	l, _ := config.Build(zap.WithCaller(false))
 	logger = l.Sugar()
 	// std.SetOutput(colorableStdout)
 	// std.SetFormatter(LogWriter(defaultFormatter))
@@ -53,6 +64,7 @@ func Error(args ...any) {
 func Debugf(format string, args ...interface{}) {
 	logger.Debugf(format, args...)
 }
+
 // Infof logs a message at level Info on the standard logger.
 func Infof(format string, args ...interface{}) {
 	logger.Infof(format, args...)

main.go

@@ -83,8 +83,10 @@ func Run(ctx context.Context, configFile string) (err error) {
 	} else {
 		log.Warn("no config file found , use default config values")
 	}
-	Engine.Logger = log.With(zap.String("plugin", "engine"))
+	Engine.Logger = log.With(zap.Bool("engine", true))
 	Engine.registerHandler()
+	// 使得RawConfig具备全量配置信息，用于合并到插件配置中
+	Engine.RawConfig = config.Struct2Config(EngineConfig.Engine)
 	go EngineConfig.Update(Engine.RawConfig)
 	for name, plugin := range Plugins {
 		plugin.RawConfig = cg.GetChild(name)

plugin.go

@@ -122,6 +122,7 @@ func (opt *Plugin) Update() {
 	opt.Context, opt.CancelFunc = context.WithCancel(Engine)
 	opt.RawConfig.Unmarshal(opt.Config)
 	opt.autoPull()
+	opt.Debug("config", zap.Any("config", opt.Config))
 	go opt.Config.Update(opt.RawConfig)
 }
 
@@ -134,16 +135,16 @@ func (opt *Plugin) autoPull() {
 			reflect.ValueOf(&pullConfig).Elem().Set(v.Field(i))
 			for streamPath, url := range pullConfig.PullList {
 				if pullConfig.PullOnStart {
-					opt.Config.(PullPlugin).PullStream(Puller{&pullConfig, streamPath, url, 0})
+					opt.Config.(PullPlugin).PullStream(Puller{Client[config.Pull]{&pullConfig, streamPath, url, 0}})
 				} else if pullConfig.PullOnSubscribe {
-					PullOnSubscribeList[streamPath] = PullOnSubscribe{opt.Config.(PullPlugin), Puller{&pullConfig, streamPath, url, 0}}
+					PullOnSubscribeList[streamPath] = PullOnSubscribe{opt.Config.(PullPlugin), Puller{Client[config.Pull]{&pullConfig, streamPath, url, 0}}}
 				}
 			}
 		} else if name == "Push" {
 			var pushConfig config.Push
 			reflect.ValueOf(&pushConfig).Elem().Set(v.Field(i))
 			for streamPath, url := range pushConfig.PushList {
-				PushOnPublishList[streamPath] = append(PushOnPublishList[streamPath], PushOnPublish{opt.Config.(PushPlugin), Pusher{&pushConfig, streamPath, url, 0}})
+				PushOnPublishList[streamPath] = append(PushOnPublishList[streamPath], PushOnPublish{opt.Config.(PushPlugin), Pusher{Client[config.Push]{&pushConfig, streamPath, url, 0}}})
 			}
 		}
 	}

publisher.go

@@ -30,13 +30,10 @@ type PullEvent int
 
 // 用于远程拉流的发布者
 type Puller struct {
-	Config     *config.Pull
+	Client[config.Pull]
-	StreamPath string
-	RemoteURL  string
-	PullCount  int
 }
 
 // 是否需要重连
 func (pub *Puller) Reconnect() bool {
-	return pub.Config.RePull == -1 || pub.PullCount <= pub.Config.RePull
+	return pub.Config.RePull == -1 || pub.ReConnectCount <= pub.Config.RePull
 }

stream.go

@@ -30,12 +30,18 @@ type SEwaitPublish struct {
 	StateEvent
 	Publisher IPublisher
 }
-
+type SEpublish struct {
+	StateEvent
+}
+type SEwaitClose struct {
+	StateEvent
+}
 type SEclose struct {
 	StateEvent
 }
 
 type SEKick struct {
+	Publisher IPublisher
 }
 
 const (
@@ -56,6 +62,8 @@ const (
 	ACTION_NOTRACKS                 // 轨道为空了
 )
 
+var StateNames = [...]string{"⌛", "🟢", "🟡", "🔴", "❌"}
+var ActionNames = [...]string{"publish", "timeout", "publish lost", "close", "last leave", "first enter", "no tracks"}
 var StreamFSM = [STATE_DESTROYED + 1]map[StreamAction]StreamState{
 	{
 		ACTION_PUBLISH:   STATE_PUBLISHING,
@@ -148,15 +156,18 @@ func findOrCreateStream(streamPath string, waitTimeout time.Duration) (s *Stream
 		return s, true
 	}
 }
-
+func (r *Stream) broadcast(event any) {
-func (r *Stream) action(action StreamAction) bool {
+	for _, sub := range r.Subscribers {
+		sub.OnEvent(event)
+	}
+}
+func (r *Stream) action(action StreamAction) (ok bool) {
 	event := StateEvent{From: r.State, Action: action}
-	if next, ok := event.Next(); ok {
+	if r.State, ok = event.Next(); ok {
 		// 给Publisher状态变更的回调，方便进行远程拉流等操作
 		var stateEvent any
-		r.Debug("state change", zap.Uint8("action", uint8(action)), zap.Uint8("oldState", uint8(r.State)), zap.Uint8("newState", uint8(next)))
+		r.Debug(Sprintf("%s%s%s", StateNames[event.From], Yellow("->"), StateNames[r.State]), zap.String("action", ActionNames[action]))
-		r.State = next
+		switch r.State {
-		switch next {
 		case STATE_WAITPUBLISH:
 			stateEvent = SEwaitPublish{event, r.Publisher}
 			Bus.Publish(Event_REQUEST_PUBLISH, r)
@@ -165,7 +176,9 @@ func (r *Stream) action(action StreamAction) bool {
 				PullOnSubscribeList[r.Path].Pull()
 			}
 		case STATE_PUBLISHING:
-			r.timeout.Reset(time.Second) // 秒级心跳，检测track的存活度
+			stateEvent = SEpublish{event}
+			r.broadcast(stateEvent)
+			r.timeout.Reset(time.Second * 5) // 5秒心跳，检测track的存活度
 			Bus.Publish(Event_PUBLISH, r)
 			if v, ok := PushOnPublishList[r.Path]; ok {
 				for _, v := range v {
@@ -173,12 +186,11 @@ func (r *Stream) action(action StreamAction) bool {
 				}
 			}
 		case STATE_WAITCLOSE:
+			stateEvent = SEwaitClose{event}
 			r.timeout.Reset(r.WaitCloseTimeout)
 		case STATE_CLOSED:
 			stateEvent = SEclose{event}
-			for _, sub := range r.Subscribers {
+			r.broadcast(stateEvent)
-				sub.OnEvent(stateEvent)
-			}
 			r.Subscribers.Reset()
 			Bus.Publish(Event_STREAMCLOSE, r)
 			Streams.Delete(r.Path)
@@ -192,9 +204,8 @@ func (r *Stream) action(action StreamAction) bool {
 		if r.Publisher != nil {
 			r.Publisher.OnEvent(stateEvent)
 		}
-		return true
 	}
-	return false
+	return
 }
 func (r *Stream) IsClosed() bool {
 	if r == nil {
@@ -218,27 +229,24 @@ func (s *Stream) run() {
 	for {
 		select {
 		case <-s.timeout.C:
-			s.Debug("timeout", zap.Uint8("action", uint8(s.State)))
 			if s.State == STATE_PUBLISHING {
 				for name, t := range s.Tracks {
 					// track 超过一定时间没有更新数据了
 					if lastWriteTime := t.LastWriteTime(); !lastWriteTime.IsZero() && time.Since(lastWriteTime) > s.PublishTimeout {
-						s.Warn("track timeout", zap.String("name", name))
+						s.Warn("track timeout", zap.String("name", name), zap.Time("lastWriteTime", lastWriteTime), zap.Duration("timeout", s.PublishTimeout))
 						delete(s.Tracks, name)
-						for _, sub := range s.Subscribers {
+						s.broadcast(TrackRemoved(t))
-							sub.OnEvent(TrackRemoved(t)) // 通知Subscriber Track已被移除
-						}
 					}
 				}
 				if len(s.Tracks) == 0 {
 					s.action(ACTION_NOTRACKS)
 				} else {
-					s.timeout.Reset(time.Second)
+					s.timeout.Reset(time.Second * 5)
 				}
 			} else {
+				s.Debug("timeout", zap.String("state", StateNames[s.State]))
 				s.action(ACTION_TIMEOUT)
 			}
-
 		case action, ok := <-s.actionChan:
 			if ok {
 				switch v := action.(type) {
@@ -253,7 +261,7 @@ func (s *Stream) run() {
 					name := v.GetName()
 					if _, ok := s.Tracks[name]; !ok {
 						s.Tracks[name] = v
-						s.Info("Track added", zap.String("name", name))
+						s.Info("TrackAdd", zap.String("name", name))
 						for _, sub := range s.Subscribers {
 							sub.OnEvent(v) // 通知Subscriber有新Track可用了
 						}

subscriber.go

@@ -21,10 +21,12 @@ type ISubscriber interface {
 type TrackPlayer struct {
 	context.Context
 	context.CancelFunc
-	AudioTrack *track.Audio
+	AudioTrack  *track.Audio
-	VideoTrack *track.Video
+	VideoTrack  *track.Video
-	vr         *AVRing[NALUSlice]
+	vr          *AVRing[NALUSlice]
-	ar         *AVRing[AudioSlice]
+	ar          *AVRing[AudioSlice]
+	startTime   time.Time   //读到第一个关键帧的时间
+	firstIFrame *VideoFrame //起始关键帧
 }
 
 // Subscriber 订阅者实体定义
@@ -66,6 +68,7 @@ func (s *Subscriber) AddTrack(t Track) bool {
 			}
 			s.VideoTrack = v
 			s.vr = v.ReadRing()
+			s.firstIFrame = (*VideoFrame)(s.vr.Read(s.TrackPlayer))
 			return true
 		}
 	} else if a, ok := t.(*track.Audio); ok {
@@ -92,12 +95,31 @@ func (s *Subscriber) Play() {
 	for s.TrackPlayer.Err() == nil {
 		if s.vr != nil {
 			for {
-				vp := s.vr.Read(s.TrackPlayer)
+				// 如果进入正常模式
-				s.OnEvent((*VideoFrame)(vp))
+				if s.firstIFrame == nil {
-				s.vr.MoveNext()
+					vp := s.vr.Read(s.TrackPlayer)
-				if vp.Timestamp.After(t) {
+					s.OnEvent((*VideoFrame)(vp))
-					t = vp.Timestamp
+					s.vr.MoveNext()
-					break
+					if vp.Timestamp.After(t) {
+						t = vp.Timestamp
+						break
+					}
+				} else {
+					if s.startTime.IsZero() {
+						s.startTime = time.Now()
+					}
+					if &s.VideoTrack.IDRing.Value != (*AVFrame[NALUSlice])(s.firstIFrame) {
+						s.firstIFrame = nil
+						s.vr = s.VideoTrack.ReadRing()
+					} else {
+						vp := s.vr.Read(s.TrackPlayer)
+						s.OnEvent((*VideoFrame)(vp))
+						fast := time.Duration(vp.AbsTime-s.firstIFrame.AbsTime)*time.Millisecond - time.Since(s.startTime)
+						if fast > 0 {
+							time.Sleep(fast)
+						}
+						s.vr.MoveNext()
+					}
 				}
 			}
 		}
@@ -118,13 +140,10 @@ func (s *Subscriber) Play() {
 
 type PushEvent int
 type Pusher struct {
-	Config    *config.Push
+	Client[config.Push]
-	StreamPath string
-	RemoteURL string
-	PushCount int
 }
 
 // 是否需要重连
 func (pub *Pusher) Reconnect() bool {
-	return pub.Config.RePush == -1 || pub.PushCount <= pub.Config.RePush
+	return pub.Config.RePush == -1 || pub.ReConnectCount <= pub.Config.RePush
 }

track/aac.go

@@ -11,7 +11,6 @@ import (
 )
 
 func NewAAC(stream IStream) (aac *AAC) {
-	stream.Debug("create aac track")
 	aac = &AAC{}
 	aac.Name = "aac"
 	aac.Stream = stream

track/base.go

@@ -34,10 +34,11 @@ type Media[T RawSlice] struct {
 	SampleSize           byte
 	DecoderConfiguration DecoderConfiguration[T] `json:"-"` //H264(SPS、PPS) H265(VPS、SPS、PPS) AAC(config)
 	// util.BytesPool                               //无锁内存池，用于发布者（在同一个协程中）复用小块的内存，通常是解包时需要临时使用
-	rtpSequence uint16      //用于生成下一个rtp包的序号
+	rtpSequence    uint16      //用于生成下一个rtp包的序号
-	orderQueue  []*RTPFrame //rtp包的缓存队列，用于乱序重排
+	orderQueue     []*RTPFrame //rtp包的缓存队列，用于乱序重排
-	lastSeq     uint16      //上一个收到的序号，用于乱序重排
+	lastSeq        uint16      //上一个收到的序号，用于乱序重排
-	lastSeq2    uint16      //记录上上一个收到的序列号
+	lastSeq2       uint16      //记录上上一个收到的序列号
+	firstTimestamp time.Time   //第一次写入的时间，用于计算总时间防止过快写入
 }
 
 func (av *Media[T]) LastWriteTime() time.Time {
@@ -154,10 +155,18 @@ func (av *Media[T]) WriteAVCC(ts uint32, frame AVCCFrame) {
 }
 
 func (av *Media[T]) Flush() {
-	if av.Prev().Value.DTS != 0 {
+	preValue := av.PreValue()
-		av.Value.DeltaTime = (av.Value.DTS - av.Prev().Value.DTS) / 90
+	if av.firstTimestamp.IsZero() {
+		av.firstTimestamp = time.Now()
+	} else {
+		av.Value.DeltaTime = (av.Value.DTS - preValue.DTS) / 90
+		av.Value.AbsTime += av.Value.DeltaTime
 	}
 	av.Base.Flush(&av.Value.BaseFrame)
+	// 如果收到的帧的时间戳超过实际消耗的时间100ms就休息一下，100ms作为一个弹性区间防止频繁调用sleep
+	if fast := time.Duration(av.Value.AbsTime)*time.Millisecond - time.Since(av.firstTimestamp); fast > time.Millisecond*100 {
+		time.Sleep(fast)
+	}
 	av.Step()
 }
 

track/h264.go

@@ -16,7 +16,6 @@ type H264 struct {
 }
 
 func NewH264(stream IStream) (vt *H264) {
-	stream.Debug("create h264 track")
 	vt = &H264{}
 	vt.Name = "h264"
 	vt.CodecID = codec.CodecID_H264

track/h265.go

@@ -15,7 +15,6 @@ type H265 struct {
 }
 
 func NewH265(stream IStream) (vt *H265) {
-	stream.Debug("create h265 track")
 	vt = &H265{}
 	vt.Name = "h265"
 	vt.CodecID = codec.CodecID_H265