diff --git a/internal/protocols/webrtc/from_stream.go b/internal/protocols/webrtc/from_stream.go index f0b541f3..f127d98d 100644 --- a/internal/protocols/webrtc/from_stream.go +++ b/internal/protocols/webrtc/from_stream.go @@ -373,7 +373,10 @@ func setupAudioTrack( media, opusFormat, func(u *unit.Unit) error { + baseTimestamp := curTimestamp + for _, orig := range u.RTPPackets { + // create a copy of the packet that we can edit freely pkt := &rtp.Packet{ Header: orig.Header, Payload: orig.Payload, @@ -384,7 +387,7 @@ func setupAudioTrack( pkt.Timestamp = curTimestamp curTimestamp += uint32(opus.PacketDuration2(pkt.Payload)) - ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-u.RTPPackets[0].Timestamp), 48000)) + ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), 48000)) track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck } @@ -487,7 +490,10 @@ func setupAudioTrack( media, g711Format, func(u *unit.Unit) error { + baseTimestamp := curTimestamp + for _, orig := range u.RTPPackets { + // create a copy of the packet that we can edit freely pkt := &rtp.Packet{ Header: orig.Header, Payload: orig.Payload, @@ -498,7 +504,7 @@ func setupAudioTrack( pkt.Timestamp = curTimestamp curTimestamp += uint32(len(pkt.Payload)) / uint32(g711Format.ChannelCount) - ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-u.RTPPackets[0].Timestamp), 8000)) + ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), 8000)) track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck } @@ -545,14 +551,15 @@ func setupAudioTrack( return nil //nolint:nilerr } + baseTimestamp := curTimestamp + 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) - ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-u.RTPPackets[0].Timestamp), - g711Format.ClockRate())) + ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), g711Format.ClockRate())) track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck } @@ -619,14 +626,15 @@ func setupAudioTrack( return nil //nolint:nilerr } + baseTimestamp := curTimestamp + 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) - ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-u.RTPPackets[0].Timestamp), - lpcmFormat.ClockRate())) + ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), lpcmFormat.ClockRate())) track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck } diff --git a/internal/protocols/webrtc/from_stream_test.go b/internal/protocols/webrtc/from_stream_test.go index 7f40d956..48f5f23f 100644 --- a/internal/protocols/webrtc/from_stream_test.go +++ b/internal/protocols/webrtc/from_stream_test.go @@ -212,3 +212,174 @@ func TestFromStreamResampleOpus(t *testing.T) { <-done } + +func TestFromStreamResampleOpusAbsoluteTimestamp(t *testing.T) { + strm := &stream.Stream{ + Desc: &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeAudio, + Formats: []format.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) + + 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.Desc, r, pcPublisher) + require.NoError(t, err) + + err = pcPublisher.Start() + require.NoError(t, err) + t.Cleanup(pcPublisher.Close) + + offer, err := pcReader.CreatePartialOffer() + require.NoError(t, err) + + answer, err := pcPublisher.CreateFullAnswer(offer) + 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.Desc.Medias[0], strm.Desc.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.GatherIncomingTracks(2 * time.Second) + require.NoError(t, err) + + tracks := pcReader.IncomingTracks() + 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: + } + return + } + + 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, _ := expectedNTP(seq) + + subStream.WriteUnit(strm.Desc.Medias[0], strm.Desc.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}, + }}, + }) + + <-ticker.C + } + }() + + select { + case <-done: + case err := <-errCh: + t.Fatal(err) + case <-time.After(8 * time.Second): + t.Fatal("absolute timestamp mapping did not become available") + } +}