hls: allow setting current time when writing packets

internal/core/hls_muxer.go

@@ -375,7 +375,7 @@ func (m *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{})
 					}
 					pts := data.pts - *videoInitialPTS
 
-					err = m.muxer.WriteH264(pts, data.h264NALUs)
+					err = m.muxer.WriteH264(time.Now(), pts, data.h264NALUs)
 					if err != nil {
 						return fmt.Errorf("muxer error: %v", err)
 					}
@@ -390,6 +390,7 @@ func (m *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{})
 
 					for i, au := range aus {
 						err = m.muxer.WriteAAC(
+							time.Now(),
 							pts+time.Duration(i)*mpeg4audio.SamplesPerAccessUnit*
 								time.Second/time.Duration(audioTrack.ClockRate()),
 							au)

internal/hls/muxer.go

@@ -76,13 +76,13 @@ func (m *Muxer) Close() {
 }
 
 // WriteH264 writes H264 NALUs, grouped by timestamp.
-func (m *Muxer) WriteH264(pts time.Duration, nalus [][]byte) error {
-	return m.variant.writeH264(pts, nalus)
+func (m *Muxer) WriteH264(now time.Time, pts time.Duration, nalus [][]byte) error {
+	return m.variant.writeH264(now, pts, nalus)
 }
 
 // WriteAAC writes AAC AUs, grouped by timestamp.
-func (m *Muxer) WriteAAC(pts time.Duration, au []byte) error {
-	return m.variant.writeAAC(pts, au)
+func (m *Muxer) WriteAAC(now time.Time, pts time.Duration, au []byte) error {
+	return m.variant.writeAAC(now, pts, au)
 }
 
 // File returns a file reader.

internal/hls/muxer_test.go

@@ -14,6 +14,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+var testTime = time.Date(2010, 0o1, 0o1, 0o1, 0o1, 0o1, 0, time.UTC)
+
 // baseline profile without POC
 var testSPS = []byte{
 	0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
@@ -46,27 +48,27 @@ func TestMuxerVideoAudio(t *testing.T) {
 	defer m.Close()
 
 	// group without IDR
-	err = m.WriteH264(1*time.Second, [][]byte{
+	err = m.WriteH264(testTime, 1*time.Second, [][]byte{
 		{0x06},
 		{0x07},
 	})
 	require.NoError(t, err)
 
 	// group with IDR
-	err = m.WriteH264(2*time.Second, [][]byte{
+	err = m.WriteH264(testTime, 2*time.Second, [][]byte{
 		testSPS, // SPS
 		{8},     // PPS
 		{5},     // IDR
 	})
 	require.NoError(t, err)
 
-	err = m.WriteAAC(3*time.Second, []byte{
+	err = m.WriteAAC(testTime, 3*time.Second, []byte{
 		0x01, 0x02, 0x03, 0x04,
 	})
 	require.NoError(t, err)
 
 	// group without IDR
-	err = m.WriteH264(4*time.Second, [][]byte{
+	err = m.WriteH264(testTime, 4*time.Second, [][]byte{
 		{1}, // non-IDR
 	})
 	require.NoError(t, err)
@@ -74,7 +76,7 @@ func TestMuxerVideoAudio(t *testing.T) {
 	time.Sleep(2 * time.Second)
 
 	// group with IDR
-	err = m.WriteH264(6*time.Second, [][]byte{
+	err = m.WriteH264(testTime, 6*time.Second, [][]byte{
 		{5}, // IDR
 	})
 	require.NoError(t, err)
@@ -205,7 +207,7 @@ func TestMuxerVideoOnly(t *testing.T) {
 	defer m.Close()
 
 	// group with IDR
-	err = m.WriteH264(2*time.Second, [][]byte{
+	err = m.WriteH264(testTime, 2*time.Second, [][]byte{
 		testSPS, // SPS
 		{8},     // PPS
 		{5},     // IDR
@@ -213,7 +215,7 @@ func TestMuxerVideoOnly(t *testing.T) {
 	require.NoError(t, err)
 
 	// group with IDR
-	err = m.WriteH264(6*time.Second, [][]byte{
+	err = m.WriteH264(testTime, 6*time.Second, [][]byte{
 		{5}, // IDR
 	})
 	require.NoError(t, err)
@@ -297,18 +299,18 @@ func TestMuxerAudioOnly(t *testing.T) {
 	defer m.Close()
 
 	for i := 0; i < 100; i++ {
-		err = m.WriteAAC(1*time.Second, []byte{
+		err = m.WriteAAC(testTime, 1*time.Second, []byte{
 			0x01, 0x02, 0x03, 0x04,
 		})
 		require.NoError(t, err)
 	}
 
-	err = m.WriteAAC(2*time.Second, []byte{
+	err = m.WriteAAC(testTime, 2*time.Second, []byte{
 		0x01, 0x02, 0x03, 0x04,
 	})
 	require.NoError(t, err)
 
-	err = m.WriteAAC(3*time.Second, []byte{
+	err = m.WriteAAC(testTime, 3*time.Second, []byte{
 		0x01, 0x02, 0x03, 0x04,
 	})
 	require.NoError(t, err)
@@ -385,7 +387,7 @@ func TestMuxerCloseBeforeFirstSegmentReader(t *testing.T) {
 	require.NoError(t, err)
 
 	// group with IDR
-	err = m.WriteH264(2*time.Second, [][]byte{
+	err = m.WriteH264(testTime, 2*time.Second, [][]byte{
 		testSPS, // SPS
 		{8},     // PPS
 		{5},     // IDR
@@ -409,7 +411,7 @@ func TestMuxerMaxSegmentSize(t *testing.T) {
 	require.NoError(t, err)
 	defer m.Close()
 
-	err = m.WriteH264(2*time.Second, [][]byte{
+	err = m.WriteH264(testTime, 2*time.Second, [][]byte{
 		testSPS,
 		{5}, // IDR
 	})
@@ -427,14 +429,14 @@ func TestMuxerDoubleRead(t *testing.T) {
 	require.NoError(t, err)
 	defer m.Close()
 
-	err = m.WriteH264(0, [][]byte{
+	err = m.WriteH264(testTime, 0, [][]byte{
 		testSPS,
 		{5}, // IDR
 		{1},
 	})
 	require.NoError(t, err)
 
-	err = m.WriteH264(2*time.Second, [][]byte{
+	err = m.WriteH264(testTime, 2*time.Second, [][]byte{
 		{5}, // IDR
 		{2},
 	})

internal/hls/muxer_variant.go

@@ -16,7 +16,7 @@ const (
 
 type muxerVariant interface {
 	close()
-	writeH264(pts time.Duration, nalus [][]byte) error
-	writeAAC(pts time.Duration, au []byte) error
+	writeH264(now time.Time, pts time.Duration, nalus [][]byte) error
+	writeAAC(now time.Time, pts time.Duration, au []byte) error
 	file(name string, msn string, part string, skip string) *MuxerFileResponse
 }

internal/hls/muxer_variant_fmp4.go

@@ -63,12 +63,12 @@ func (v *muxerVariantFMP4) close() {
 	v.playlist.close()
 }
 
-func (v *muxerVariantFMP4) writeH264(pts time.Duration, nalus [][]byte) error {
-	return v.segmenter.writeH264(pts, nalus)
+func (v *muxerVariantFMP4) writeH264(now time.Time, pts time.Duration, nalus [][]byte) error {
+	return v.segmenter.writeH264(now, pts, nalus)
 }
 
-func (v *muxerVariantFMP4) writeAAC(pts time.Duration, au []byte) error {
-	return v.segmenter.writeAAC(pts, au)
+func (v *muxerVariantFMP4) writeAAC(now time.Time, pts time.Duration, au []byte) error {
+	return v.segmenter.writeAAC(now, pts, au)
 }
 
 func (v *muxerVariantFMP4) file(name string, msn string, part string, skip string) *MuxerFileResponse {

internal/hls/muxer_variant_fmp4_segmenter.go

@@ -122,7 +122,7 @@ func (m *muxerVariantFMP4Segmenter) adjustPartDuration(du time.Duration) {
 	}
 }
 
-func (m *muxerVariantFMP4Segmenter) writeH264(pts time.Duration, nalus [][]byte) error {
+func (m *muxerVariantFMP4Segmenter) writeH264(now time.Time, pts time.Duration, nalus [][]byte) error {
 	idrPresent := false
 	nonIDRPresent := false
 
@@ -141,14 +141,14 @@ func (m *muxerVariantFMP4Segmenter) writeH264(pts time.Duration, nalus [][]byte)
 		return nil
 	}
 
-	return m.writeH264Entry(&fmp4.VideoSample{
+	return m.writeH264Entry(now, &fmp4.VideoSample{
 		PTS:        pts,
 		NALUs:      nalus,
 		IDRPresent: idrPresent,
 	})
 }
 
-func (m *muxerVariantFMP4Segmenter) writeH264Entry(sample *fmp4.VideoSample) error {
+func (m *muxerVariantFMP4Segmenter) writeH264Entry(now time.Time, sample *fmp4.VideoSample) error {
 	if !m.videoFirstIDRReceived {
 		// skip sample silently until we find one with an IDR
 		if !sample.IDRPresent {
@@ -190,8 +190,6 @@ func (m *muxerVariantFMP4Segmenter) writeH264Entry(sample *fmp4.VideoSample) err
 	}
 	sample.Next = m.nextVideoSample
 
-	now := time.Now()
-
 	if m.currentSegment == nil {
 		// create first segment
 		m.currentSegment = newMuxerVariantFMP4Segment(
@@ -253,14 +251,14 @@ func (m *muxerVariantFMP4Segmenter) writeH264Entry(sample *fmp4.VideoSample) err
 	return nil
 }
 
-func (m *muxerVariantFMP4Segmenter) writeAAC(pts time.Duration, au []byte) error {
-	return m.writeAACEntry(&fmp4.AudioSample{
+func (m *muxerVariantFMP4Segmenter) writeAAC(now time.Time, pts time.Duration, au []byte) error {
+	return m.writeAACEntry(now, &fmp4.AudioSample{
 		PTS: pts,
 		AU:  au,
 	})
 }
 
-func (m *muxerVariantFMP4Segmenter) writeAACEntry(sample *fmp4.AudioSample) error {
+func (m *muxerVariantFMP4Segmenter) writeAACEntry(now time.Time, sample *fmp4.AudioSample) error {
 	if m.videoTrack != nil {
 		// wait for the video track
 		if !m.videoFirstIDRReceived {
@@ -278,8 +276,6 @@ func (m *muxerVariantFMP4Segmenter) writeAACEntry(sample *fmp4.AudioSample) erro
 	}
 	sample.Next = m.nextAudioSample
 
-	now := time.Now()
-
 	if m.videoTrack == nil {
 		if m.currentSegment == nil {
 			// create first segment

internal/hls/muxer_variant_mpegts.go

@@ -39,12 +39,12 @@ func (v *muxerVariantMPEGTS) close() {
 	v.playlist.close()
 }
 
-func (v *muxerVariantMPEGTS) writeH264(pts time.Duration, nalus [][]byte) error {
-	return v.segmenter.writeH264(pts, nalus)
+func (v *muxerVariantMPEGTS) writeH264(now time.Time, pts time.Duration, nalus [][]byte) error {
+	return v.segmenter.writeH264(now, pts, nalus)
 }
 
-func (v *muxerVariantMPEGTS) writeAAC(pts time.Duration, au []byte) error {
-	return v.segmenter.writeAAC(pts, au)
+func (v *muxerVariantMPEGTS) writeAAC(now time.Time, pts time.Duration, au []byte) error {
+	return v.segmenter.writeAAC(now, pts, au)
 }
 
 func (v *muxerVariantMPEGTS) file(name string, msn string, part string, skip string) *MuxerFileResponse {

internal/hls/muxer_variant_mpegts_segmenter.go

@@ -49,9 +49,7 @@ func newMuxerVariantMPEGTSSegmenter(
 	return m
 }
 
-func (m *muxerVariantMPEGTSSegmenter) writeH264(pts time.Duration, nalus [][]byte) error {
-	now := time.Now()
-
+func (m *muxerVariantMPEGTSSegmenter) writeH264(now time.Time, pts time.Duration, nalus [][]byte) error {
 	idrPresent := false
 	nonIDRPresent := false
 
@@ -127,9 +125,7 @@ func (m *muxerVariantMPEGTSSegmenter) writeH264(pts time.Duration, nalus [][]byt
 	return nil
 }
 
-func (m *muxerVariantMPEGTSSegmenter) writeAAC(pts time.Duration, au []byte) error {
-	now := time.Now()
-
+func (m *muxerVariantMPEGTSSegmenter) writeAAC(now time.Time, pts time.Duration, au []byte) error {
 	if m.videoTrack == nil {
 		if m.currentSegment == nil {
 			m.startPCR = now