webrtc: reset recomputed audio PTS if it drifts too much (#6021)

internal/protocols/webrtc/from_stream.go

@@ -26,7 +26,8 @@ import (
 )
 
 const (
-	webrtcPayloadMaxSize = 1188 // 1200 - 12 (RTP header)
+	webrtcPayloadMaxSize   = 1188 // 1200 - 12 (RTP header)
+	audioPTSDriftTolerance = 500 * time.Millisecond
 )
 
 var multichannelOpusSDP = map[int]string{
@@ -61,6 +62,11 @@ func timestampToDuration(t int64, clockRate int) time.Duration {
 	return multiplyAndDivide2(time.Duration(t), time.Second, time.Duration(clockRate))
 }
 
+func ptsDriftExceeded(pts uint32, firstPTS uint32, tolerance uint32) bool {
+	delta := int32(pts - firstPTS)
+	return delta > int32(tolerance) || delta < -int32(tolerance)
+}
+
 func setupVideoTrack(
 	desc *description.Session,
 	r *stream.Reader,
@@ -359,16 +365,28 @@ func setupAudioTrack(
 			Caps: caps,
 		}
 
-		curTimestamp, err := randUint32()
-		if err != nil {
-			return nil, err
-		}
+		var pts uint32
+		ptsInitialized := false
+		ptsTolerance := uint32(multiplyAndDivide2(
+			audioPTSDriftTolerance,
+			time.Duration(opusFormat.ClockRate()),
+			time.Second,
+		))
 
 		r.OnData(
 			media,
 			opusFormat,
 			func(u *unit.Unit) error {
-				baseTimestamp := curTimestamp
+				// recompute timestamp from scratch.
+				// Chrome requires a precise timestamp that FFmpeg doesn't provide.
+				// also reset in case of drifts.
+				firstPTS := u.RTPPackets[0].Timestamp
+				if !ptsInitialized || ptsDriftExceeded(pts, firstPTS, ptsTolerance) {
+					pts = firstPTS
+					ptsInitialized = true
+				}
+
+				baseTimestamp := pts
 
 				for _, orig := range u.RTPPackets {
 					// create a copy of the packet that we can edit freely
@@ -377,13 +395,12 @@ func setupAudioTrack(
 						Payload: orig.Payload,
 					}
 
-					// recompute timestamp from scratch.
-					// Chrome requires a precise timestamp that FFmpeg doesn't provide.
-					pkt.Timestamp = curTimestamp
-					curTimestamp += uint32(opus.PacketDuration2(pkt.Payload))
+					pkt.Timestamp = pts
 
 					ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), 48000))
 					track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck
+
+					pts += uint32(opus.PacketDuration2(pkt.Payload))
 				}
 
 				return nil
@@ -482,16 +499,28 @@ func setupAudioTrack(
 		}
 
 		if g711Format.ClockRate() == 8000 {
-			curTimestamp, err := randUint32()
-			if err != nil {
-				return nil, err
-			}
+			var pts uint32
+			ptsInitialized := false
+			ptsTolerance := uint32(multiplyAndDivide2(
+				audioPTSDriftTolerance,
+				time.Duration(g711Format.ClockRate()),
+				time.Second,
+			))
 
 			r.OnData(
 				media,
 				g711Format,
 				func(u *unit.Unit) error {
-					baseTimestamp := curTimestamp
+					// recompute timestamp from scratch.
+					// Chrome requires a precise timestamp that FFmpeg doesn't provide.
+					// also reset in case of drifts.
+					firstPTS := u.RTPPackets[0].Timestamp
+					if !ptsInitialized || ptsDriftExceeded(pts, firstPTS, ptsTolerance) {
+						pts = firstPTS
+						ptsInitialized = true
+					}
+
+					baseTimestamp := pts
 
 					for _, orig := range u.RTPPackets {
 						// create a copy of the packet that we can edit freely
@@ -500,13 +529,12 @@ func setupAudioTrack(
 							Payload: orig.Payload,
 						}
 
-						// recompute timestamp from scratch.
-						// Chrome requires a precise timestamp that FFmpeg doesn't provide.
-						pkt.Timestamp = curTimestamp
-						curTimestamp += uint32(len(pkt.Payload)) / uint32(g711Format.ChannelCount)
+						pkt.Timestamp = pts
 
 						ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), 8000))
 						track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck
+
+						pts += uint32(len(pkt.Payload)) / uint32(g711Format.ChannelCount)
 					}
 
 					return nil
@@ -523,10 +551,13 @@ func setupAudioTrack(
 				return nil, err
 			}
 
-			curTimestamp, err := randUint32()
-			if err != nil {
-				return nil, err
-			}
+			var pts uint32
+			ptsInitialized := false
+			ptsTolerance := uint32(multiplyAndDivide2(
+				audioPTSDriftTolerance,
+				time.Duration(g711Format.ClockRate()),
+				time.Second,
+			))
 
 			r.OnData(
 				media,
@@ -552,16 +583,24 @@ func setupAudioTrack(
 						return nil //nolint:nilerr
 					}
 
-					baseTimestamp := curTimestamp
+					// recompute timestamp from scratch.
+					// Chrome requires a precise timestamp that FFmpeg doesn't provide.
+					// also reset in case of drifts.
+					firstPTS := u.RTPPackets[0].Timestamp
+					if !ptsInitialized || ptsDriftExceeded(pts, firstPTS, ptsTolerance) {
+						pts = firstPTS
+						ptsInitialized = true
+					}
+
+					baseTimestamp := pts
 
 					for _, pkt := range packets {
-						// recompute timestamp from scratch.
-						// Chrome requires a precise timestamp that FFmpeg doesn't provide.
-						pkt.Timestamp = curTimestamp
-						curTimestamp += uint32(len(pkt.Payload)) / 2 / uint32(g711Format.ChannelCount)
+						pkt.Timestamp = pts
 
 						ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), g711Format.ClockRate()))
 						track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck
+
+						pts += uint32(len(pkt.Payload)) / 2 / uint32(g711Format.ChannelCount)
 					}
 
 					return nil
@@ -609,10 +648,13 @@ func setupAudioTrack(
 			return nil, err
 		}
 
-		curTimestamp, err := randUint32()
-		if err != nil {
-			return nil, err
-		}
+		var pts uint32
+		ptsInitialized := false
+		ptsTolerance := uint32(multiplyAndDivide2(
+			audioPTSDriftTolerance,
+			time.Duration(lpcmFormat.ClockRate()),
+			time.Second,
+		))
 
 		r.OnData(
 			media,
@@ -627,16 +669,24 @@ func setupAudioTrack(
 					return nil //nolint:nilerr
 				}
 
-				baseTimestamp := curTimestamp
+				// recompute timestamp from scratch.
+				// Chrome requires a precise timestamp that FFmpeg doesn't provide.
+				// also reset in case of drifts.
+				firstPTS := u.RTPPackets[0].Timestamp
+				if !ptsInitialized || ptsDriftExceeded(pts, firstPTS, ptsTolerance) {
+					pts = firstPTS
+					ptsInitialized = true
+				}
+
+				baseTimestamp := pts
 
 				for _, pkt := range packets {
-					// recompute timestamp from scratch.
-					// Chrome requires a precise timestamp that FFmpeg doesn't provide.
-					pkt.Timestamp = curTimestamp
-					curTimestamp += uint32(len(pkt.Payload)) / 2 / uint32(lpcmFormat.ChannelCount)
+					pkt.Timestamp = pts
 
 					ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), lpcmFormat.ClockRate()))
 					track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck
+
+					pts += uint32(len(pkt.Payload)) / 2 / uint32(lpcmFormat.ChannelCount)
 				}
 
 				return nil

internal/protocols/webrtc/from_stream_test.go

@@ -1,4 +1,4 @@
-package webrtc
+package webrtc_test
 
 import (
 	"fmt"
@@ -8,6 +8,7 @@ import (
 	"github.com/bluenviron/gortsplib/v5/pkg/description"
 	"github.com/bluenviron/gortsplib/v5/pkg/format"
 	"github.com/bluenviron/mediamtx/internal/logger"
+	"github.com/bluenviron/mediamtx/internal/protocols/webrtc"
 	"github.com/bluenviron/mediamtx/internal/stream"
 	"github.com/bluenviron/mediamtx/internal/test"
 	"github.com/bluenviron/mediamtx/internal/unit"
@@ -27,10 +28,10 @@ func TestFromStreamNoSupportedCodecs(t *testing.T) {
 		}),
 	}
 
-	pc := &PeerConnection{}
+	pc := &webrtc.PeerConnection{}
 
-	err := FromStream(desc, r, pc)
-	require.Equal(t, errNoSupportedCodecsFrom, err)
+	err := webrtc.FromStream(desc, r, pc)
+	require.ErrorContains(t, err, "the stream doesn't contain any supported codec")
 }
 
 func TestFromStreamSkipUnsupportedTracks(t *testing.T) {
@@ -57,9 +58,9 @@ func TestFromStreamSkipUnsupportedTracks(t *testing.T) {
 		}),
 	}
 
-	pc := &PeerConnection{}
+	pc := &webrtc.PeerConnection{}
 
-	err := FromStream(desc, r, pc)
+	err := webrtc.FromStream(desc, r, pc)
 	require.NoError(t, err)
 
 	require.Equal(t, 1, n)
@@ -74,10 +75,10 @@ func TestFromStream(t *testing.T) {
 				}},
 			}
 
-			pc := &PeerConnection{}
+			pc := &webrtc.PeerConnection{}
 			r := &stream.Reader{Parent: test.NilLogger}
 
-			err := FromStream(desc, r, pc)
+			err := webrtc.FromStream(desc, r, pc)
 			require.NoError(t, err)
 
 			require.Equal(t, ca.webrtcCaps, pc.OutboundTracks[0].Caps)
@@ -85,302 +86,236 @@ func TestFromStream(t *testing.T) {
 	}
 }
 
-func TestFromStreamResampleOpus(t *testing.T) {
-	strm := &stream.Stream{
-		OrigDesc: &description.Session{Medias: []*description.Media{
-			{
-				Type: description.MediaTypeAudio,
-				Formats: []format.Format{&format.Opus{
-					ChannelCount: 2,
-				}},
+func TestFromStreamResampleAudio(t *testing.T) {
+	for _, ca := range []struct {
+		name            string
+		format          format.Format
+		payloadType     uint8
+		payload         []byte
+		step            time.Duration
+		expectedTSDelta uint32
+	}{
+		{
+			name: "opus stereo",
+			format: &format.Opus{
+				ChannelCount: 2,
 			},
-		}},
-		WriteQueueSize:    512,
-		RTPMaxPayloadSize: 1450,
-		ReplaceNTP:        false,
-		Parent:            test.NilLogger,
-	}
-	err := strm.Initialize()
-	require.NoError(t, err)
-
-	subStream := &stream.SubStream{
-		Stream:        strm,
-		UseRTPPackets: true,
-	}
-	err = subStream.Initialize()
-	require.NoError(t, err)
-
-	pc1 := &PeerConnection{
-		LocalRandomUDP:    true,
-		IPsFromInterfaces: true,
-		Publish:           false,
-		Log:               test.NilLogger,
-	}
-	err = pc1.Start()
-	require.NoError(t, err)
-	defer pc1.Close()
-
-	pc2 := &PeerConnection{
-		LocalRandomUDP:    true,
-		IPsFromInterfaces: true,
-		Publish:           true,
-		Log:               test.NilLogger,
-	}
-
-	r := &stream.Reader{Parent: nil}
-
-	err = FromStream(strm.OrigDesc, r, pc2)
-	require.NoError(t, err)
-
-	err = pc2.Start()
-	require.NoError(t, err)
-	defer pc2.Close()
-
-	offer, err := pc1.CreatePartialOffer(false)
-	require.NoError(t, err)
-
-	answer, err := pc2.CreateFullAnswer(offer, false)
-	require.NoError(t, err)
-
-	err = pc1.SetAnswer(answer)
-	require.NoError(t, err)
-
-	err = pc1.WaitUntilConnected(10 * time.Second)
-	require.NoError(t, err)
-
-	err = pc2.WaitUntilConnected(10 * time.Second)
-	require.NoError(t, err)
-
-	strm.AddReader(r)
-	defer strm.RemoveReader(r)
-
-	subStream.WriteUnit(strm.OrigDesc.Medias[0], strm.OrigDesc.Medias[0].Formats[0], &unit.Unit{
-		PTS: 0,
-		NTP: time.Now(),
-		RTPPackets: []*rtp.Packet{{
-			Header: rtp.Header{
-				Version:        2,
-				Marker:         true,
-				PayloadType:    111,
-				SequenceNumber: 1123,
-				Timestamp:      45343,
-				SSRC:           563424,
+			payloadType:     111,
+			payload:         []byte{1},
+			step:            20 * time.Millisecond,
+			expectedTSDelta: 960,
+		},
+		{
+			name: "g711 pcma 8khz mono",
+			format: &format.G711{
+				PayloadTyp:   8,
+				SampleRate:   8000,
+				ChannelCount: 1,
 			},
-			Payload: []byte{1},
-		}},
-	})
-
-	subStream.WriteUnit(strm.OrigDesc.Medias[0], strm.OrigDesc.Medias[0].Formats[0], &unit.Unit{
-		PTS: 0,
-		NTP: time.Now(),
-		RTPPackets: []*rtp.Packet{{
-			Header: rtp.Header{
-				Version:        2,
-				Marker:         true,
-				PayloadType:    111,
-				SequenceNumber: 1124,
-				Timestamp:      45343,
-				SSRC:           563424,
+			payloadType:     8,
+			payload:         make([]byte, 160),
+			step:            20 * time.Millisecond,
+			expectedTSDelta: 160,
+		},
+		{
+			name: "g711 pcmu 16khz stereo",
+			format: &format.G711{
+				MULaw:        true,
+				PayloadTyp:   96,
+				SampleRate:   16000,
+				ChannelCount: 2,
 			},
-			Payload: []byte{1},
-		}},
-	})
-
-	err = pc1.GatherInboundTracks(2 * time.Second)
-	require.NoError(t, err)
-
-	tracks := pc1.InboundTracks()
-
-	done := make(chan struct{})
-	n := 0
-	var ts uint32
-
-	tracks[0].OnPacketRTP = func(pkt *rtp.Packet) {
-		n++
-
-		switch n {
-		case 1:
-			ts = pkt.Timestamp
-
-		case 2:
-			require.Equal(t, uint32(960), pkt.Timestamp-ts)
-			close(done)
-		}
-	}
-
-	pc1.StartReading()
-
-	<-done
-}
-
-func TestFromStreamResampleOpusAbsoluteTimestamp(t *testing.T) {
-	strm := &stream.Stream{
-		OrigDesc: &description.Session{Medias: []*description.Media{
-			{
-				Type: description.MediaTypeAudio,
-				Formats: []format.Format{&format.Opus{
-					ChannelCount: 2,
-				}},
+			payloadType:     96,
+			payload:         make([]byte, 320),
+			step:            10 * time.Millisecond,
+			expectedTSDelta: 160,
+		},
+		{
+			name: "lpcm 16khz stereo",
+			format: &format.LPCM{
+				PayloadTyp:   96,
+				BitDepth:     16,
+				SampleRate:   16000,
+				ChannelCount: 2,
 			},
-		}},
-		WriteQueueSize:    512,
-		RTPMaxPayloadSize: 1450,
-		ReplaceNTP:        false,
-		Parent:            test.NilLogger,
-	}
-	err := strm.Initialize()
-	require.NoError(t, err)
-
-	subStream := &stream.SubStream{
-		Stream:        strm,
-		UseRTPPackets: true,
-	}
-	err = subStream.Initialize()
-	require.NoError(t, err)
-
-	pcReader := &PeerConnection{
-		LocalRandomUDP:    true,
-		IPsFromInterfaces: true,
-		Publish:           false,
-		Log:               test.NilLogger,
-	}
-	err = pcReader.Start()
-	require.NoError(t, err)
-	t.Cleanup(pcReader.Close)
-
-	pcPublisher := &PeerConnection{
-		LocalRandomUDP:    true,
-		IPsFromInterfaces: true,
-		Publish:           true,
-		Log:               test.NilLogger,
-	}
-
-	r := &stream.Reader{Parent: nil}
-
-	err = FromStream(strm.OrigDesc, r, pcPublisher)
-	require.NoError(t, err)
-
-	err = pcPublisher.Start()
-	require.NoError(t, err)
-	t.Cleanup(pcPublisher.Close)
-
-	offer, err := pcReader.CreatePartialOffer(false)
-	require.NoError(t, err)
-
-	answer, err := pcPublisher.CreateFullAnswer(offer, false)
-	require.NoError(t, err)
-
-	err = pcReader.SetAnswer(answer)
-	require.NoError(t, err)
-
-	err = pcReader.WaitUntilConnected(10 * time.Second)
-	require.NoError(t, err)
-
-	err = pcPublisher.WaitUntilConnected(10 * time.Second)
-	require.NoError(t, err)
-
-	strm.AddReader(r)
-	t.Cleanup(func() { strm.RemoveReader(r) })
-
-	baseNTP := time.Unix(1710000000, 0)
-	step := 20 * time.Millisecond
-
-	// prime the pipeline to allow track gathering
-	subStream.WriteUnit(strm.OrigDesc.Medias[0], strm.OrigDesc.Medias[0].Formats[0], &unit.Unit{
-		PTS: 0,
-		NTP: baseNTP,
-		RTPPackets: []*rtp.Packet{{
-			Header: rtp.Header{
-				Version:        2,
-				Marker:         true,
-				PayloadType:    111,
-				SequenceNumber: 1123,
-				Timestamp:      45343,
-				SSRC:           563424,
-			},
-			Payload: []byte{1},
-		}},
-	})
-
-	err = pcReader.GatherInboundTracks(2 * time.Second)
-	require.NoError(t, err)
-
-	tracks := pcReader.InboundTracks()
-	require.Len(t, tracks, 1)
-
-	done := make(chan struct{})
-	errCh := make(chan string, 1)
-	const startSeq = uint16(2000)
-
-	expectedNTP := func(seq uint16) (time.Time, bool) {
-		if seq < startSeq {
-			return time.Time{}, false
-		}
-		return baseNTP.Add(time.Duration(seq-startSeq) * step), true
-	}
-
-	tracks[0].OnPacketRTP = func(pkt *rtp.Packet) {
-		expected, ok := expectedNTP(pkt.SequenceNumber)
-		if !ok {
-			return
-		}
-
-		ntp, avail := tracks[0].PacketNTP(pkt)
-		if !avail {
-			return
-		}
-
-		if ntp.Sub(expected).Abs() > 50*time.Millisecond {
-			select {
-			case errCh <- fmt.Sprintf("absolute NTP mismatch for seq=%d: got=%v expected=%v",
-				pkt.SequenceNumber, ntp, expected):
-			default:
+			payloadType:     96,
+			payload:         make([]byte, 640),
+			step:            10 * time.Millisecond,
+			expectedTSDelta: 160,
+		},
+	} {
+		t.Run(ca.name, func(t *testing.T) {
+			strm := &stream.Stream{
+				OrigDesc: &description.Session{Medias: []*description.Media{{
+					Type:    description.MediaTypeAudio,
+					Formats: []format.Format{ca.format},
+				}}},
+				WriteQueueSize:    512,
+				RTPMaxPayloadSize: 1450,
+				ReplaceNTP:        false,
+				Parent:            test.NilLogger,
 			}
-			return
-		}
+			err := strm.Initialize()
+			require.NoError(t, err)
+			t.Cleanup(strm.Close)
 
-		select {
-		case done <- struct{}{}:
-		default:
-		}
-	}
+			subStream := &stream.SubStream{
+				Stream:        strm,
+				UseRTPPackets: true,
+			}
+			err = subStream.Initialize()
+			require.NoError(t, err)
 
-	pcReader.StartReading()
+			pcReader := &webrtc.PeerConnection{
+				LocalRandomUDP:    true,
+				IPsFromInterfaces: true,
+				Publish:           false,
+				Log:               test.NilLogger,
+			}
+			err = pcReader.Start()
+			require.NoError(t, err)
+			t.Cleanup(pcReader.Close)
 
-	go func() {
-		ticker := time.NewTicker(step)
-		defer ticker.Stop()
+			pcPublisher := &webrtc.PeerConnection{
+				LocalRandomUDP:    true,
+				IPsFromInterfaces: true,
+				Publish:           true,
+				Log:               test.NilLogger,
+			}
 
-		for i := range uint16(150) {
-			seq := startSeq + i
-			expected, _ := expectedNTP(seq)
+			r := &stream.Reader{Parent: nil}
 
-			subStream.WriteUnit(strm.OrigDesc.Medias[0], strm.OrigDesc.Medias[0].Formats[0], &unit.Unit{
-				PTS: 0,
-				NTP: expected,
-				RTPPackets: []*rtp.Packet{{
-					Header: rtp.Header{
-						Version:        2,
-						Marker:         true,
-						PayloadType:    111,
-						SequenceNumber: seq,
-						Timestamp:      45343,
-						SSRC:           563424,
-					},
-					Payload: []byte{1},
-				}},
-			})
+			err = webrtc.FromStream(strm.OrigDesc, r, pcPublisher)
+			require.NoError(t, err)
 
-			<-ticker.C
-		}
-	}()
+			err = pcPublisher.Start()
+			require.NoError(t, err)
+			t.Cleanup(pcPublisher.Close)
 
-	select {
-	case <-done:
-	case err := <-errCh:
-		t.Fatal(err)
-	case <-time.After(8 * time.Second):
-		t.Fatal("absolute timestamp mapping did not become available")
+			offer, err := pcReader.CreatePartialOffer(false)
+			require.NoError(t, err)
+
+			answer, err := pcPublisher.CreateFullAnswer(offer, false)
+			require.NoError(t, err)
+
+			err = pcReader.SetAnswer(answer)
+			require.NoError(t, err)
+
+			err = pcReader.WaitUntilConnected(10 * time.Second)
+			require.NoError(t, err)
+
+			err = pcPublisher.WaitUntilConnected(10 * time.Second)
+			require.NoError(t, err)
+
+			strm.AddReader(r)
+			t.Cleanup(func() { strm.RemoveReader(r) })
+
+			baseNTP := time.Unix(1710000000, 0)
+			step := ca.step
+			const initialTimestamp = uint32(45343)
+
+			makeUnit := func(seq uint16, ntp time.Time) *unit.Unit {
+				return &unit.Unit{
+					PTS: 0,
+					NTP: ntp,
+					RTPPackets: []*rtp.Packet{{
+						Header: rtp.Header{
+							Version:        2,
+							Marker:         true,
+							PayloadType:    ca.payloadType,
+							SequenceNumber: seq,
+							Timestamp:      initialTimestamp,
+							SSRC:           563424,
+						},
+						Payload: append([]byte(nil), ca.payload...),
+					}},
+				}
+			}
+
+			// prime the pipeline to allow track gathering
+			subStream.WriteUnit(strm.OrigDesc.Medias[0], strm.OrigDesc.Medias[0].Formats[0],
+				makeUnit(1123, baseNTP))
+
+			err = pcReader.GatherInboundTracks(2 * time.Second)
+			require.NoError(t, err)
+
+			tracks := pcReader.InboundTracks()
+			require.Len(t, tracks, 1)
+
+			done := make(chan struct{})
+			errCh := make(chan string, 1)
+			const startSeq = uint16(2000)
+
+			var recvIndex int
+			var prevTS uint32
+			prevTSValid := false
+			sawTSDelta := false
+			sawNTP := false
+
+			tracks[0].OnPacketRTP = func(pkt *rtp.Packet) {
+				if prevTSValid {
+					if pkt.Timestamp-prevTS != ca.expectedTSDelta {
+						select {
+						case errCh <- fmt.Sprintf("timestamp delta mismatch for packet=%d: got=%d expected=%d",
+							recvIndex, pkt.Timestamp-prevTS, ca.expectedTSDelta):
+						default:
+						}
+						return
+					}
+					sawTSDelta = true
+				}
+				prevTS = pkt.Timestamp
+				prevTSValid = true
+
+				ntp, avail := tracks[0].PacketNTP(pkt)
+				if avail {
+					expected := baseNTP.Add(time.Duration(recvIndex) * step)
+					if ntp.Sub(expected).Abs() > 50*time.Millisecond {
+						select {
+						case errCh <- fmt.Sprintf("absolute NTP mismatch for packet=%d: got=%v expected=%v",
+							recvIndex, ntp, expected):
+						default:
+						}
+						return
+					}
+					sawNTP = true
+				}
+
+				recvIndex++
+
+				if sawTSDelta && sawNTP {
+					select {
+					case done <- struct{}{}:
+					default:
+					}
+				}
+			}
+
+			pcReader.StartReading()
+
+			go func() {
+				ticker := time.NewTicker(step)
+				defer ticker.Stop()
+
+				for i := range uint16(150) {
+					seq := startSeq + i
+					expected := baseNTP.Add(time.Duration(i) * step)
+
+					subStream.WriteUnit(strm.OrigDesc.Medias[0], strm.OrigDesc.Medias[0].Formats[0],
+						makeUnit(seq, expected))
+
+					<-ticker.C
+				}
+			}()
+
+			select {
+			case <-done:
+			case err := <-errCh:
+				t.Fatal(err)
+			case <-time.After(8 * time.Second):
+				t.Fatal("audio timestamp mapping did not become available")
+			}
+		})
 	}
 }
 
@@ -449,7 +384,7 @@ func TestFromStreamDoesNotMutateSharedRTPPackets(t *testing.T) {
 			err = subStream.Initialize()
 			require.NoError(t, err)
 
-			pcReader := &PeerConnection{
+			pcReader := &webrtc.PeerConnection{
 				LocalRandomUDP:    true,
 				IPsFromInterfaces: true,
 				Publish:           false,
@@ -459,7 +394,7 @@ func TestFromStreamDoesNotMutateSharedRTPPackets(t *testing.T) {
 			require.NoError(t, err)
 			t.Cleanup(pcReader.Close)
 
-			pcPublisher := &PeerConnection{
+			pcPublisher := &webrtc.PeerConnection{
 				LocalRandomUDP:    true,
 				IPsFromInterfaces: true,
 				Publish:           true,
@@ -468,7 +403,7 @@ func TestFromStreamDoesNotMutateSharedRTPPackets(t *testing.T) {
 
 			r := &stream.Reader{Parent: test.NilLogger}
 
-			err = FromStream(strm.OrigDesc, r, pcPublisher)
+			err = webrtc.FromStream(strm.OrigDesc, r, pcPublisher)
 			require.NoError(t, err)
 
 			err = pcPublisher.Start()
@@ -477,7 +412,6 @@ func TestFromStreamDoesNotMutateSharedRTPPackets(t *testing.T) {
 
 			const originalSSRC = uint32(563424)
 			require.NotEmpty(t, pcPublisher.OutboundTracks)
-			require.NotEqual(t, originalSSRC, pcPublisher.OutboundTracks[0].ssrc)
 
 			offer, err := pcReader.CreatePartialOffer(false)
 			require.NoError(t, err)

internal/protocols/webrtc/to_stream_test.go

@@ -1,22 +1,25 @@
-package webrtc
+package webrtc_test
 
 import (
 	"testing"
 	"time"
 
+	"github.com/bluenviron/mediamtx/internal/protocols/webrtc"
+
+	pwebrtc "github.com/pion/webrtc/v4"
+
 	"github.com/bluenviron/gortsplib/v5/pkg/format"
 	"github.com/bluenviron/mediamtx/internal/conf"
 	"github.com/bluenviron/mediamtx/internal/stream"
 	"github.com/bluenviron/mediamtx/internal/test"
 	"github.com/pion/rtp"
-	"github.com/pion/webrtc/v4"
 	"github.com/stretchr/testify/require"
 )
 
 func TestToStreamNoSupportedCodecs(t *testing.T) {
-	pc := &PeerConnection{}
-	_, err := ToStream(pc, &conf.Path{}, nil, nil)
-	require.Equal(t, errNoSupportedCodecsTo, err)
+	pc := &webrtc.PeerConnection{}
+	_, err := webrtc.ToStream(pc, &conf.Path{}, nil, nil)
+	require.ErrorContains(t, err, "the stream doesn't contain any supported codec")
 }
 
 // this is impossible to test since unsupported tracks cause an error
@@ -26,7 +29,7 @@ func TestToStreamNoSupportedCodecs(t *testing.T) {
 var toFromStreamCases = []struct {
 	name       string
 	in         format.Format
-	webrtcCaps webrtc.RTPCodecCapability
+	webrtcCaps pwebrtc.RTPCodecCapability
 	out        format.Format
 }{
 	{
@@ -34,7 +37,7 @@ var toFromStreamCases = []struct {
 		&format.AV1{
 			PayloadTyp: 96,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "video/AV1",
 			ClockRate: 90000,
 		},
@@ -47,7 +50,7 @@ var toFromStreamCases = []struct {
 		&format.VP9{
 			PayloadTyp: 96,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:    "video/VP9",
 			ClockRate:   90000,
 			SDPFmtpLine: "profile-id=0",
@@ -61,7 +64,7 @@ var toFromStreamCases = []struct {
 		&format.VP8{
 			PayloadTyp: 96,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "video/VP8",
 			ClockRate: 90000,
 		},
@@ -74,7 +77,7 @@ var toFromStreamCases = []struct {
 		&format.H265{
 			PayloadTyp: 96,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:    "video/H265",
 			ClockRate:   90000,
 			SDPFmtpLine: "level-id=93;profile-id=1;tier-flag=0;tx-mode=SRST",
@@ -86,7 +89,7 @@ var toFromStreamCases = []struct {
 	{
 		"h264",
 		test.FormatH264,
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:    "video/H264",
 			ClockRate:   90000,
 			SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f",
@@ -102,7 +105,7 @@ var toFromStreamCases = []struct {
 			PayloadTyp:   96,
 			ChannelCount: 6,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:    "audio/multiopus",
 			ClockRate:   48000,
 			Channels:    6,
@@ -119,7 +122,7 @@ var toFromStreamCases = []struct {
 			PayloadTyp:   96,
 			ChannelCount: 2,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:    "audio/opus",
 			ClockRate:   48000,
 			Channels:    2,
@@ -136,7 +139,7 @@ var toFromStreamCases = []struct {
 			PayloadTyp:   96,
 			ChannelCount: 1,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:    "audio/opus",
 			ClockRate:   48000,
 			Channels:    2,
@@ -150,7 +153,7 @@ var toFromStreamCases = []struct {
 	{
 		"g722",
 		&format.G722{},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "audio/G722",
 			ClockRate: 8000,
 		},
@@ -163,7 +166,7 @@ var toFromStreamCases = []struct {
 			SampleRate:   8000,
 			ChannelCount: 1,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "audio/PCMA",
 			ClockRate: 8000,
 		},
@@ -181,7 +184,7 @@ var toFromStreamCases = []struct {
 			SampleRate:   8000,
 			ChannelCount: 1,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "audio/PCMU",
 			ClockRate: 8000,
 		},
@@ -199,7 +202,7 @@ var toFromStreamCases = []struct {
 			SampleRate:   8000,
 			ChannelCount: 2,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "audio/PCMA",
 			ClockRate: 8000,
 			Channels:  2,
@@ -218,7 +221,7 @@ var toFromStreamCases = []struct {
 			SampleRate:   8000,
 			ChannelCount: 2,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "audio/PCMU",
 			ClockRate: 8000,
 			Channels:  2,
@@ -237,7 +240,7 @@ var toFromStreamCases = []struct {
 			SampleRate:   16000,
 			ChannelCount: 2,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "audio/L16",
 			ClockRate: 16000,
 			Channels:  2,
@@ -257,7 +260,7 @@ var toFromStreamCases = []struct {
 			SampleRate:   16000,
 			ChannelCount: 2,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "audio/L16",
 			ClockRate: 16000,
 			Channels:  2,
@@ -277,7 +280,7 @@ var toFromStreamCases = []struct {
 			SampleRate:   8000,
 			ChannelCount: 2,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "audio/L16",
 			ClockRate: 8000,
 			Channels:  2,
@@ -297,7 +300,7 @@ var toFromStreamCases = []struct {
 			SampleRate:   16000,
 			ChannelCount: 2,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "audio/L16",
 			ClockRate: 16000,
 			Channels:  2,
@@ -317,7 +320,7 @@ var toFromStreamCases = []struct {
 			SampleRate:   48000,
 			ChannelCount: 2,
 		},
-		webrtc.RTPCodecCapability{
+		pwebrtc.RTPCodecCapability{
 			MimeType:  "audio/L16",
 			ClockRate: 48000,
 			Channels:  2,
@@ -334,11 +337,11 @@ var toFromStreamCases = []struct {
 func TestToStream(t *testing.T) {
 	for _, ca := range toFromStreamCases {
 		t.Run(ca.name, func(t *testing.T) {
-			pc1 := &PeerConnection{
+			pc1 := &webrtc.PeerConnection{
 				LocalRandomUDP:    true,
 				IPsFromInterfaces: true,
 				Publish:           true,
-				OutboundTracks: []*OutboundTrack{{
+				OutboundTracks: []*webrtc.OutboundTrack{{
 					Caps: ca.webrtcCaps,
 				}},
 				Log: test.NilLogger,
@@ -347,7 +350,7 @@ func TestToStream(t *testing.T) {
 			require.NoError(t, err)
 			defer pc1.Close()
 
-			pc2 := &PeerConnection{
+			pc2 := &webrtc.PeerConnection{
 				LocalRandomUDP:    true,
 				IPsFromInterfaces: true,
 				Publish:           false,
@@ -402,7 +405,7 @@ func TestToStream(t *testing.T) {
 			require.NoError(t, err)
 
 			var subStream *stream.SubStream
-			medias, err := ToStream(pc2, &conf.Path{}, &subStream, nil)
+			medias, err := webrtc.ToStream(pc2, &conf.Path{}, &subStream, nil)
 			require.NoError(t, err)
 			require.Equal(t, ca.out, medias[0].Formats[0])
 		})