rtp*: correctly decode timestamp in case of an overflow (~13h)

pkg/rtpaac/decoder.go

@@ -15,8 +15,10 @@ var ErrMorePacketsNeeded = errors.New("need more packets")
 // Decoder is a RTP/AAC decoder.
 type Decoder struct {
 	clockRate time.Duration
-	initialTs    uint32
+
-	initialTsSet bool
+	tsAdd     int64
+	tsInitial *int64
+	tsPrev    *int64
 
 	// for Decode()
 	isDecodingFragmented bool
@@ -31,7 +33,19 @@ func NewDecoder(clockRate int) *Decoder {
 }
 
 func (d *Decoder) decodeTimestamp(ts uint32) time.Duration {
-	return (time.Duration(ts) - time.Duration(d.initialTs)) * time.Second / d.clockRate
+	ts64 := int64(ts) + d.tsAdd
+
+	if d.tsPrev != nil && (ts64-*d.tsPrev) < -0xFFFF {
+		ts64 += 0xFFFFFFFF
+		d.tsAdd += 0xFFFFFFFF
+	}
+	d.tsPrev = &ts64
+
+	if d.tsInitial == nil {
+		d.tsInitial = &ts64
+	}
+
+	return time.Duration(ts64-*d.tsInitial) * time.Second / d.clockRate
 }
 
 // Decode decodes AUs from a RTP/AAC packet.
@@ -52,11 +66,6 @@ func (d *Decoder) Decode(pkt *rtp.Packet) ([][]byte, time.Duration, error) {
 	pkt.Payload = pkt.Payload[2:]
 
 	if !d.isDecodingFragmented {
-		if !d.initialTsSet {
-			d.initialTsSet = true
-			d.initialTs = pkt.Timestamp
-		}
-
 		if pkt.Header.Marker {
 			// AU-headers
 			// AAC headers are 16 bits, where

pkg/rtpaac/rtpaac_test.go

@@ -255,6 +255,35 @@ func TestDecode(t *testing.T) {
 	}
 }
 
+func TestDecodeTimestampOverflow(t *testing.T) {
+	d := NewDecoder(90000)
+	var pts time.Duration
+
+	for _, ts := range []uint32{
+		4294877296,
+		90001,
+		3240090001,
+		565122706,
+	} {
+		pkt := rtp.Packet{
+			Header: rtp.Header{
+				Version:        2,
+				Marker:         true,
+				PayloadType:    96,
+				SequenceNumber: 0x01,
+				Timestamp:      ts,
+				SSRC:           0xba9da416,
+			},
+			Payload: []byte("\x00\x00"),
+		}
+		var err error
+		_, pts, err = d.Decode(&pkt)
+		require.NoError(t, err)
+	}
+
+	require.Equal(t, 15*60*60*time.Second+2*time.Second, pts)
+}
+
 func TestDecodeErrors(t *testing.T) {
 	for _, ca := range []struct {
 		name string

pkg/rtph264/decoder.go

@@ -36,8 +36,9 @@ func (r PacketConnReader) Read(p []byte) (int, error) {
 
 // Decoder is a RTP/H264 decoder.
 type Decoder struct {
-	initialTs    uint32
+	tsAdd     int64
-	initialTsSet bool
+	tsInitial *int64
+	tsPrev    *int64
 
 	// for Decode()
 	startingPacketReceived bool
@@ -54,17 +55,24 @@ func NewDecoder() *Decoder {
 }
 
 func (d *Decoder) decodeTimestamp(ts uint32) time.Duration {
-	return (time.Duration(ts) - time.Duration(d.initialTs)) * time.Second / rtpClockRate
+	ts64 := int64(ts) + d.tsAdd
+
+	if d.tsPrev != nil && (ts64-*d.tsPrev) < -0xFFFF {
+		ts64 += 0xFFFFFFFF
+		d.tsAdd += 0xFFFFFFFF
+	}
+	d.tsPrev = &ts64
+
+	if d.tsInitial == nil {
+		d.tsInitial = &ts64
+	}
+
+	return time.Duration(ts64-*d.tsInitial) * time.Second / rtpClockRate
 }
 
 // Decode decodes NALUs from a RTP/H264 packet.
 func (d *Decoder) Decode(pkt *rtp.Packet) ([][]byte, time.Duration, error) {
 	if !d.isDecodingFragmented {
-		if !d.initialTsSet {
-			d.initialTsSet = true
-			d.initialTs = pkt.Timestamp
-		}
-
 		if len(pkt.Payload) < 1 {
 			return nil, 0, fmt.Errorf("payload is too short")
 		}

pkg/rtph264/rtph264_test.go

@@ -259,6 +259,35 @@ func TestDecode(t *testing.T) {
 	}
 }
 
+func TestDecodeTimestampOverflow(t *testing.T) {
+	d := NewDecoder()
+	var pts time.Duration
+
+	for _, ts := range []uint32{
+		4294877296,
+		90001,
+		3240090001,
+		565122706,
+	} {
+		pkt := rtp.Packet{
+			Header: rtp.Header{
+				Version:        2,
+				Marker:         true,
+				PayloadType:    96,
+				SequenceNumber: 0x01,
+				Timestamp:      ts,
+				SSRC:           0xba9da416,
+			},
+			Payload: []byte("\x00\x00"),
+		}
+		var err error
+		_, pts, err = d.Decode(&pkt)
+		require.NoError(t, err)
+	}
+
+	require.Equal(t, 15*60*60*time.Second+2*time.Second, pts)
+}
+
 func TestDecodePartOfFragmentedBeforeSingle(t *testing.T) {
 	d := NewDecoder()