refactor rtp demuxer

av/format/rtp/aac_depacketizer.go

@@ -12,11 +12,11 @@ import (
 )
 
 type aacDepacketizer struct {
+	depacketizer
 	meta        *codec.AudioMeta
 	w           codec.FrameWriter
 	sizeLength  int
 	indexLength int
-	syncClock SyncClock
 }
 
 // NewAacDepacketizer 实例化 AAC 解包器
@@ -31,15 +31,6 @@ func NewAacDepacketizer(meta *codec.AudioMeta, w codec.FrameWriter) Depacketizer
 	return aacdp
 }
 
-func (aacdp *aacDepacketizer) Control(basePts *int64, p *Packet) error {
-	if ok := aacdp.syncClock.Decode(p.Data); ok {
-		if *basePts == 0 {
-			*basePts = aacdp.syncClock.NTPTime
-		}
-	}
-	return nil
-}
-
 //  以下是当 sizelength=13;indexlength=3;indexdeltalength=3 时
 //  Au-header = 13+3 bits(2byte) 的示意图
 // 	0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5
@@ -136,7 +127,3 @@ func (aacdp *aacDepacketizer) depacketizeFor1ByteAUHeader(basePts int64, packet
 
 	return
 }
-
-func (aacdp *aacDepacketizer) rtp2ntp(timestamp uint32) int64 {
-	return aacdp.syncClock.Rtp2Ntp(timestamp)
-}

av/format/rtp/demuxer.go

@@ -16,7 +16,7 @@ import (
 
 // 网络播放时 PTS（Presentation Time Stamp）的延时
 const (
-	ptsDelay = int64(time.Second)
+	ptsDelay = int64(time.Second)/2
 )
 
 // Depacketizer 解包器
@@ -27,8 +27,35 @@ type Depacketizer interface {
 
 type emptyDepacketizer struct{}
 
-func (emptyDepacketizer) Control(basePts *int64, p *Packet) error    { return nil }
+func (emptyDepacketizer) Control(basePts *int64, p *Packet) error     { return nil }
-func (emptyDepacketizer) Depacketize(basePts int64, p *Packet) error { return nil }
+func (emptyDepacketizer) Depacketize(basePts int64, p *Packet) error  { return nil }
+
+type depacketizer struct {
+	syncClock SyncClock
+}
+
+// func (dp *depacketizer) ForcInitSyncClock(basePts *int64, p *Packet) {
+// 	if dp.syncClock.NTPTime == 0 {
+// 		dp.syncClock.RTPTime = p.Timestamp
+// 		dp.syncClock.NTPTime = time.Now().Local().UnixNano()
+// 		if *basePts == 0 {
+// 			*basePts = dp.syncClock.NTPTime
+// 		}
+// 	}
+// }
+
+func (dp *depacketizer) Control(basePts *int64, p *Packet) error {
+	if ok := dp.syncClock.Decode(p.Data); ok {
+		if *basePts == 0 {
+			*basePts = dp.syncClock.NTPTime
+		}
+	}
+	return nil
+}
+
+func (dp *depacketizer) rtp2ntp(timestamp uint32) int64 {
+	return dp.syncClock.Rtp2Ntp(timestamp)
+}
 
 // Demuxer 帧转换器
 type Demuxer struct {

av/format/rtp/h264_depacketizer.go

@@ -13,6 +13,7 @@ import (
 )
 
 type h264Depacketizer struct {
+	depacketizer
 	fragments []*Packet // 分片包
 	meta      *codec.VideoMeta
 	metaReady bool
@@ -20,7 +21,6 @@ type h264Depacketizer struct {
 	dtsStep   float64
 	startOn   time.Time
 	w         codec.FrameWriter
-	syncClock SyncClock
 }
 
 // NewH264Depacketizer 实例化 H264 帧提取器
@@ -34,15 +34,6 @@ func NewH264Depacketizer(meta *codec.VideoMeta, w codec.FrameWriter) Depacketize
 	return h264dp
 }
 
-func (h264dp *h264Depacketizer) Control(basePts *int64, p *Packet) error {
-	if ok := h264dp.syncClock.Decode(p.Data); ok {
-		if *basePts == 0 {
-			*basePts = h264dp.syncClock.NTPTime
-		}
-	}
-	return nil
-}
-
 func (h264dp *h264Depacketizer) Depacketize(basePts int64, packet *Packet) (err error) {
 	if h264dp.syncClock.NTPTime == 0 { // 未收到同步时钟信息，忽略任意包
 		return
@@ -124,7 +115,7 @@ func (h264dp *h264Depacketizer) depacketizeStapa(basePts int64, packet *Packet)
 		}
 		copy(frame.Payload, payload[off:])
 		frame.Payload[0] = 0 | (header & 0x60) | (frame.Payload[0] & 0x1F)
-		if err = h264dp.writeFrame(basePts, packet.Timestamp,frame); err != nil {
+		if err = h264dp.writeFrame(basePts, packet.Timestamp, frame); err != nil {
 			return
 		}
 
@@ -191,16 +182,12 @@ func (h264dp *h264Depacketizer) depacketizeFuA(basePts int64, packet *Packet) (e
 		// 清空分片缓存
 		h264dp.fragments = h264dp.fragments[:0]
 
-		err = h264dp.writeFrame(basePts, packet.Timestamp,frame)
+		err = h264dp.writeFrame(basePts, packet.Timestamp, frame)
 	}
 
 	return
 }
 
-func (h264dp *h264Depacketizer) rtp2ntp(timestamp uint32) int64 {
-	return h264dp.syncClock.Rtp2Ntp(timestamp)
-}
-
 func (h264dp *h264Depacketizer) writeFrame(basePts int64, rtpTimestamp uint32, frame *codec.Frame) error {
 	nalType := frame.Payload[0] & 0x1f
 	switch nalType {
@@ -228,7 +215,7 @@ func (h264dp *h264Depacketizer) writeFrame(basePts int64, rtpTimestamp uint32, f
 		h264dp.metaReady = true
 	}
 
-	frame.Pts = h264dp.rtp2ntp(rtpTimestamp) - basePts+ptsDelay
+	frame.Pts = h264dp.rtp2ntp(rtpTimestamp) - basePts + ptsDelay
 	if h264dp.dtsStep > 0 {
 		frame.Dts = int64(h264dp.nextDts)
 		h264dp.nextDts += h264dp.dtsStep

av/format/rtp/h265_depacketizer.go

@@ -12,6 +12,7 @@ import (
 )
 
 type h265Depacketizer struct {
+	depacketizer
 	fragments []*Packet // 分片包
 	meta      *codec.VideoMeta
 	metaReady bool
@@ -19,7 +20,6 @@ type h265Depacketizer struct {
 	dtsStep   float64
 	startOn   time.Time
 	w         codec.FrameWriter
-	syncClock SyncClock
 }
 
 // NewH265Depacketizer 实例化 H265 帧提取器
@@ -33,15 +33,6 @@ func NewH265Depacketizer(meta *codec.VideoMeta, w codec.FrameWriter) Depacketize
 	return h265dp
 }
 
-func (h265dp *h265Depacketizer) Control(basePts *int64, p *Packet) error {
-	if ok := h265dp.syncClock.Decode(p.Data); ok {
-		if *basePts == 0 {
-			*basePts = h265dp.syncClock.NTPTime
-		}
-	}
-	return nil
-}
-
 /*
  * decode the HEVC payload header according to section 4 of draft version 6:
  *
@@ -177,10 +168,6 @@ func (h265dp *h265Depacketizer) depacketizeFu(basePts int64, packet *Packet) (er
 	return
 }
 
-func (h265dp *h265Depacketizer) rtp2ntp(timestamp uint32) int64 {
-	return h265dp.syncClock.Rtp2Ntp(timestamp)
-}
-
 func (h265dp *h265Depacketizer) writeFrame(basePts int64, rtpTimestamp uint32, frame *codec.Frame) error {
 	nalType := (frame.Payload[0] >> 1) & 0x3f
 	switch nalType {