hls source: support streams that start with negative DTS

internal/hls/client.go

@@ -18,6 +18,7 @@ import (
 	"github.com/asticode/go-astits"
 	"github.com/grafov/m3u8"
 
+	"github.com/aler9/rtsp-simple-server/internal/hls/mpegtstimedec"
 	"github.com/aler9/rtsp-simple-server/internal/logger"
 )
 
@@ -61,7 +62,8 @@ type Client struct {
 	segmentQueue          *clientSegmentQueue
 	pmtDownloaded         bool
 	clockInitialized      bool
-	clockStartPTS         time.Duration
+	timeDec               *mpegtstimedec.Decoder
+	startDTS              time.Duration
 
 	videoPID  *uint16
 	audioPID  *uint16
@@ -130,6 +132,7 @@ func NewClient(
 			},
 		},
 		segmentQueue:  newClientSegmentQueue(),
+		timeDec:       mpegtstimedec.New(),
 		allocateProcs: make(chan clientAllocateProcsReq),
 		outErr:        make(chan error, 1),
 	}
@@ -481,36 +484,49 @@ func (c *Client) processSegment(innerCtx context.Context, byts []byte) error {
 			return fmt.Errorf("PTS is missing")
 		}
 
-		pts := time.Duration(float64(data.PES.Header.OptionalHeader.PTS.Base) * float64(time.Second) / 90000)
-
-		if !c.clockInitialized {
-			c.clockInitialized = true
-			c.clockStartPTS = pts
-			now := time.Now()
-
-			if c.videoPID != nil {
-				c.videoProc.clockStartRTC = now
-			}
-
-			if c.audioPID != nil {
-				c.audioProc.clockStartRTC = now
-			}
-		}
+		pts := c.timeDec.Decode(data.PES.Header.OptionalHeader.PTS.Base)
 
 		if c.videoPID != nil && data.PID == *c.videoPID {
 			var dts time.Duration
 			if data.PES.Header.OptionalHeader.PTSDTSIndicator == astits.PTSDTSIndicatorBothPresent {
-				dts = time.Duration(float64(data.PES.Header.OptionalHeader.DTS.Base) * float64(time.Second) / 90000)
+				diff := time.Duration((data.PES.Header.OptionalHeader.PTS.Base-
+					data.PES.Header.OptionalHeader.DTS.Base)&0x1FFFFFFFF) *
+					time.Second / 90000
+				dts = pts - diff
 			} else {
 				dts = pts
 			}
 
-			pts -= c.clockStartPTS
-			dts -= c.clockStartPTS
+			if !c.clockInitialized {
+				c.clockInitialized = true
+				now := time.Now()
+				if c.videoPID != nil {
+					c.videoProc.clockStartRTC = now
+				}
+				if c.audioPID != nil {
+					c.audioProc.clockStartRTC = now
+				}
+				c.startDTS = dts
+			}
+
+			pts -= c.startDTS
+			dts -= c.startDTS
 
 			c.videoProc.process(data.PES.Data, pts, dts)
 		} else if c.audioPID != nil && data.PID == *c.audioPID {
-			pts -= c.clockStartPTS
+			if !c.clockInitialized {
+				c.clockInitialized = true
+				now := time.Now()
+				if c.videoPID != nil {
+					c.videoProc.clockStartRTC = now
+				}
+				if c.audioPID != nil {
+					c.audioProc.clockStartRTC = now
+				}
+				c.startDTS = pts
+			}
+
+			pts -= c.startDTS
 
 			c.audioProc.process(data.PES.Data, pts)
 		}

internal/hls/client_test.go

@@ -153,8 +153,9 @@ func newTestHLSServer(ca string) (*testHLSServer, error) {
 				Header: &astits.PESHeader{
 					OptionalHeader: &astits.PESOptionalHeader{
 						MarkerBits:      2,
-						PTSDTSIndicator: astits.PTSDTSIndicatorOnlyPTS,
-						PTS:             &astits.ClockReference{Base: int64(1 * 90000)},
+						PTSDTSIndicator: astits.PTSDTSIndicatorBothPresent,
+						PTS:             &astits.ClockReference{Base: 90000},                   // +1 sec
+						DTS:             &astits.ClockReference{Base: 0x1FFFFFFFF - 90000 + 1}, // -1 sec
 					},
 					StreamID: 224, // = video
 				},
@@ -217,10 +218,11 @@ func TestClient(t *testing.T) {
 					return nil
 				},
 				func(pts time.Duration, nalus [][]byte) {
+					require.Equal(t, 2*time.Second, pts)
 					require.Equal(t, [][]byte{
 						{7, 1, 2, 3},
 						{8},
-						{0x05},
+						{5},
 					}, nalus)
 					close(packetRecv)
 				},

internal/hls/mpegtstimedec/decoder.go

@@ -0,0 +1,51 @@
+// Package mpegtstimedec contains a MPEG-TS timestamp decoder.
+package mpegtstimedec
+
+import (
+	"time"
+)
+
+const (
+	maximum           = 0x1FFFFFFFF // 33 bits
+	negativeThreshold = 0xFFFFFFF
+	clockRate         = 90000
+)
+
+// Decoder is a MPEG-TS timestamp decoder.
+type Decoder struct {
+	initialized bool
+	tsOverall   time.Duration
+	tsPrev      int64
+}
+
+// New allocates a Decoder.
+func New() *Decoder {
+	return &Decoder{}
+}
+
+// Decode decodes a MPEG-TS timestamp.
+func (d *Decoder) Decode(ts int64) time.Duration {
+	if !d.initialized {
+		d.initialized = true
+		d.tsPrev = ts
+		return 0
+	}
+
+	diff := (ts - d.tsPrev) & maximum
+
+	// negative difference
+	if diff > negativeThreshold {
+		diff = (d.tsPrev - ts) & maximum
+		d.tsPrev = ts
+		d.tsOverall -= time.Duration(diff)
+	} else {
+		d.tsPrev = ts
+		d.tsOverall += time.Duration(diff)
+	}
+
+	// avoid an int64 overflow and preserve resolution by splitting division into two parts:
+	// first add the integer part, then the decimal part.
+	secs := d.tsOverall / clockRate
+	dec := d.tsOverall % clockRate
+	return secs*time.Second + dec*time.Second/clockRate
+}

internal/hls/mpegtstimedec/decoder_test.go

@@ -0,0 +1,79 @@
+package mpegtstimedec
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNegativeDiff(t *testing.T) {
+	d := New()
+
+	i := int64(0)
+	pts := d.Decode(i)
+	require.Equal(t, time.Duration(0), pts)
+
+	i += 90000 * 2
+	pts = d.Decode(i)
+	require.Equal(t, 2*time.Second, pts)
+
+	i -= 90000 * 1
+	pts = d.Decode(i)
+	require.Equal(t, 1*time.Second, pts)
+
+	i += 90000 * 2
+	pts = d.Decode(i)
+	require.Equal(t, 3*time.Second, pts)
+}
+
+func TestOverflow(t *testing.T) {
+	d := New()
+
+	i := int64(0x1FFFFFFFF - 20)
+	secs := time.Duration(0)
+	pts := d.Decode(i)
+	require.Equal(t, time.Duration(0), pts)
+
+	const stride = 150
+	lim := int64(uint64(0x1FFFFFFFF - (stride * 90000)))
+
+	for n := 0; n < 100; n++ {
+		// overflow
+		i += 90000 * stride
+		secs += stride
+		pts = d.Decode(i)
+		require.Equal(t, secs*time.Second, pts)
+
+		// reach 2^32 slowly
+		secs += stride
+		i += 90000 * stride
+		for ; i < lim; i += 90000 * stride {
+			pts = d.Decode(i)
+			require.Equal(t, secs*time.Second, pts)
+			secs += stride
+		}
+	}
+}
+
+func TestOverflowAndBack(t *testing.T) {
+	d := New()
+
+	pts := d.Decode(0x1FFFFFFFF - 90000 + 1)
+	require.Equal(t, time.Duration(0), pts)
+
+	pts = d.Decode(90000)
+	require.Equal(t, 2*time.Second, pts)
+
+	pts = d.Decode(0x1FFFFFFFF - 90000 + 1)
+	require.Equal(t, time.Duration(0), pts)
+
+	pts = d.Decode(0x1FFFFFFFF - 90000*2 + 1)
+	require.Equal(t, -1*time.Second, pts)
+
+	pts = d.Decode(0x1FFFFFFFF - 90000 + 1)
+	require.Equal(t, time.Duration(0), pts)
+
+	pts = d.Decode(90000)
+	require.Equal(t, 2*time.Second, pts)
+}