refactor: 分离解析推流和转发流

stream/source.go

@@ -5,7 +5,6 @@ import (
 	"github.com/lkmio/avformat"
 	"github.com/lkmio/avformat/collections"
 	"github.com/lkmio/lkm/log"
-	"github.com/lkmio/transport"
 	"net"
 	"net/url"
 	"strings"
@@ -14,11 +13,10 @@ import (
 	"time"
 
 	"github.com/lkmio/avformat/utils"
-	"github.com/lkmio/lkm/transcode"
 )
 
 var (
-	StreamEndInfoBride func(s Source) *StreamEndInfo
+	StreamEndInfoBride func(source string, tracks []*Track, streams map[TransStreamID]TransStream) *StreamEndInfo
 )
 
 // Source 对推流源的封装
@@ -39,20 +37,6 @@ type Source interface {
 	// OriginTracks 返回所有的推流track
 	OriginTracks() []*Track
 
-	// TranscodeTracks 返回所有的转码track
-	TranscodeTracks() []*Track
-
-	// AddSink 添加Sink, 在此之前请确保Sink已经握手、授权通过. 如果Source还未WriteHeader，先将Sink添加到等待队列.
-	// 匹配拉流期望的编码器, 创建TransStream或向已经存在TransStream添加Sink
-	AddSink(sink Sink)
-
-	// RemoveSink 同步删除Sink
-	RemoveSink(sink Sink)
-
-	RemoveSinkWithID(id SinkID)
-
-	FindSink(id SinkID) Sink
-
 	SetState(state SessionState)
 
 	// Close 关闭Source
@@ -78,21 +62,15 @@ type Source interface {
 	SetUrlValues(values url.Values)
 
 	// PostEvent 切换到主协程执行当前函数
-	PostEvent(cb func())
+	postEvent(cb func())
 
-	ExecuteSyncEvent(cb func())
+	executeSyncEvent(cb func())
 
 	// LastPacketTime 返回最近收流时间戳
 	LastPacketTime() time.Time
 
 	SetLastPacketTime(time2 time.Time)
 
-	// SinkCount 返回拉流计数
-	SinkCount() int
-
-	// LastStreamEndTime 返回最近结束拉流时间戳
-	LastStreamEndTime() time.Time
-
 	IsClosed() bool
 
 	StreamPipe() chan []byte
@@ -103,15 +81,11 @@ type Source interface {
 
 	SetCreateTime(time time.Time)
 
-	Sinks() []Sink
-
 	GetBitrateStatistics() *BitrateStatistics
 
-	GetTransStreams() map[TransStreamID]TransStream
-
-	GetStreamEndInfo() *StreamEndInfo
-
 	ProbeTimeout()
+
+	GetTransStreamPublisher() TransStreamPublisher
 }
 
 type PublishSource struct {
@@ -120,38 +94,22 @@ type PublishSource struct {
 	state SessionState
 	Conn  net.Conn
 
-	TransDemuxer     avformat.Demuxer       // 负责从推流协议中解析出AVStream和AVPacket
-	recordSink       Sink                   // 每个Source的录制流
-	recordFilePath   string                 // 录制流文件路径
-	hlsStream        TransStream            // HLS传输流, 如果开启, 在@see writeHeader 函数中直接创建, 如果等拉流时再创建, 会进一步加大HLS延迟.
-	audioTranscoders []transcode.Transcoder // 音频解码器
-	videoTranscoders []transcode.Transcoder // 视频解码器
-	originTracks     TrackManager           // 推流的音视频Streams
-	allStreamTracks  TrackManager           // 推流Streams+转码器获得的Stream
-	gopBuffer        GOPBuffer              // GOP缓存, 音频和视频混合使用, 以视频关键帧为界, 缓存第二个视频关键帧时, 释放前一组gop. 如果不存在视频流, 不缓存音频
+	streamPipe        *NonBlockingChannel[[]byte] // 推流数据管道
+	mainContextEvents chan func()                 // 切换到主协程执行函数的事件管道
+	streamPublisher   TransStreamPublisher        // 解析出来的AVStream和AVPacket, 交由streamPublisher处理
 
-	closed     atomic.Bool // source是否已经关闭
-	completed  atomic.Bool // 所有推流track是否解析完毕, @see writeHeader 函数中赋值为true
+	TransDemuxer avformat.Demuxer // 负责从推流协议中解析出AVStream和AVPacket
+	originTracks TrackManager     // 推流的音视频Streams
+
+	closed     atomic.Bool // 是否已经关闭
+	completed  atomic.Bool // 推流track是否解析完毕, @see writeHeader 函数中赋值为true
 	existVideo bool        // 是否存在视频
 
-	TransStreams         map[TransStreamID]TransStream     // 所有输出流
-	ForwardTransStream   TransStream                       // 转发流
-	sinks                map[SinkID]Sink                   // 保存所有Sink
-	TransStreamSinks     map[TransStreamID]map[SinkID]Sink // 输出流对应的Sink
-	streamEndInfo        *StreamEndInfo                    // 之前推流源信息
-	accumulateTimestamps bool                              // 是否累加时间戳
-	timestampModeDecided bool                              // 是否已经决定使用推流的时间戳，或者累加时间戳
-
-	streamPipe        chan []byte // 推流数据管道
-	mainContextEvents chan func() // 切换到主协程执行函数的事件管道
-
-	lastPacketTime    time.Time          // 最近收到推流包的时间
-	lastStreamEndTime time.Time          // 最近拉流端结束拉流的时间
-	sinkCount         int                // 拉流端计数
-	urlValues         url.Values         // 推流url携带的参数
-	createTime        time.Time          // source创建时间
-	statistics        *BitrateStatistics // 码流统计
-	streamLogger      avformat.OnUnpackStream2FileHandler
+	lastPacketTime time.Time          // 最近收到推流包的时间
+	urlValues      url.Values         // 推流url携带的参数
+	createTime     time.Time          // source创建时间
+	statistics     *BitrateStatistics // 码流统计
+	streamLogger   avformat.OnUnpackStream2FileHandler
 }
 
 func (s *PublishSource) SetLastPacketTime(time2 time.Time) {
@@ -163,31 +121,26 @@ func (s *PublishSource) IsClosed() bool {
 }
 
 func (s *PublishSource) StreamPipe() chan []byte {
-	return s.streamPipe
+	return s.streamPipe.Channel
 }
 
 func (s *PublishSource) MainContextEvents() chan func() {
 	return s.mainContextEvents
 }
 
-func (s *PublishSource) LastStreamEndTime() time.Time {
-	return s.lastStreamEndTime
-}
-
 func (s *PublishSource) LastPacketTime() time.Time {
 	return s.lastPacketTime
 }
 
-func (s *PublishSource) SinkCount() int {
-	return s.sinkCount
-}
-
 func (s *PublishSource) GetID() string {
 	return s.ID
 }
 
 func (s *PublishSource) SetID(id string) {
 	s.ID = id
+	if s.streamPublisher != nil {
+		s.streamPublisher.SetSourceID(id)
+	}
 }
 
 func (s *PublishSource) Init(receiveQueueSize int) {
@@ -195,52 +148,16 @@ func (s *PublishSource) Init(receiveQueueSize int) {
 
 	// 初始化事件接收管道
 	// -2是为了保证从管道取到流, 到处理完流整个过程安全的, 不会被覆盖
-	s.streamPipe = make(chan []byte, receiveQueueSize-2)
+	s.streamPipe = NewNonBlockingChannel[[]byte](receiveQueueSize - 1)
 	s.mainContextEvents = make(chan func(), 128)
-
-	s.TransStreams = make(map[TransStreamID]TransStream, 10)
-	s.sinks = make(map[SinkID]Sink, 128)
-	s.TransStreamSinks = make(map[TransStreamID]map[SinkID]Sink, len(transStreamFactories)+1)
 	s.statistics = NewBitrateStatistics()
+	s.streamPublisher = NewTransStreamPublisher(s.ID)
 	// 设置探测时长
 	s.TransDemuxer.SetProbeDuration(AppConfig.ProbeTimeout)
 }
 
-func (s *PublishSource) CreateDefaultOutStreams() {
-	if s.TransStreams == nil {
-		s.TransStreams = make(map[TransStreamID]TransStream, 10)
-	}
-
-	// 创建录制流
-	if AppConfig.Record.Enable {
-		sink, path, err := CreateRecordStream(s.ID)
-		if err != nil {
-			log.Sugar.Errorf("创建录制sink失败 source:%s err:%s", s.ID, err.Error())
-		} else {
-			s.recordSink = sink
-			s.recordFilePath = path
-		}
-	}
-
-	// 创建HLS输出流
-	if AppConfig.Hls.Enable {
-		streams := s.OriginTracks()
-		utils.Assert(len(streams) > 0)
-
-		id := GenerateTransStreamID(TransStreamHls, streams...)
-		hlsStream, err := s.CreateTransStream(id, TransStreamHls, streams)
-		if err != nil {
-			panic(err)
-		}
-
-		s.DispatchGOPBuffer(hlsStream)
-		s.hlsStream = hlsStream
-		s.TransStreams[id] = s.hlsStream
-	}
-}
-
 func (s *PublishSource) Input(data []byte) error {
-	s.streamPipe <- data
+	s.streamPipe.Post(data)
 	s.statistics.Input(len(data))
 	return nil
 }
@@ -249,300 +166,6 @@ func (s *PublishSource) OriginTracks() []*Track {
 	return s.originTracks.All()
 }
 
-func (s *PublishSource) TranscodeTracks() []*Track {
-	return s.allStreamTracks.All()
-}
-
-func IsSupportMux(protocol TransStreamProtocol, _, _ utils.AVCodecID) bool {
-	if TransStreamRtmp == protocol || TransStreamFlv == protocol {
-
-	}
-
-	return true
-}
-
-func (s *PublishSource) CreateTransStream(id TransStreamID, protocol TransStreamProtocol, tracks []*Track) (TransStream, error) {
-	log.Sugar.Infof("创建%s-stream source: %s", protocol.String(), s.ID)
-
-	source := SourceManager.Find(s.ID)
-	utils.Assert(source != nil)
-	transStream, err := CreateTransStream(source, protocol, tracks)
-	if err != nil {
-		log.Sugar.Errorf("创建传输流失败 err: %s source: %s", err.Error(), s.ID)
-		return nil, err
-	}
-
-	for _, track := range tracks {
-		// 重新拷贝一个track，传输流内部使用track的时间戳，
-		newTrack := *track
-		if err = transStream.AddTrack(&newTrack); err != nil {
-			return nil, err
-		}
-	}
-
-	transStream.SetID(id)
-	transStream.SetProtocol(protocol)
-
-	// 创建输出流对应的拉流队列
-	s.TransStreamSinks[id] = make(map[SinkID]Sink, 128)
-	_ = transStream.WriteHeader()
-
-	// 设置转发流
-	if TransStreamGBCascadedForward == transStream.GetProtocol() {
-		s.ForwardTransStream = transStream
-	}
-
-	return transStream, err
-}
-
-func (s *PublishSource) DispatchGOPBuffer(transStream TransStream) {
-	if s.gopBuffer != nil {
-		s.gopBuffer.PeekAll(func(packet *avformat.AVPacket) {
-			s.DispatchPacket(transStream, packet)
-		})
-	}
-}
-
-// DispatchPacket 分发AVPacket
-func (s *PublishSource) DispatchPacket(transStream TransStream, packet *avformat.AVPacket) {
-	data, timestamp, videoKey, err := transStream.Input(packet)
-	if err != nil || len(data) < 1 {
-		return
-	}
-
-	s.DispatchBuffer(transStream, packet.Index, data, timestamp, videoKey)
-}
-
-// DispatchBuffer 分发传输流
-func (s *PublishSource) DispatchBuffer(transStream TransStream, index int, data []*collections.ReferenceCounter[[]byte], timestamp int64, keyVideo bool) {
-	sinks := s.TransStreamSinks[transStream.GetID()]
-	exist := transStream.IsExistVideo()
-
-	for _, sink := range sinks {
-
-		if sink.GetSentPacketCount() < 1 {
-			// 如果存在视频, 确保向sink发送的第一帧是关键帧
-			if exist && !keyVideo {
-				continue
-			}
-
-			if extraData, _, _ := transStream.ReadExtraData(timestamp); len(extraData) > 0 {
-				if ok := s.write(sink, index, extraData, timestamp, false); !ok {
-					continue
-				}
-			}
-		}
-
-		if ok := s.write(sink, index, data, timestamp, keyVideo); !ok {
-			continue
-		}
-	}
-}
-
-func (s *PublishSource) pendingSink(sink Sink) {
-	log.Sugar.Errorf("向sink推流超时,关闭连接. %s-sink: %s source: %s", sink.GetProtocol().String(), sink.GetID(), s.ID)
-	go sink.Close()
-}
-
-// 向sink推流
-func (s *PublishSource) write(sink Sink, index int, data []*collections.ReferenceCounter[[]byte], timestamp int64, keyVideo bool) bool {
-	err := sink.Write(index, data, timestamp, keyVideo)
-	if err == nil {
-		sink.IncreaseSentPacketCount()
-		return true
-	}
-
-	// 推流超时, 可能是服务器或拉流端带宽不够、拉流端不读取数据等情况造成内核发送缓冲区满, 进而阻塞.
-	// 直接关闭连接. 当然也可以将sink先挂起, 后续再继续推流.
-	if _, ok := err.(transport.ZeroWindowSizeError); ok {
-		s.pendingSink(sink)
-	}
-
-	return false
-}
-
-// 创建sink需要的输出流
-func (s *PublishSource) doAddSink(sink Sink, resume bool) bool {
-	// 暂时不考虑多路视频流，意味着只能1路视频流和多路音频流，同理originStreams和allStreams里面的Stream互斥. 同时多路音频流的Codec必须一致
-	audioCodecId, videoCodecId := sink.DesiredAudioCodecId(), sink.DesiredVideoCodecId()
-	audioTrack := s.originTracks.FindWithType(utils.AVMediaTypeAudio)
-	videoTrack := s.originTracks.FindWithType(utils.AVMediaTypeVideo)
-
-	disableAudio := audioTrack == nil
-	disableVideo := videoTrack == nil || !sink.EnableVideo()
-	if disableAudio && disableVideo {
-		return false
-	}
-
-	// 不支持对期望编码的流封装. 降级
-	if (utils.AVCodecIdNONE != audioCodecId || utils.AVCodecIdNONE != videoCodecId) && !IsSupportMux(sink.GetProtocol(), audioCodecId, videoCodecId) {
-		audioCodecId = utils.AVCodecIdNONE
-		videoCodecId = utils.AVCodecIdNONE
-	}
-
-	if !disableAudio && utils.AVCodecIdNONE == audioCodecId {
-		audioCodecId = audioTrack.Stream.CodecID
-	}
-	if !disableVideo && utils.AVCodecIdNONE == videoCodecId {
-		videoCodecId = videoTrack.Stream.CodecID
-	}
-
-	// 创建音频转码器
-	if !disableAudio && audioCodecId != audioTrack.Stream.CodecID {
-		utils.Assert(false)
-	}
-
-	// 创建视频转码器
-	if !disableVideo && videoCodecId != videoTrack.Stream.CodecID {
-		utils.Assert(false)
-	}
-
-	// 查找传输流需要的所有track
-	var tracks []*Track
-	for _, track := range s.originTracks.All() {
-		if disableVideo && track.Stream.MediaType == utils.AVMediaTypeVideo {
-			continue
-		}
-
-		tracks = append(tracks, track)
-	}
-
-	transStreamId := GenerateTransStreamID(sink.GetProtocol(), tracks...)
-	transStream, exist := s.TransStreams[transStreamId]
-	if !exist {
-		var err error
-		transStream, err = s.CreateTransStream(transStreamId, sink.GetProtocol(), tracks)
-		if err != nil {
-			log.Sugar.Errorf("添加sink失败,创建传输流发生err: %s source: %s", err.Error(), s.ID)
-			return false
-		}
-
-		s.TransStreams[transStreamId] = transStream
-	}
-
-	sink.SetTransStreamID(transStreamId)
-
-	{
-		sink.Lock()
-		defer sink.UnLock()
-
-		if SessionStateClosed == sink.GetState() {
-			log.Sugar.Warnf("添加sink失败, sink已经断开连接 %s", sink.String())
-			return false
-		} else {
-			sink.SetState(SessionStateTransferring)
-		}
-	}
-
-	err := sink.StartStreaming(transStream)
-	if err != nil {
-		log.Sugar.Errorf("添加sink失败,开始推流发生err: %s sink: %s source: %s ", err.Error(), SinkId2String(sink.GetID()), s.ID)
-		return false
-	}
-
-	// 还没做好准备(rtsp拉流还在协商sdp中), 暂不推流
-	if !sink.IsReady() {
-		return true
-	}
-
-	// 累加拉流计数
-	if !resume && s.recordSink != sink {
-		s.sinkCount++
-		log.Sugar.Infof("sink count: %d source: %s", s.sinkCount, s.ID)
-	}
-
-	s.sinks[sink.GetID()] = sink
-	s.TransStreamSinks[transStreamId][sink.GetID()] = sink
-
-	// TCP拉流开启异步发包, 一旦出现网络不好的链路, 其余正常链路不受影响.
-	_, ok := sink.GetConn().(*transport.Conn)
-	if ok && sink.IsTCPStreaming() {
-		sink.EnableAsyncWriteMode(24)
-	}
-
-	// 发送已有的缓存数据
-	// 此处发送缓存数据，必须要存在关键帧的输出流才发，否则等DispatchPacket时再发送extra。
-	data, timestamp, _ := transStream.ReadKeyFrameBuffer()
-	if len(data) > 0 {
-		if extraData, _, _ := transStream.ReadExtraData(timestamp); len(extraData) > 0 {
-			s.write(sink, 0, extraData, timestamp, false)
-		}
-
-		s.write(sink, 0, data, timestamp, true)
-	}
-
-	// 新建传输流，发送已经缓存的音视频帧
-	if !exist && AppConfig.GOPCache && s.existVideo && TransStreamGBCascadedForward != transStream.GetProtocol() {
-		s.DispatchGOPBuffer(transStream)
-	}
-
-	return true
-}
-
-func (s *PublishSource) AddSink(sink Sink) {
-	s.PostEvent(func() {
-		if !s.completed.Load() {
-			AddSinkToWaitingQueue(sink.GetSourceID(), sink)
-		} else {
-			if !s.doAddSink(sink, false) {
-				go sink.Close()
-			}
-		}
-	})
-}
-
-func (s *PublishSource) RemoveSink(sink Sink) {
-	s.ExecuteSyncEvent(func() {
-		s.doRemoveSink(sink)
-	})
-}
-
-func (s *PublishSource) RemoveSinkWithID(id SinkID) {
-	s.PostEvent(func() {
-		sink, ok := s.sinks[id]
-		if ok {
-			s.doRemoveSink(sink)
-		}
-	})
-}
-
-func (s *PublishSource) FindSink(id SinkID) Sink {
-	var result Sink
-	s.ExecuteSyncEvent(func() {
-		sink, ok := s.sinks[id]
-		if ok {
-			result = sink
-		}
-	})
-
-	return result
-}
-
-func (s *PublishSource) cleanupSinkStreaming(sink Sink) {
-	transStreamSinks := s.TransStreamSinks[sink.GetTransStreamID()]
-	delete(transStreamSinks, sink.GetID())
-	s.lastStreamEndTime = time.Now()
-
-	if sink.GetProtocol() == TransStreamHls {
-		// 从HLS拉流队列删除Sink
-		_, _ = SinkManager.Remove(sink.GetID())
-	}
-
-	sink.StopStreaming(s.TransStreams[sink.GetTransStreamID()])
-}
-
-func (s *PublishSource) doRemoveSink(sink Sink) bool {
-	s.cleanupSinkStreaming(sink)
-	delete(s.sinks, sink.GetID())
-
-	s.sinkCount--
-	log.Sugar.Infof("sink count: %d source: %s", s.sinkCount, s.ID)
-	utils.Assert(s.sinkCount > -1)
-
-	HookPlayDoneEvent(sink)
-	return true
-}
-
 func (s *PublishSource) SetState(state SessionState) {
 	s.state = state
 }
@@ -556,24 +179,14 @@ func (s *PublishSource) DoClose() {
 
 	s.closed.Store(true)
 
-	// 释放GOP缓存
-	if s.gopBuffer != nil {
-		s.gopBuffer.PopAll(func(packet *avformat.AVPacket) {
-			s.TransDemuxer.DiscardHeadPacket(packet.BufferIndex)
-		})
-		s.gopBuffer = nil
-	}
-
-	// 关闭推流源的解复用器
+	// 关闭推流源的解复用器, 不再接收数据
 	if s.TransDemuxer != nil {
 		s.TransDemuxer.Close()
 		s.TransDemuxer = nil
 	}
 
-	// 关闭录制流
-	if s.recordSink != nil {
-		s.recordSink.Close()
-	}
+	// 等传输流发布器关闭结束
+	s.streamPublisher.close()
 
 	// 释放解复用器
 	// 释放转码器
@@ -581,62 +194,10 @@ func (s *PublishSource) DoClose() {
 	_, err := SourceManager.Remove(s.ID)
 	if err != nil {
 		// source不存在, 在创建source时, 未添加到manager中, 目前只有1078流会出现这种情况(tcp连接到端口, 没有推流或推流数据无效, 无法定位到手机号, 以至于无法执行PreparePublishSource函数), 将不再处理后续事情.
-		log.Sugar.Errorf("删除源失败 source:%s err:%s", s.ID, err.Error())
+		log.Sugar.Errorf("删除源失败 source: %s err: %s", s.ID, err.Error())
 		return
 	}
 
-	// 保留推流信息
-	if s.sinkCount > 0 && len(s.originTracks.All()) > 0 {
-		sourceHistory := StreamEndInfoBride(s)
-		streamEndInfoManager.Add(sourceHistory)
-	}
-
-	// 关闭所有输出流
-	for _, transStream := range s.TransStreams {
-		// 发送剩余包
-		data, ts, _ := transStream.Close()
-		if len(data) > 0 {
-			s.DispatchBuffer(transStream, -1, data, ts, true)
-		}
-
-		// 如果是tcp传输流, 归还合并写缓冲区
-		if !transStream.IsTCPStreaming() || transStream.GetMWBuffer() == nil {
-			continue
-		} else if buffers := transStream.GetMWBuffer().Close(); buffers != nil {
-			AddMWBuffersToPending(s.ID, transStream.GetID(), buffers)
-		}
-	}
-
-	// 将所有sink添加到等待队列
-	for _, sink := range s.sinks {
-		transStreamID := sink.GetTransStreamID()
-		sink.SetTransStreamID(0)
-		if s.recordSink == sink {
-			continue
-		}
-
-		{
-			sink.Lock()
-
-			if SessionStateClosed == sink.GetState() {
-				log.Sugar.Warnf("添加到sink到等待队列失败, sink已经断开连接 %s", sink.String())
-			} else {
-				sink.SetState(SessionStateWaiting)
-				AddSinkToWaitingQueue(s.ID, sink)
-			}
-
-			sink.UnLock()
-		}
-
-		if SessionStateClosed != sink.GetState() {
-			sink.StopStreaming(s.TransStreams[transStreamID])
-		}
-	}
-
-	s.TransStreams = nil
-	s.sinks = nil
-	s.TransStreamSinks = nil
-
 	// 异步hook
 	go func() {
 		if s.Conn != nil {
@@ -645,10 +206,6 @@ func (s *PublishSource) DoClose() {
 		}
 
 		HookPublishDoneEvent(s)
-
-		if s.recordSink != nil {
-			HookRecordEvent(s, s.recordFilePath)
-		}
 	}()
 }
 
@@ -658,7 +215,7 @@ func (s *PublishSource) Close() {
 	}
 
 	// 同步执行, 确保close后, 主协程已经退出, 不会再处理任何推拉流、查询等任何事情.
-	s.ExecuteSyncEvent(func() {
+	s.executeSyncEvent(func() {
 		s.DoClose()
 	})
 }
@@ -672,46 +229,13 @@ func (s *PublishSource) writeHeader() {
 
 	s.completed.Store(true)
 
+	s.streamPublisher.Post(&StreamEvent{StreamEventTypeTrackCompleted, nil})
+
 	if len(s.originTracks.All()) == 0 {
 		log.Sugar.Errorf("没有一路track, 删除source: %s", s.ID)
 		s.DoClose()
 		return
 	}
-
-	// 尝试恢复上次推流的会话
-	if streamInfo := streamEndInfoManager.Remove(s.ID); streamInfo != nil && EqualsTracks(streamInfo, s.originTracks.All()) {
-		s.streamEndInfo = streamInfo
-
-		// 恢复每路track的时间戳
-		tracks := s.originTracks.All()
-		for _, track := range tracks {
-			timestamps := streamInfo.Timestamps[track.Stream.CodecID]
-			track.Dts = timestamps[0]
-			track.Pts = timestamps[1]
-		}
-	}
-
-	// 纠正GOP中的时间戳
-	if s.gopBuffer != nil && s.gopBuffer.Size() != 0 {
-		s.gopBuffer.PeekAll(func(packet *avformat.AVPacket) {
-			s.CorrectTimestamp(packet)
-		})
-	}
-
-	// 创建录制流和HLS
-	s.CreateDefaultOutStreams()
-
-	// 将等待队列的sink添加到输出流队列
-	sinks := PopWaitingSinks(s.ID)
-	if s.recordSink != nil {
-		sinks = append(sinks, s.recordSink)
-	}
-
-	for _, sink := range sinks {
-		if !s.doAddSink(sink, false) {
-			go sink.Close()
-		}
-	}
 }
 
 func (s *PublishSource) IsCompleted() bool {
@@ -729,31 +253,6 @@ func (s *PublishSource) NotTrackAdded(index int) bool {
 	return true
 }
 
-func (s *PublishSource) CorrectTimestamp(packet *avformat.AVPacket) {
-	// 对比第一包的时间戳和上次推流的最后时间戳。如果小于上次的推流时间戳，则在原来的基础上累加。
-	if s.streamEndInfo != nil && !s.timestampModeDecided {
-		s.timestampModeDecided = true
-
-		timestamps := s.streamEndInfo.Timestamps[packet.CodecID]
-		s.accumulateTimestamps = true
-		log.Sugar.Infof("累加时间戳 上次推流dts: %d, pts: %d", timestamps[0], timestamps[1])
-	}
-
-	track := s.originTracks.Find(packet.CodecID)
-	duration := packet.GetDuration(packet.Timebase)
-
-	// 根据duration来累加时间戳
-	if s.accumulateTimestamps {
-		offset := packet.Pts - packet.Dts
-		packet.Dts = track.Dts + duration
-		packet.Pts = packet.Dts + offset
-	}
-
-	track.Dts = packet.Dts
-	track.Pts = packet.Pts
-	track.FrameDuration = int(duration)
-}
-
 func (s *PublishSource) OnNewTrack(track avformat.Track) {
 	if AppConfig.Debug {
 		s.streamLogger.Path = "dump/" + strings.ReplaceAll(s.ID, "/", "_")
@@ -770,17 +269,14 @@ func (s *PublishSource) OnNewTrack(track avformat.Track) {
 		return
 	}
 
-	s.originTracks.Add(NewTrack(stream, 0, 0))
-	s.allStreamTracks.Add(NewTrack(stream, 0, 0))
+	newTrack := NewTrack(stream, 0, 0)
+	s.originTracks.Add(newTrack)
 
 	if utils.AVMediaTypeVideo == stream.MediaType {
 		s.existVideo = true
 	}
 
-	// 创建GOPBuffer
-	if AppConfig.GOPCache && s.existVideo && s.gopBuffer == nil {
-		s.gopBuffer = NewStreamBuffer()
-	}
+	s.streamPublisher.Post(&StreamEvent{StreamEventTypeTrack, newTrack})
 }
 
 func (s *PublishSource) OnTrackComplete() {
@@ -810,32 +306,16 @@ func (s *PublishSource) OnPacket(packet *avformat.AVPacket) {
 		return
 	}
 
-	// 保存到GOP缓存
-	if AppConfig.GOPCache && s.existVideo {
-		// GOP队列溢出
-		if s.gopBuffer.RequiresClear(packet) {
-			s.gopBuffer.PopAll(func(packet *avformat.AVPacket) {
-				s.TransDemuxer.DiscardHeadPacket(packet.BufferIndex)
-			})
-		}
+	packetPtr := collections.NewReferenceCounter(packet)
+	packetPtr.Refer() // 引用计数加1
 
-		s.gopBuffer.AddPacket(packet)
-	}
+	packets := s.originTracks.FindWithType(packet.MediaType).Packets
+	packets.Add(packetPtr)
+	s.streamPublisher.Post(&StreamEvent{StreamEventTypePacket, packetPtr})
 
-	// track解析完毕后，才能生成传输流
-	if s.completed.Load() {
-		s.CorrectTimestamp(packet)
-
-		// 分发给各个传输流
-		for _, transStream := range s.TransStreams {
-			if TransStreamGBCascadedForward != transStream.GetProtocol() {
-				s.DispatchPacket(transStream, packet)
-			}
-		}
-	}
-
-	// 未开启GOP缓存或只存在音频流, 立即释放
-	if !AppConfig.GOPCache || !s.existVideo {
+	// 释放未引用的AVPacket
+	for old := packets.Get(0); old.UseCount() < 2; old = packets.Get(0) {
+		packets.Remove(0).Release()
 		s.TransDemuxer.DiscardHeadPacket(packet.BufferIndex)
 	}
 }
@@ -872,15 +352,15 @@ func (s *PublishSource) SetUrlValues(values url.Values) {
 	s.urlValues = values
 }
 
-func (s *PublishSource) PostEvent(cb func()) {
+func (s *PublishSource) postEvent(cb func()) {
 	s.mainContextEvents <- cb
 }
 
-func (s *PublishSource) ExecuteSyncEvent(cb func()) {
+func (s *PublishSource) executeSyncEvent(cb func()) {
 	group := sync.WaitGroup{}
 	group.Add(1)
 
-	s.PostEvent(func() {
+	s.postEvent(func() {
 		cb()
 		group.Done()
 	})
@@ -896,32 +376,16 @@ func (s *PublishSource) SetCreateTime(time time.Time) {
 	s.createTime = time
 }
 
-func (s *PublishSource) Sinks() []Sink {
-	var sinks []Sink
-
-	s.ExecuteSyncEvent(func() {
-		for _, sink := range s.sinks {
-			sinks = append(sinks, sink)
-		}
-	})
-
-	return sinks
-}
-
 func (s *PublishSource) GetBitrateStatistics() *BitrateStatistics {
 	return s.statistics
 }
 
-func (s *PublishSource) GetTransStreams() map[TransStreamID]TransStream {
-	return s.TransStreams
-}
-
-func (s *PublishSource) GetStreamEndInfo() *StreamEndInfo {
-	return s.streamEndInfo
-}
-
 func (s *PublishSource) ProbeTimeout() {
 	if s.TransDemuxer != nil {
 		s.TransDemuxer.ProbeComplete()
 	}
 }
+
+func (s *PublishSource) GetTransStreamPublisher() TransStreamPublisher {
+	return s.streamPublisher
+}