diff --git a/internal/core/api_test.go b/internal/core/api_test.go index 3390727c..a54ec12c 100644 --- a/internal/core/api_test.go +++ b/internal/core/api_test.go @@ -21,10 +21,12 @@ import ( srt "github.com/datarhei/gosrt" "github.com/google/uuid" "github.com/pion/rtp" + pwebrtc "github.com/pion/webrtc/v3" "github.com/stretchr/testify/require" "github.com/bluenviron/mediamtx/internal/protocols/rtmp" "github.com/bluenviron/mediamtx/internal/protocols/webrtc" + "github.com/bluenviron/mediamtx/internal/protocols/whip" "github.com/bluenviron/mediamtx/internal/test" ) @@ -520,7 +522,7 @@ func TestAPIProtocolListGet(t *testing.T) { require.NoError(t, err2) }() - c := &webrtc.WHIPClient{ + c := &whip.Client{ HTTPClient: hc, URL: u, Log: test.NilLogger, @@ -993,13 +995,21 @@ func TestAPIProtocolKick(t *testing.T) { u, err := url.Parse("http://localhost:8889/mypath/whip") require.NoError(t, err) - c := &webrtc.WHIPClient{ + c := &whip.Client{ HTTPClient: hc, URL: u, Log: test.NilLogger, } - _, err = c.Publish(context.Background(), medi.Formats[0], nil) + track := &webrtc.OutgoingTrack{ + Caps: pwebrtc.RTPCodecCapability{ + MimeType: pwebrtc.MimeTypeH264, + ClockRate: 90000, + SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", + }, + } + + err = c.Publish(context.Background(), []*webrtc.OutgoingTrack{track}) require.NoError(t, err) defer func() { require.Error(t, c.Close()) diff --git a/internal/core/metrics_test.go b/internal/core/metrics_test.go index e586c840..bb6eca63 100644 --- a/internal/core/metrics_test.go +++ b/internal/core/metrics_test.go @@ -18,10 +18,12 @@ import ( "github.com/bluenviron/mediacommon/pkg/formats/mpegts" srt "github.com/datarhei/gosrt" "github.com/pion/rtp" + pwebrtc "github.com/pion/webrtc/v3" "github.com/stretchr/testify/require" "github.com/bluenviron/mediamtx/internal/protocols/rtmp" "github.com/bluenviron/mediamtx/internal/protocols/webrtc" + "github.com/bluenviron/mediamtx/internal/protocols/whip" "github.com/bluenviron/mediamtx/internal/test" ) @@ -171,17 +173,25 @@ webrtc_sessions_bytes_sent 0 defer tr.CloseIdleConnections() hc2 := &http.Client{Transport: tr} - s := &webrtc.WHIPClient{ + s := &whip.Client{ HTTPClient: hc2, URL: su, Log: test.NilLogger, } - tracks, err := s.Publish(context.Background(), test.MediaH264.Formats[0], nil) + track := &webrtc.OutgoingTrack{ + Caps: pwebrtc.RTPCodecCapability{ + MimeType: pwebrtc.MimeTypeH264, + ClockRate: 90000, + SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", + }, + } + + err = s.Publish(context.Background(), []*webrtc.OutgoingTrack{track}) require.NoError(t, err) defer checkClose(t, s.Close) - err = tracks[0].WriteRTP(&rtp.Packet{ + err = track.WriteRTP(&rtp.Packet{ Header: rtp.Header{ Version: 2, Marker: true, diff --git a/internal/core/path_test.go b/internal/core/path_test.go index b473a801..65030e03 100644 --- a/internal/core/path_test.go +++ b/internal/core/path_test.go @@ -24,7 +24,7 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/protocols/rtmp" - "github.com/bluenviron/mediamtx/internal/protocols/webrtc" + "github.com/bluenviron/mediamtx/internal/protocols/whip" "github.com/bluenviron/mediamtx/internal/test" ) @@ -394,7 +394,7 @@ func TestPathRunOnRead(t *testing.T) { u, err := url.Parse("http://localhost:8889/test/whep?query=value") require.NoError(t, err) - c := &webrtc.WHIPClient{ + c := &whip.Client{ HTTPClient: hc, URL: u, Log: test.NilLogger, diff --git a/internal/protocols/hls/from_stream.go b/internal/protocols/hls/from_stream.go new file mode 100644 index 00000000..f362563f --- /dev/null +++ b/internal/protocols/hls/from_stream.go @@ -0,0 +1,232 @@ +// Package hls contains HLS utilities. +package hls + +import ( + "errors" + "fmt" + + "github.com/bluenviron/gohlslib" + "github.com/bluenviron/gohlslib/pkg/codecs" + "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/mediamtx/internal/asyncwriter" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/unit" +) + +// ErrNoSupportedCodecs is returned by FromStream when there are no supported codecs. +var ErrNoSupportedCodecs = errors.New( + "the stream doesn't contain any supported codec, which are currently H265, H264, Opus, MPEG-4 Audio") + +func setupVideoTrack( + stream *stream.Stream, + writer *asyncwriter.Writer, + muxer *gohlslib.Muxer, +) *gohlslib.Track { + var videoFormatAV1 *format.AV1 + videoMedia := stream.Desc().FindFormat(&videoFormatAV1) + + if videoFormatAV1 != nil { + stream.AddReader(writer, videoMedia, videoFormatAV1, func(u unit.Unit) error { + tunit := u.(*unit.AV1) + + if tunit.TU == nil { + return nil + } + + err := muxer.WriteAV1(tunit.NTP, tunit.PTS, tunit.TU) + if err != nil { + return fmt.Errorf("muxer error: %w", err) + } + + return nil + }) + + return &gohlslib.Track{ + Codec: &codecs.AV1{}, + } + } + + var videoFormatVP9 *format.VP9 + videoMedia = stream.Desc().FindFormat(&videoFormatVP9) + + if videoFormatVP9 != nil { + stream.AddReader(writer, videoMedia, videoFormatVP9, func(u unit.Unit) error { + tunit := u.(*unit.VP9) + + if tunit.Frame == nil { + return nil + } + + err := muxer.WriteVP9(tunit.NTP, tunit.PTS, tunit.Frame) + if err != nil { + return fmt.Errorf("muxer error: %w", err) + } + + return nil + }) + + return &gohlslib.Track{ + Codec: &codecs.VP9{}, + } + } + + var videoFormatH265 *format.H265 + videoMedia = stream.Desc().FindFormat(&videoFormatH265) + + if videoFormatH265 != nil { + stream.AddReader(writer, videoMedia, videoFormatH265, func(u unit.Unit) error { + tunit := u.(*unit.H265) + + if tunit.AU == nil { + return nil + } + + err := muxer.WriteH265(tunit.NTP, tunit.PTS, tunit.AU) + if err != nil { + return fmt.Errorf("muxer error: %w", err) + } + + return nil + }) + + vps, sps, pps := videoFormatH265.SafeParams() + + return &gohlslib.Track{ + Codec: &codecs.H265{ + VPS: vps, + SPS: sps, + PPS: pps, + }, + } + } + + var videoFormatH264 *format.H264 + videoMedia = stream.Desc().FindFormat(&videoFormatH264) + + if videoFormatH264 != nil { + stream.AddReader(writer, videoMedia, videoFormatH264, func(u unit.Unit) error { + tunit := u.(*unit.H264) + + if tunit.AU == nil { + return nil + } + + err := muxer.WriteH264(tunit.NTP, tunit.PTS, tunit.AU) + if err != nil { + return fmt.Errorf("muxer error: %w", err) + } + + return nil + }) + + sps, pps := videoFormatH264.SafeParams() + + return &gohlslib.Track{ + Codec: &codecs.H264{ + SPS: sps, + PPS: pps, + }, + } + } + + return nil +} + +func setupAudioTrack( + stream *stream.Stream, + writer *asyncwriter.Writer, + muxer *gohlslib.Muxer, + l logger.Writer, +) *gohlslib.Track { + var audioFormatOpus *format.Opus + audioMedia := stream.Desc().FindFormat(&audioFormatOpus) + + if audioMedia != nil { + stream.AddReader(writer, audioMedia, audioFormatOpus, func(u unit.Unit) error { + tunit := u.(*unit.Opus) + + err := muxer.WriteOpus( + tunit.NTP, + tunit.PTS, + tunit.Packets) + if err != nil { + return fmt.Errorf("muxer error: %w", err) + } + + return nil + }) + + return &gohlslib.Track{ + Codec: &codecs.Opus{ + ChannelCount: audioFormatOpus.ChannelCount, + }, + } + } + + var audioFormatMPEG4Audio *format.MPEG4Audio + audioMedia = stream.Desc().FindFormat(&audioFormatMPEG4Audio) + + if audioMedia != nil { + co := audioFormatMPEG4Audio.GetConfig() + if co == nil { + l.Log(logger.Warn, "skipping MPEG-4 audio track: tracks without explicit configuration are not supported") + } else { + stream.AddReader(writer, audioMedia, audioFormatMPEG4Audio, func(u unit.Unit) error { + tunit := u.(*unit.MPEG4Audio) + + if tunit.AUs == nil { + return nil + } + + err := muxer.WriteMPEG4Audio( + tunit.NTP, + tunit.PTS, + tunit.AUs) + if err != nil { + return fmt.Errorf("muxer error: %w", err) + } + + return nil + }) + + return &gohlslib.Track{ + Codec: &codecs.MPEG4Audio{ + Config: *co, + }, + } + } + } + + return nil +} + +// FromStream maps a MediaMTX stream to a HLS muxer. +func FromStream( + stream *stream.Stream, + writer *asyncwriter.Writer, + muxer *gohlslib.Muxer, + l logger.Writer, +) error { + videoTrack := setupVideoTrack( + stream, + writer, + muxer, + ) + + audioTrack := setupAudioTrack( + stream, + writer, + muxer, + l, + ) + + if videoTrack == nil && audioTrack == nil { + return ErrNoSupportedCodecs + } + + muxer.VideoTrack = videoTrack + muxer.AudioTrack = audioTrack + + return nil +} diff --git a/internal/protocols/hls/to_stream.go b/internal/protocols/hls/to_stream.go new file mode 100644 index 00000000..4e993454 --- /dev/null +++ b/internal/protocols/hls/to_stream.go @@ -0,0 +1,154 @@ +package hls + +import ( + "fmt" + "time" + + "github.com/bluenviron/gohlslib" + "github.com/bluenviron/gohlslib/pkg/codecs" + "github.com/bluenviron/gortsplib/v4/pkg/description" + "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/unit" +) + +// ToStream maps a HLS stream to a MediaMTX stream. +func ToStream( + c *gohlslib.Client, + tracks []*gohlslib.Track, + stream **stream.Stream, +) ([]*description.Media, error) { + var medias []*description.Media //nolint:prealloc + + for _, track := range tracks { + var medi *description.Media + + switch tcodec := track.Codec.(type) { + case *codecs.AV1: + medi = &description.Media{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.AV1{ + PayloadTyp: 96, + }}, + } + + c.OnDataAV1(track, func(pts time.Duration, tu [][]byte) { + (*stream).WriteUnit(medi, medi.Formats[0], &unit.AV1{ + Base: unit.Base{ + NTP: time.Now(), + PTS: pts, + }, + TU: tu, + }) + }) + + case *codecs.VP9: + medi = &description.Media{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP9{ + PayloadTyp: 96, + }}, + } + + c.OnDataVP9(track, func(pts time.Duration, frame []byte) { + (*stream).WriteUnit(medi, medi.Formats[0], &unit.VP9{ + Base: unit.Base{ + NTP: time.Now(), + PTS: pts, + }, + Frame: frame, + }) + }) + + case *codecs.H264: + medi = &description.Media{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H264{ + PayloadTyp: 96, + PacketizationMode: 1, + SPS: tcodec.SPS, + PPS: tcodec.PPS, + }}, + } + + c.OnDataH26x(track, func(pts time.Duration, _ time.Duration, au [][]byte) { + (*stream).WriteUnit(medi, medi.Formats[0], &unit.H264{ + Base: unit.Base{ + NTP: time.Now(), + PTS: pts, + }, + AU: au, + }) + }) + + case *codecs.H265: + medi = &description.Media{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H265{ + PayloadTyp: 96, + VPS: tcodec.VPS, + SPS: tcodec.SPS, + PPS: tcodec.PPS, + }}, + } + + c.OnDataH26x(track, func(pts time.Duration, _ time.Duration, au [][]byte) { + (*stream).WriteUnit(medi, medi.Formats[0], &unit.H265{ + Base: unit.Base{ + NTP: time.Now(), + PTS: pts, + }, + AU: au, + }) + }) + + case *codecs.MPEG4Audio: + medi = &description.Media{ + Type: description.MediaTypeAudio, + Formats: []format.Format{&format.MPEG4Audio{ + PayloadTyp: 96, + SizeLength: 13, + IndexLength: 3, + IndexDeltaLength: 3, + Config: &tcodec.Config, + }}, + } + + c.OnDataMPEG4Audio(track, func(pts time.Duration, aus [][]byte) { + (*stream).WriteUnit(medi, medi.Formats[0], &unit.MPEG4Audio{ + Base: unit.Base{ + NTP: time.Now(), + PTS: pts, + }, + AUs: aus, + }) + }) + + case *codecs.Opus: + medi = &description.Media{ + Type: description.MediaTypeAudio, + Formats: []format.Format{&format.Opus{ + PayloadTyp: 96, + ChannelCount: tcodec.ChannelCount, + }}, + } + + c.OnDataOpus(track, func(pts time.Duration, packets [][]byte) { + (*stream).WriteUnit(medi, medi.Formats[0], &unit.Opus{ + Base: unit.Base{ + NTP: time.Now(), + PTS: pts, + }, + Packets: packets, + }) + }) + + default: + return nil, fmt.Errorf("unsupported track: %T", track.Codec) + } + + medias = append(medias, medi) + } + + return medias, nil +} diff --git a/internal/protocols/webrtc/from_stream.go b/internal/protocols/webrtc/from_stream.go new file mode 100644 index 00000000..0b87d099 --- /dev/null +++ b/internal/protocols/webrtc/from_stream.go @@ -0,0 +1,539 @@ +package webrtc + +import ( + "crypto/rand" + "errors" + "fmt" + "time" + + "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/gortsplib/v4/pkg/format/rtpav1" + "github.com/bluenviron/gortsplib/v4/pkg/format/rtph264" + "github.com/bluenviron/gortsplib/v4/pkg/format/rtplpcm" + "github.com/bluenviron/gortsplib/v4/pkg/format/rtpvp8" + "github.com/bluenviron/gortsplib/v4/pkg/format/rtpvp9" + "github.com/bluenviron/mediacommon/pkg/codecs/g711" + "github.com/bluenviron/mediamtx/internal/asyncwriter" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/unit" + "github.com/pion/webrtc/v3" +) + +const ( + webrtcPayloadMaxSize = 1188 // 1200 - 12 (RTP header) +) + +var errNoSupportedCodecs = errors.New( + "the stream doesn't contain any supported codec, which are currently AV1, VP9, VP8, H264, Opus, G722, G711, LPCM") + +func uint16Ptr(v uint16) *uint16 { + return &v +} + +func randUint32() (uint32, error) { + var b [4]byte + _, err := rand.Read(b[:]) + if err != nil { + return 0, err + } + return uint32(b[0])<<24 | uint32(b[1])<<16 | uint32(b[2])<<8 | uint32(b[3]), nil +} + +func setupVideoTrack( + stream *stream.Stream, + writer *asyncwriter.Writer, + pc *PeerConnection, +) (format.Format, error) { + var av1Format *format.AV1 + media := stream.Desc().FindFormat(&av1Format) + + if av1Format != nil { + track := &OutgoingTrack{ + Caps: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeAV1, + ClockRate: 90000, + }, + } + pc.OutgoingTracks = append(pc.OutgoingTracks, track) + + encoder := &rtpav1.Encoder{ + PayloadType: 105, + PayloadMaxSize: webrtcPayloadMaxSize, + } + err := encoder.Init() + if err != nil { + return nil, err + } + + stream.AddReader(writer, media, av1Format, func(u unit.Unit) error { + tunit := u.(*unit.AV1) + + if tunit.TU == nil { + return nil + } + + packets, err := encoder.Encode(tunit.TU) + if err != nil { + return nil //nolint:nilerr + } + + for _, pkt := range packets { + pkt.Timestamp += tunit.RTPPackets[0].Timestamp + track.WriteRTP(pkt) //nolint:errcheck + } + + return nil + }) + + return av1Format, nil + } + + var vp9Format *format.VP9 + media = stream.Desc().FindFormat(&vp9Format) + + if vp9Format != nil { + track := &OutgoingTrack{ + Caps: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeVP9, + ClockRate: 90000, + SDPFmtpLine: "profile-id=0", + }, + } + pc.OutgoingTracks = append(pc.OutgoingTracks, track) + + encoder := &rtpvp9.Encoder{ + PayloadType: 96, + PayloadMaxSize: webrtcPayloadMaxSize, + InitialPictureID: uint16Ptr(8445), + } + err := encoder.Init() + if err != nil { + return nil, err + } + + stream.AddReader(writer, media, vp9Format, func(u unit.Unit) error { + tunit := u.(*unit.VP9) + + if tunit.Frame == nil { + return nil + } + + packets, err := encoder.Encode(tunit.Frame) + if err != nil { + return nil //nolint:nilerr + } + + for _, pkt := range packets { + pkt.Timestamp += tunit.RTPPackets[0].Timestamp + track.WriteRTP(pkt) //nolint:errcheck + } + + return nil + }) + + return vp9Format, nil + } + + var vp8Format *format.VP8 + media = stream.Desc().FindFormat(&vp8Format) + + if vp8Format != nil { + track := &OutgoingTrack{ + Caps: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeVP8, + ClockRate: 90000, + }, + } + pc.OutgoingTracks = append(pc.OutgoingTracks, track) + + encoder := &rtpvp8.Encoder{ + PayloadType: 96, + PayloadMaxSize: webrtcPayloadMaxSize, + } + err := encoder.Init() + if err != nil { + return nil, err + } + + stream.AddReader(writer, media, vp8Format, func(u unit.Unit) error { + tunit := u.(*unit.VP8) + + if tunit.Frame == nil { + return nil + } + + packets, err := encoder.Encode(tunit.Frame) + if err != nil { + return nil //nolint:nilerr + } + + for _, pkt := range packets { + pkt.Timestamp += tunit.RTPPackets[0].Timestamp + track.WriteRTP(pkt) //nolint:errcheck + } + + return nil + }) + + return vp8Format, nil + } + + var h264Format *format.H264 + media = stream.Desc().FindFormat(&h264Format) + + if h264Format != nil { + track := &OutgoingTrack{ + Caps: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeH264, + ClockRate: 90000, + SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", + }, + } + pc.OutgoingTracks = append(pc.OutgoingTracks, track) + + encoder := &rtph264.Encoder{ + PayloadType: 96, + PayloadMaxSize: webrtcPayloadMaxSize, + } + err := encoder.Init() + if err != nil { + return nil, err + } + + firstReceived := false + var lastPTS time.Duration + + stream.AddReader(writer, media, h264Format, func(u unit.Unit) error { + tunit := u.(*unit.H264) + + if tunit.AU == nil { + return nil + } + + if !firstReceived { + firstReceived = true + } else if tunit.PTS < lastPTS { + return fmt.Errorf("WebRTC doesn't support H264 streams with B-frames") + } + lastPTS = tunit.PTS + + packets, err := encoder.Encode(tunit.AU) + if err != nil { + return nil //nolint:nilerr + } + + for _, pkt := range packets { + pkt.Timestamp += tunit.RTPPackets[0].Timestamp + track.WriteRTP(pkt) //nolint:errcheck + } + + return nil + }) + + return h264Format, nil + } + + return nil, nil +} + +func setupAudioTrack( + stream *stream.Stream, + writer *asyncwriter.Writer, + pc *PeerConnection, +) (format.Format, error) { + var opusFormat *format.Opus + media := stream.Desc().FindFormat(&opusFormat) + + if opusFormat != nil { + var caps webrtc.RTPCodecCapability + + switch opusFormat.ChannelCount { + case 1, 2: + caps = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeOpus, + ClockRate: 48000, + Channels: 2, + SDPFmtpLine: func() string { + s := "minptime=10;useinbandfec=1" + if opusFormat.ChannelCount == 2 { + s += ";stereo=1;sprop-stereo=1" + } + return s + }(), + } + + case 3, 4, 5, 6, 7, 8: + caps = webrtc.RTPCodecCapability{ + MimeType: mimeTypeMultiopus, + ClockRate: 48000, + Channels: uint16(opusFormat.ChannelCount), + SDPFmtpLine: multichannelOpusSDP[opusFormat.ChannelCount], + } + + default: + return nil, fmt.Errorf("unsupported channel count: %d", opusFormat.ChannelCount) + } + + track := &OutgoingTrack{ + Caps: caps, + } + pc.OutgoingTracks = append(pc.OutgoingTracks, track) + + stream.AddReader(writer, media, opusFormat, func(u unit.Unit) error { + for _, pkt := range u.GetRTPPackets() { + track.WriteRTP(pkt) //nolint:errcheck + } + + return nil + }) + + return opusFormat, nil + } + + var g722Format *format.G722 + media = stream.Desc().FindFormat(&g722Format) + + if g722Format != nil { + track := &OutgoingTrack{ + Caps: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeG722, + ClockRate: 8000, + }, + } + pc.OutgoingTracks = append(pc.OutgoingTracks, track) + + stream.AddReader(writer, media, g722Format, func(u unit.Unit) error { + for _, pkt := range u.GetRTPPackets() { + track.WriteRTP(pkt) //nolint:errcheck + } + + return nil + }) + + return g722Format, nil + } + + var g711Format *format.G711 + media = stream.Desc().FindFormat(&g711Format) + + if g711Format != nil { + // These are the sample rates and channels supported by Chrome. + // Different sample rates and channels can be streamed too but we don't want compatibility issues. + // https://webrtc.googlesource.com/src/+/refs/heads/main/modules/audio_coding/codecs/pcm16b/audio_decoder_pcm16b.cc#23 + if g711Format.ClockRate() != 8000 && g711Format.ClockRate() != 16000 && + g711Format.ClockRate() != 32000 && g711Format.ClockRate() != 48000 { + return nil, fmt.Errorf("unsupported clock rate: %d", g711Format.ClockRate()) + } + if g711Format.ChannelCount != 1 && g711Format.ChannelCount != 2 { + return nil, fmt.Errorf("unsupported channel count: %d", g711Format.ChannelCount) + } + + var caps webrtc.RTPCodecCapability + + if g711Format.SampleRate == 8000 { + if g711Format.MULaw { + if g711Format.ChannelCount != 1 { + caps = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypePCMU, + ClockRate: uint32(g711Format.SampleRate), + Channels: uint16(g711Format.ChannelCount), + } + } else { + caps = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypePCMU, + ClockRate: 8000, + } + } + } else { + if g711Format.ChannelCount != 1 { + caps = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypePCMA, + ClockRate: uint32(g711Format.SampleRate), + Channels: uint16(g711Format.ChannelCount), + } + } else { + caps = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypePCMA, + ClockRate: 8000, + } + } + } + } else { + caps = webrtc.RTPCodecCapability{ + MimeType: mimeTypeL16, + ClockRate: uint32(g711Format.ClockRate()), + Channels: uint16(g711Format.ChannelCount), + } + } + + track := &OutgoingTrack{ + Caps: caps, + } + pc.OutgoingTracks = append(pc.OutgoingTracks, track) + + if g711Format.SampleRate == 8000 { + curTimestamp, err := randUint32() + if err != nil { + return nil, err + } + + stream.AddReader(writer, media, g711Format, func(u unit.Unit) error { + for _, pkt := range u.GetRTPPackets() { + // 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) + + track.WriteRTP(pkt) //nolint:errcheck + } + + return nil + }) + } else { + encoder := &rtplpcm.Encoder{ + PayloadType: 96, + PayloadMaxSize: webrtcPayloadMaxSize, + BitDepth: 16, + ChannelCount: g711Format.ChannelCount, + } + err := encoder.Init() + if err != nil { + return nil, err + } + + curTimestamp, err := randUint32() + if err != nil { + return nil, err + } + + stream.AddReader(writer, media, g711Format, func(u unit.Unit) error { + tunit := u.(*unit.G711) + + if tunit.Samples == nil { + return nil + } + + var lpcmSamples []byte + if g711Format.MULaw { + lpcmSamples = g711.DecodeMulaw(tunit.Samples) + } else { + lpcmSamples = g711.DecodeAlaw(tunit.Samples) + } + + packets, err := encoder.Encode(lpcmSamples) + if err != nil { + return nil //nolint:nilerr + } + + 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) + + track.WriteRTP(pkt) //nolint:errcheck + } + + return nil + }) + } + + return g711Format, nil + } + + var lpcmFormat *format.LPCM + media = stream.Desc().FindFormat(&lpcmFormat) + + if lpcmFormat != nil { + if lpcmFormat.BitDepth != 16 { + return nil, fmt.Errorf("unsupported LPCM bit depth: %d", lpcmFormat.BitDepth) + } + + // These are the sample rates and channels supported by Chrome. + // Different sample rates and channels can be streamed too but we don't want compatibility issues. + // https://webrtc.googlesource.com/src/+/refs/heads/main/modules/audio_coding/codecs/pcm16b/audio_decoder_pcm16b.cc#23 + if lpcmFormat.ClockRate() != 8000 && lpcmFormat.ClockRate() != 16000 && + lpcmFormat.ClockRate() != 32000 && lpcmFormat.ClockRate() != 48000 { + return nil, fmt.Errorf("unsupported clock rate: %d", lpcmFormat.ClockRate()) + } + if lpcmFormat.ChannelCount != 1 && lpcmFormat.ChannelCount != 2 { + return nil, fmt.Errorf("unsupported channel count: %d", lpcmFormat.ChannelCount) + } + + track := &OutgoingTrack{ + Caps: webrtc.RTPCodecCapability{ + MimeType: mimeTypeL16, + ClockRate: uint32(lpcmFormat.ClockRate()), + Channels: uint16(lpcmFormat.ChannelCount), + }, + } + pc.OutgoingTracks = append(pc.OutgoingTracks, track) + + encoder := &rtplpcm.Encoder{ + PayloadType: 96, + BitDepth: 16, + ChannelCount: lpcmFormat.ChannelCount, + PayloadMaxSize: webrtcPayloadMaxSize, + } + err := encoder.Init() + if err != nil { + return nil, err + } + + curTimestamp, err := randUint32() + if err != nil { + return nil, err + } + + stream.AddReader(writer, media, lpcmFormat, func(u unit.Unit) error { + tunit := u.(*unit.LPCM) + + if tunit.Samples == nil { + return nil + } + + packets, err := encoder.Encode(tunit.Samples) + if err != nil { + return nil //nolint:nilerr + } + + 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) + + track.WriteRTP(pkt) //nolint:errcheck + } + + return nil + }) + + return lpcmFormat, nil + } + + return nil, nil +} + +// FromStream maps a MediaMTX stream to a WebRTC connection +func FromStream( + stream *stream.Stream, + writer *asyncwriter.Writer, + pc *PeerConnection, +) error { + videoFormat, err := setupVideoTrack(stream, writer, pc) + if err != nil { + return err + } + + audioFormat, err := setupAudioTrack(stream, writer, pc) + if err != nil { + return err + } + + if videoFormat == nil && audioFormat == nil { + return errNoSupportedCodecs + } + + return nil +} diff --git a/internal/protocols/webrtc/from_stream_test.go b/internal/protocols/webrtc/from_stream_test.go new file mode 100644 index 00000000..50a16024 --- /dev/null +++ b/internal/protocols/webrtc/from_stream_test.go @@ -0,0 +1,45 @@ +package webrtc + +import ( + "testing" + + "github.com/bluenviron/gortsplib/v4/pkg/description" + "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/mediamtx/internal/asyncwriter" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/test" + "github.com/stretchr/testify/require" +) + +func TestFromStream(t *testing.T) { + for _, ca := range toFromStreamCases { + if ca.in == nil { + continue + } + t.Run(ca.name, func(t *testing.T) { + desc := &description.Session{ + Medias: []*description.Media{{ + Formats: []format.Format{ca.in}, + }}, + } + + stream, err := stream.New( + 1460, + desc, + false, + test.NilLogger, + ) + require.NoError(t, err) + defer stream.Close() + + writer := asyncwriter.New(0, nil) + + pc := &PeerConnection{} + + err = FromStream(stream, writer, pc) + require.NoError(t, err) + + require.Equal(t, ca.webrtcCaps, pc.OutgoingTracks[0].Caps) + }) + } +} diff --git a/internal/protocols/webrtc/incoming_track.go b/internal/protocols/webrtc/incoming_track.go index 71016e96..e8f7f3cf 100644 --- a/internal/protocols/webrtc/incoming_track.go +++ b/internal/protocols/webrtc/incoming_track.go @@ -1,12 +1,8 @@ package webrtc import ( - "fmt" - "strings" "time" - "github.com/bluenviron/gortsplib/v4/pkg/description" - "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/gortsplib/v4/pkg/liberrors" "github.com/bluenviron/gortsplib/v4/pkg/rtpreorderer" "github.com/pion/rtcp" @@ -17,10 +13,7 @@ import ( ) const ( - keyFrameInterval = 2 * time.Second -) - -const ( + keyFrameInterval = 2 * time.Second mimeTypeMultiopus = "audio/multiopus" mimeTypeL16 = "audio/L16" ) @@ -234,140 +227,34 @@ var incomingAudioCodecs = []webrtc.RTPCodecParameters{ // IncomingTrack is an incoming track. type IncomingTrack struct { - track *webrtc.TrackRemote - log logger.Writer + OnPacketRTP func(*rtp.Packet) - typ description.MediaType - format format.Format - reorderer *rtpreorderer.Reorderer - pkts []*rtp.Packet + track *webrtc.TrackRemote + receiver *webrtc.RTPReceiver + writeRTCP func([]rtcp.Packet) error + log logger.Writer } -func newIncomingTrack( - track *webrtc.TrackRemote, - receiver *webrtc.RTPReceiver, - writeRTCP func([]rtcp.Packet) error, - log logger.Writer, -) (*IncomingTrack, error) { - t := &IncomingTrack{ - track: track, - log: log, - reorderer: rtpreorderer.New(), - } +func (t *IncomingTrack) initialize() { + t.OnPacketRTP = func(*rtp.Packet) {} +} - switch strings.ToLower(track.Codec().MimeType) { - case strings.ToLower(webrtc.MimeTypeAV1): - t.typ = description.MediaTypeVideo - t.format = &format.AV1{ - PayloadTyp: uint8(track.PayloadType()), - } +// ClockRate returns the clock rate. Needed by rtptime.GlobalDecoder +func (t *IncomingTrack) ClockRate() int { + return int(t.track.Codec().ClockRate) +} - case strings.ToLower(webrtc.MimeTypeVP9): - t.typ = description.MediaTypeVideo - t.format = &format.VP9{ - PayloadTyp: uint8(track.PayloadType()), - } - - case strings.ToLower(webrtc.MimeTypeVP8): - t.typ = description.MediaTypeVideo - t.format = &format.VP8{ - PayloadTyp: uint8(track.PayloadType()), - } - - case strings.ToLower(webrtc.MimeTypeH265): - t.typ = description.MediaTypeVideo - t.format = &format.H265{ - PayloadTyp: uint8(track.PayloadType()), - } - - case strings.ToLower(webrtc.MimeTypeH264): - t.typ = description.MediaTypeVideo - t.format = &format.H264{ - PayloadTyp: uint8(track.PayloadType()), - PacketizationMode: 1, - } - - case strings.ToLower(mimeTypeMultiopus): - t.typ = description.MediaTypeAudio - t.format = &format.Opus{ - PayloadTyp: uint8(track.PayloadType()), - ChannelCount: int(track.Codec().Channels), - } - - case strings.ToLower(webrtc.MimeTypeOpus): - t.typ = description.MediaTypeAudio - t.format = &format.Opus{ - PayloadTyp: uint8(track.PayloadType()), - ChannelCount: func() int { - if strings.Contains(track.Codec().SDPFmtpLine, "stereo=1") { - return 2 - } - return 1 - }(), - } - - case strings.ToLower(webrtc.MimeTypeG722): - t.typ = description.MediaTypeAudio - t.format = &format.G722{} - - case strings.ToLower(webrtc.MimeTypePCMU): - t.typ = description.MediaTypeAudio - - channels := track.Codec().Channels - if channels == 0 { - channels = 1 - } - - payloadType := uint8(0) - if channels > 1 { - payloadType = 118 - } - - t.format = &format.G711{ - PayloadTyp: payloadType, - MULaw: true, - SampleRate: 8000, - ChannelCount: int(channels), - } - - case strings.ToLower(webrtc.MimeTypePCMA): - t.typ = description.MediaTypeAudio - - channels := track.Codec().Channels - if channels == 0 { - channels = 1 - } - - payloadType := uint8(8) - if channels > 1 { - payloadType = 119 - } - - t.format = &format.G711{ - PayloadTyp: payloadType, - MULaw: false, - SampleRate: 8000, - ChannelCount: int(channels), - } - - case strings.ToLower(mimeTypeL16): - t.typ = description.MediaTypeAudio - t.format = &format.LPCM{ - PayloadTyp: uint8(track.PayloadType()), - BitDepth: 16, - SampleRate: int(track.Codec().ClockRate), - ChannelCount: int(track.Codec().Channels), - } - - default: - return nil, fmt.Errorf("unsupported codec: %+v", track.Codec().RTPCodecCapability) - } +// PTSEqualsDTS returns whether PTS equals DTS. Needed by rtptime.GlobalDecoder +func (*IncomingTrack) PTSEqualsDTS(*rtp.Packet) bool { + return true +} +func (t *IncomingTrack) start() { // read incoming RTCP packets to make interceptors work go func() { buf := make([]byte, 1500) for { - _, _, err := receiver.Read(buf) + _, _, err := t.receiver.Read(buf) if err != nil { return } @@ -375,13 +262,13 @@ func newIncomingTrack( }() // send period key frame requests - if t.typ == description.MediaTypeVideo { + if t.track.Kind() == webrtc.RTPCodecTypeVideo { go func() { keyframeTicker := time.NewTicker(keyFrameInterval) defer keyframeTicker.Stop() for range keyframeTicker.C { - err := writeRTCP([]rtcp.Packet{ + err := t.writeRTCP([]rtcp.Packet{ &rtcp.PictureLossIndication{ MediaSSRC: uint32(t.track.SSRC()), }, @@ -393,52 +280,30 @@ func newIncomingTrack( }() } - return t, nil -} + // read incoming RTP packets + go func() { + reorderer := rtpreorderer.New() -// Format returns the track format. -func (t *IncomingTrack) Format() format.Format { - return t.format -} - -// ReadRTP reads a RTP packet. -func (t *IncomingTrack) ReadRTP() (*rtp.Packet, error) { - for { - if len(t.pkts) != 0 { - var pkt *rtp.Packet - pkt, t.pkts = t.pkts[0], t.pkts[1:] - - // sometimes Chrome sends empty RTP packets. ignore them. - if len(pkt.Payload) == 0 { - continue + for { + pkt, _, err := t.track.ReadRTP() + if err != nil { + return } - return pkt, nil + packets, lost := reorderer.Process(pkt) + if lost != 0 { + t.log.Log(logger.Warn, (liberrors.ErrClientRTPPacketsLost{Lost: lost}).Error()) + // do not return + } + + for _, pkt := range packets { + // sometimes Chrome sends empty RTP packets. ignore them. + if len(pkt.Payload) == 0 { + continue + } + + t.OnPacketRTP(pkt) + } } - - pkt, _, err := t.track.ReadRTP() - if err != nil { - return nil, err - } - - var lost int - t.pkts, lost = t.reorderer.Process(pkt) - if lost != 0 { - t.log.Log(logger.Warn, (liberrors.ErrClientRTPPacketsLost{Lost: lost}).Error()) - // do not return - } - - if len(t.pkts) == 0 { - continue - } - - pkt, t.pkts = t.pkts[0], t.pkts[1:] - - // sometimes Chrome sends empty RTP packets. ignore them. - if len(pkt.Payload) == 0 { - continue - } - - return pkt, nil - } + }() } diff --git a/internal/protocols/webrtc/outgoing_track.go b/internal/protocols/webrtc/outgoing_track.go index adc60e51..65f026e9 100644 --- a/internal/protocols/webrtc/outgoing_track.go +++ b/internal/protocols/webrtc/outgoing_track.go @@ -1,9 +1,8 @@ package webrtc import ( - "fmt" + "strings" - "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/pion/rtp" "github.com/pion/webrtc/v3" ) @@ -19,212 +18,16 @@ var multichannelOpusSDP = map[int]string{ // OutgoingTrack is a WebRTC outgoing track type OutgoingTrack struct { - Format format.Format + Caps webrtc.RTPCodecCapability track *webrtc.TrackLocalStaticRTP } -func (t *OutgoingTrack) codecParameters() (webrtc.RTPCodecParameters, error) { - switch forma := t.Format.(type) { - case *format.AV1: - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeAV1, - ClockRate: 90000, - }, - PayloadType: 96, - }, nil - - case *format.VP9: - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeVP9, - ClockRate: 90000, - SDPFmtpLine: "profile-id=0", - }, - PayloadType: 96, - }, nil - - case *format.VP8: - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeVP8, - ClockRate: 90000, - }, - PayloadType: 96, - }, nil - - case *format.H265: - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeH265, - ClockRate: 90000, - }, - PayloadType: 96, - }, nil - - case *format.H264: - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeH264, - ClockRate: 90000, - SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", - }, - PayloadType: 96, - }, nil - - case *format.Opus: - switch forma.ChannelCount { - case 1, 2: - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeOpus, - ClockRate: 48000, - Channels: 2, - SDPFmtpLine: func() string { - s := "minptime=10;useinbandfec=1" - if forma.ChannelCount == 2 { - s += ";stereo=1;sprop-stereo=1" - } - return s - }(), - }, - PayloadType: 96, - }, nil - - case 3, 4, 5, 6, 7, 8: - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: mimeTypeMultiopus, - ClockRate: 48000, - Channels: uint16(forma.ChannelCount), - SDPFmtpLine: multichannelOpusSDP[forma.ChannelCount], - }, - PayloadType: 96, - }, nil - - default: - return webrtc.RTPCodecParameters{}, fmt.Errorf("unsupported channel count: %d", forma.ChannelCount) - } - - case *format.G722: - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeG722, - ClockRate: 8000, - }, - PayloadType: 9, - }, nil - - case *format.G711: - // These are the sample rates and channels supported by Chrome. - // Different sample rates and channels can be streamed too but we don't want compatibility issues. - // https://webrtc.googlesource.com/src/+/refs/heads/main/modules/audio_coding/codecs/pcm16b/audio_decoder_pcm16b.cc#23 - if forma.ClockRate() != 8000 && forma.ClockRate() != 16000 && - forma.ClockRate() != 32000 && forma.ClockRate() != 48000 { - return webrtc.RTPCodecParameters{}, fmt.Errorf("unsupported clock rate: %d", forma.ClockRate()) - } - if forma.ChannelCount != 1 && forma.ChannelCount != 2 { - return webrtc.RTPCodecParameters{}, fmt.Errorf("unsupported channel count: %d", forma.ChannelCount) - } - - if forma.SampleRate == 8000 { - if forma.MULaw { - if forma.ChannelCount != 1 { - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypePCMU, - ClockRate: uint32(forma.SampleRate), - Channels: uint16(forma.ChannelCount), - }, - PayloadType: 96, - }, nil - } - - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypePCMU, - ClockRate: 8000, - }, - PayloadType: 0, - }, nil - } - - if forma.ChannelCount != 1 { - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypePCMA, - ClockRate: uint32(forma.SampleRate), - Channels: uint16(forma.ChannelCount), - }, - PayloadType: 96, - }, nil - } - - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypePCMA, - ClockRate: 8000, - }, - PayloadType: 8, - }, nil - } - - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: mimeTypeL16, - ClockRate: uint32(forma.ClockRate()), - Channels: uint16(forma.ChannelCount), - }, - PayloadType: 96, - }, nil - - case *format.LPCM: - if forma.BitDepth != 16 { - return webrtc.RTPCodecParameters{}, fmt.Errorf("unsupported LPCM bit depth: %d", forma.BitDepth) - } - - // These are the sample rates and channels supported by Chrome. - // Different sample rates and channels can be streamed too but we don't want compatibility issues. - // https://webrtc.googlesource.com/src/+/refs/heads/main/modules/audio_coding/codecs/pcm16b/audio_decoder_pcm16b.cc#23 - if forma.ClockRate() != 8000 && forma.ClockRate() != 16000 && - forma.ClockRate() != 32000 && forma.ClockRate() != 48000 { - return webrtc.RTPCodecParameters{}, fmt.Errorf("unsupported clock rate: %d", forma.ClockRate()) - } - if forma.ChannelCount != 1 && forma.ChannelCount != 2 { - return webrtc.RTPCodecParameters{}, fmt.Errorf("unsupported channel count: %d", forma.ChannelCount) - } - - return webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: mimeTypeL16, - ClockRate: uint32(forma.ClockRate()), - Channels: uint16(forma.ChannelCount), - }, - PayloadType: 96, - }, nil - - default: - return webrtc.RTPCodecParameters{}, fmt.Errorf("unsupported track type: %T", forma) - } -} - func (t *OutgoingTrack) isVideo() bool { - switch t.Format.(type) { - case *format.AV1, - *format.VP9, - *format.VP8, - *format.H265, - *format.H264: - return true - } - - return false + return strings.Split(t.Caps.MimeType, "/")[0] == "video" } func (t *OutgoingTrack) setup(p *PeerConnection) error { - params, _ := t.codecParameters() //nolint:errcheck - var trackID string if t.isVideo() { trackID = "video" @@ -234,7 +37,7 @@ func (t *OutgoingTrack) setup(p *PeerConnection) error { var err error t.track, err = webrtc.NewTrackLocalStaticRTP( - params.RTPCodecCapability, + t.Caps, trackID, webrtcStreamID, ) diff --git a/internal/protocols/webrtc/peer_connection.go b/internal/protocols/webrtc/peer_connection.go index 5f1daee6..231e319a 100644 --- a/internal/protocols/webrtc/peer_connection.go +++ b/internal/protocols/webrtc/peer_connection.go @@ -1,3 +1,4 @@ +// Package webrtc contains WebRTC utilities. package webrtc import ( @@ -89,9 +90,9 @@ type PeerConnection struct { done chan struct{} gatheringDone chan struct{} incomingTrack chan trackRecvPair - - ctx context.Context - ctxCancel context.CancelFunc + ctx context.Context + ctxCancel context.CancelFunc + incomingTracks []*IncomingTrack } // Start starts the peer connection. @@ -132,11 +133,6 @@ func (co *PeerConnection) Start() error { audioSetupped := false for _, tr := range co.OutgoingTracks { - params, err := tr.codecParameters() - if err != nil { - return err - } - var codecType webrtc.RTPCodecType if tr.isVideo() { codecType = webrtc.RTPCodecTypeVideo @@ -146,7 +142,10 @@ func (co *PeerConnection) Start() error { audioSetupped = true } - err = mediaEngine.RegisterCodec(params, codecType) + err := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: tr.Caps, + PayloadType: 96, + }, codecType) if err != nil { return err } @@ -404,28 +403,29 @@ func (co *PeerConnection) GatherIncomingTracks(ctx context.Context) ([]*Incoming maxTrackCount := len(sdp.MediaDescriptions) - var tracks []*IncomingTrack - t := time.NewTimer(time.Duration(co.TrackGatherTimeout)) defer t.Stop() for { select { case <-t.C: - if len(tracks) != 0 { - return tracks, nil + if len(co.incomingTracks) != 0 { + return co.incomingTracks, nil } return nil, fmt.Errorf("deadline exceeded while waiting tracks") case pair := <-co.incomingTrack: - track, err := newIncomingTrack(pair.track, pair.receiver, co.wr.WriteRTCP, co.Log) - if err != nil { - return nil, err + t := &IncomingTrack{ + track: pair.track, + receiver: pair.receiver, + writeRTCP: co.wr.WriteRTCP, + log: co.Log, } - tracks = append(tracks, track) + t.initialize() + co.incomingTracks = append(co.incomingTracks, t) - if len(tracks) >= maxTrackCount { - return tracks, nil + if len(co.incomingTracks) >= maxTrackCount { + return co.incomingTracks, nil } case <-co.Disconnected(): @@ -479,6 +479,13 @@ func (co *PeerConnection) LocalCandidate() string { return "" } +// StartReading starts reading all incoming tracks. +func (co *PeerConnection) StartReading() { + for _, track := range co.incomingTracks { + track.start() + } +} + // RemoteCandidate returns the remote candidate. func (co *PeerConnection) RemoteCandidate() string { var cid string diff --git a/internal/protocols/webrtc/peer_connection_test.go b/internal/protocols/webrtc/peer_connection_test.go index 3f2531e8..72b5a084 100644 --- a/internal/protocols/webrtc/peer_connection_test.go +++ b/internal/protocols/webrtc/peer_connection_test.go @@ -5,10 +5,8 @@ import ( "testing" "time" - "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/test" - "github.com/pion/rtp" "github.com/pion/sdp/v3" "github.com/pion/webrtc/v3" "github.com/stretchr/testify/require" @@ -36,392 +34,6 @@ func TestPeerConnectionCloseImmediately(t *testing.T) { pc.Close() } -func TestPeerConnectionPublishRead(t *testing.T) { - for _, ca := range []struct { - name string - in format.Format - webrtcOut webrtc.RTPCodecCapability - out format.Format - }{ - { - "av1", - &format.AV1{ - PayloadTyp: 96, - }, - webrtc.RTPCodecCapability{ - MimeType: "video/AV1", - ClockRate: 90000, - }, - &format.AV1{ - PayloadTyp: 96, - }, - }, - { - "vp9", - &format.VP9{ - PayloadTyp: 96, - }, - webrtc.RTPCodecCapability{ - MimeType: "video/VP9", - ClockRate: 90000, - SDPFmtpLine: "profile-id=0", - }, - &format.VP9{ - PayloadTyp: 96, - }, - }, - { - "vp8", - &format.VP8{ - PayloadTyp: 96, - }, - webrtc.RTPCodecCapability{ - MimeType: "video/VP8", - ClockRate: 90000, - }, - &format.VP8{ - PayloadTyp: 96, - }, - }, - { - "h265", - test.FormatH265, - webrtc.RTPCodecCapability{ - MimeType: "video/H265", - ClockRate: 90000, - }, - &format.H265{ - PayloadTyp: 96, - }, - }, - { - "h264", - test.FormatH264, - webrtc.RTPCodecCapability{ - MimeType: "video/H264", - ClockRate: 90000, - SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", - }, - &format.H264{ - PayloadTyp: 96, - PacketizationMode: 1, - }, - }, - { - "opus multichannel", - &format.Opus{ - PayloadTyp: 112, - ChannelCount: 6, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/multiopus", - ClockRate: 48000, - Channels: 6, - SDPFmtpLine: "channel_mapping=0,4,1,2,3,5;num_streams=4;coupled_streams=2", - }, - &format.Opus{ - PayloadTyp: 96, - ChannelCount: 6, - }, - }, - { - "opus stereo", - &format.Opus{ - PayloadTyp: 111, - ChannelCount: 2, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/opus", - ClockRate: 48000, - Channels: 2, - SDPFmtpLine: "minptime=10;useinbandfec=1;stereo=1;sprop-stereo=1", - }, - &format.Opus{ - PayloadTyp: 96, - ChannelCount: 2, - }, - }, - { - "opus mono", - &format.Opus{ - PayloadTyp: 111, - ChannelCount: 1, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/opus", - ClockRate: 48000, - Channels: 2, - SDPFmtpLine: "minptime=10;useinbandfec=1", - }, - &format.Opus{ - PayloadTyp: 96, - ChannelCount: 1, - }, - }, - { - "g722", - &format.G722{}, - webrtc.RTPCodecCapability{ - MimeType: "audio/G722", - ClockRate: 8000, - }, - &format.G722{}, - }, - { - "g711 pcma 8khz mono", - &format.G711{ - PayloadTyp: 8, - SampleRate: 8000, - ChannelCount: 1, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/PCMA", - ClockRate: 8000, - }, - &format.G711{ - PayloadTyp: 8, - SampleRate: 8000, - ChannelCount: 1, - }, - }, - { - "g711 pcmu 8khz mono", - &format.G711{ - MULaw: true, - PayloadTyp: 0, - SampleRate: 8000, - ChannelCount: 1, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/PCMU", - ClockRate: 8000, - }, - &format.G711{ - MULaw: true, - PayloadTyp: 0, - SampleRate: 8000, - ChannelCount: 1, - }, - }, - { - "g711 pcma 8khz stereo", - &format.G711{ - PayloadTyp: 96, - SampleRate: 8000, - ChannelCount: 2, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/PCMA", - ClockRate: 8000, - Channels: 2, - }, - &format.G711{ - PayloadTyp: 119, - SampleRate: 8000, - ChannelCount: 2, - }, - }, - { - "g711 pcmu 8khz stereo", - &format.G711{ - MULaw: true, - PayloadTyp: 96, - SampleRate: 8000, - ChannelCount: 2, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/PCMU", - ClockRate: 8000, - Channels: 2, - }, - &format.G711{ - MULaw: true, - PayloadTyp: 118, - SampleRate: 8000, - ChannelCount: 2, - }, - }, - { - "g711 pcma 16khz stereo", - &format.G711{ - PayloadTyp: 96, - SampleRate: 16000, - ChannelCount: 2, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/L16", - ClockRate: 16000, - Channels: 2, - }, - &format.LPCM{ - PayloadTyp: 96, - BitDepth: 16, - SampleRate: 16000, - ChannelCount: 2, - }, - }, - { - "g711 pcmu 16khz stereo", - &format.G711{ - MULaw: true, - PayloadTyp: 96, - SampleRate: 16000, - ChannelCount: 2, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/L16", - ClockRate: 16000, - Channels: 2, - }, - &format.LPCM{ - PayloadTyp: 96, - BitDepth: 16, - SampleRate: 16000, - ChannelCount: 2, - }, - }, - { - "l16 8khz stereo", - &format.LPCM{ - PayloadTyp: 96, - BitDepth: 16, - SampleRate: 8000, - ChannelCount: 2, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/L16", - ClockRate: 8000, - Channels: 2, - }, - &format.LPCM{ - PayloadTyp: 96, - BitDepth: 16, - SampleRate: 8000, - ChannelCount: 2, - }, - }, - { - "l16 16khz stereo", - &format.LPCM{ - PayloadTyp: 96, - BitDepth: 16, - SampleRate: 16000, - ChannelCount: 2, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/L16", - ClockRate: 16000, - Channels: 2, - }, - &format.LPCM{ - PayloadTyp: 96, - BitDepth: 16, - SampleRate: 16000, - ChannelCount: 2, - }, - }, - { - "l16 48khz stereo", - &format.LPCM{ - PayloadTyp: 96, - BitDepth: 16, - SampleRate: 48000, - ChannelCount: 2, - }, - webrtc.RTPCodecCapability{ - MimeType: "audio/L16", - ClockRate: 48000, - Channels: 2, - }, - &format.LPCM{ - PayloadTyp: 96, - BitDepth: 16, - SampleRate: 48000, - ChannelCount: 2, - }, - }, - } { - t.Run(ca.name, func(t *testing.T) { - pc1 := &PeerConnection{ - HandshakeTimeout: conf.StringDuration(10 * time.Second), - TrackGatherTimeout: conf.StringDuration(2 * time.Second), - LocalRandomUDP: true, - IPsFromInterfaces: true, - Publish: true, - OutgoingTracks: []*OutgoingTrack{{ - Format: ca.in, - }}, - Log: test.NilLogger, - } - err := pc1.Start() - require.NoError(t, err) - defer pc1.Close() - - pc2 := &PeerConnection{ - HandshakeTimeout: conf.StringDuration(10 * time.Second), - TrackGatherTimeout: conf.StringDuration(2 * time.Second), - LocalRandomUDP: true, - IPsFromInterfaces: true, - Publish: false, - Log: test.NilLogger, - } - err = pc2.Start() - require.NoError(t, err) - defer pc2.Close() - - offer, err := pc1.CreatePartialOffer() - require.NoError(t, err) - - answer, err := pc2.CreateFullAnswer(context.Background(), offer) - require.NoError(t, err) - - err = pc1.SetAnswer(answer) - require.NoError(t, err) - - go func() { - for { - select { - case cnd := <-pc1.NewLocalCandidate(): - err2 := pc2.AddRemoteCandidate(cnd) - require.NoError(t, err2) - - case <-pc1.Connected(): - return - } - } - }() - - err = pc1.WaitUntilConnected(context.Background()) - require.NoError(t, err) - - err = pc2.WaitUntilConnected(context.Background()) - require.NoError(t, err) - - err = pc1.OutgoingTracks[0].WriteRTP(&rtp.Packet{ - Header: rtp.Header{ - Version: 2, - Marker: true, - PayloadType: 111, - SequenceNumber: 1123, - Timestamp: 45343, - SSRC: 563424, - }, - Payload: []byte{5, 2}, - }) - require.NoError(t, err) - - inc, err := pc2.GatherIncomingTracks(context.Background()) - require.NoError(t, err) - - exp := ca.webrtcOut - exp.RTCPFeedback = inc[0].track.Codec().RTPCodecCapability.RTCPFeedback - require.Equal(t, exp, inc[0].track.Codec().RTPCodecCapability) - - require.Equal(t, ca.out, inc[0].Format()) - }) - } -} - // test that an audio codec is present regardless of the fact that an audio track is. func TestPeerConnectionFallbackCodecs(t *testing.T) { pc1 := &PeerConnection{ @@ -443,8 +55,9 @@ func TestPeerConnectionFallbackCodecs(t *testing.T) { IPsFromInterfaces: true, Publish: true, OutgoingTracks: []*OutgoingTrack{{ - Format: &format.AV1{ - PayloadTyp: 96, + Caps: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeAV1, + ClockRate: 90000, }, }}, Log: test.NilLogger, diff --git a/internal/protocols/webrtc/to_stream.go b/internal/protocols/webrtc/to_stream.go new file mode 100644 index 00000000..d4ba2c7b --- /dev/null +++ b/internal/protocols/webrtc/to_stream.go @@ -0,0 +1,149 @@ +package webrtc + +import ( + "fmt" + "strings" + "time" + + "github.com/bluenviron/gortsplib/v4/pkg/description" + "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/gortsplib/v4/pkg/rtptime" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/pion/rtp" + "github.com/pion/webrtc/v3" +) + +// ToStream maps a WebRTC connection to a MediaMTX stream. +func ToStream(pc *PeerConnection, stream **stream.Stream) ([]*description.Media, error) { + var medias []*description.Media //nolint:prealloc + timeDecoder := rtptime.NewGlobalDecoder() + + for _, track := range pc.incomingTracks { + var typ description.MediaType + var forma format.Format + + switch strings.ToLower(track.track.Codec().MimeType) { + case strings.ToLower(webrtc.MimeTypeAV1): + typ = description.MediaTypeVideo + forma = &format.AV1{ + PayloadTyp: uint8(track.track.PayloadType()), + } + + case strings.ToLower(webrtc.MimeTypeVP9): + typ = description.MediaTypeVideo + forma = &format.VP9{ + PayloadTyp: uint8(track.track.PayloadType()), + } + + case strings.ToLower(webrtc.MimeTypeVP8): + typ = description.MediaTypeVideo + forma = &format.VP8{ + PayloadTyp: uint8(track.track.PayloadType()), + } + + case strings.ToLower(webrtc.MimeTypeH265): + typ = description.MediaTypeVideo + forma = &format.H265{ + PayloadTyp: uint8(track.track.PayloadType()), + } + + case strings.ToLower(webrtc.MimeTypeH264): + typ = description.MediaTypeVideo + forma = &format.H264{ + PayloadTyp: uint8(track.track.PayloadType()), + PacketizationMode: 1, + } + + case strings.ToLower(mimeTypeMultiopus): + typ = description.MediaTypeAudio + forma = &format.Opus{ + PayloadTyp: uint8(track.track.PayloadType()), + ChannelCount: int(track.track.Codec().Channels), + } + + case strings.ToLower(webrtc.MimeTypeOpus): + typ = description.MediaTypeAudio + forma = &format.Opus{ + PayloadTyp: uint8(track.track.PayloadType()), + ChannelCount: func() int { + if strings.Contains(track.track.Codec().SDPFmtpLine, "stereo=1") { + return 2 + } + return 1 + }(), + } + + case strings.ToLower(webrtc.MimeTypeG722): + typ = description.MediaTypeAudio + forma = &format.G722{} + + case strings.ToLower(webrtc.MimeTypePCMU): + channels := int(track.track.Codec().Channels) + if channels == 0 { + channels = 1 + } + + typ = description.MediaTypeAudio + forma = &format.G711{ + PayloadTyp: func() uint8 { + if channels > 1 { + return 118 + } + return 0 + }(), + MULaw: true, + SampleRate: 8000, + ChannelCount: channels, + } + + case strings.ToLower(webrtc.MimeTypePCMA): + channels := int(track.track.Codec().Channels) + if channels == 0 { + channels = 1 + } + + typ = description.MediaTypeAudio + forma = &format.G711{ + PayloadTyp: func() uint8 { + if channels > 1 { + return 119 + } + return 8 + }(), + MULaw: false, + SampleRate: 8000, + ChannelCount: channels, + } + + case strings.ToLower(mimeTypeL16): + typ = description.MediaTypeAudio + forma = &format.LPCM{ + PayloadTyp: uint8(track.track.PayloadType()), + BitDepth: 16, + SampleRate: int(track.track.Codec().ClockRate), + ChannelCount: int(track.track.Codec().Channels), + } + + default: + return nil, fmt.Errorf("unsupported codec: %+v", track.track.Codec().RTPCodecCapability) + } + + medi := &description.Media{ + Type: typ, + Formats: []format.Format{forma}, + } + + track.OnPacketRTP = func(pkt *rtp.Packet) { + pts, ok := timeDecoder.Decode(track, pkt) + if !ok { + return + } + + (*stream).WriteRTPPacket(medi, forma, pkt, time.Now(), pts) + } + + medias = append(medias, medi) + } + + return medias, nil +} diff --git a/internal/protocols/webrtc/to_stream_test.go b/internal/protocols/webrtc/to_stream_test.go new file mode 100644 index 00000000..c01d6030 --- /dev/null +++ b/internal/protocols/webrtc/to_stream_test.go @@ -0,0 +1,406 @@ +package webrtc + +import ( + "context" + "testing" + "time" + + "github.com/bluenviron/gortsplib/v4/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/v3" + "github.com/stretchr/testify/require" +) + +var toFromStreamCases = []struct { + name string + in format.Format + webrtcCaps webrtc.RTPCodecCapability + out format.Format +}{ + { + "av1", + &format.AV1{ + PayloadTyp: 96, + }, + webrtc.RTPCodecCapability{ + MimeType: "video/AV1", + ClockRate: 90000, + }, + &format.AV1{ + PayloadTyp: 96, + }, + }, + { + "vp9", + &format.VP9{ + PayloadTyp: 96, + }, + webrtc.RTPCodecCapability{ + MimeType: "video/VP9", + ClockRate: 90000, + SDPFmtpLine: "profile-id=0", + }, + &format.VP9{ + PayloadTyp: 96, + }, + }, + { + "vp8", + &format.VP8{ + PayloadTyp: 96, + }, + webrtc.RTPCodecCapability{ + MimeType: "video/VP8", + ClockRate: 90000, + }, + &format.VP8{ + PayloadTyp: 96, + }, + }, + { + "h265", + nil, + webrtc.RTPCodecCapability{ + MimeType: "video/H265", + ClockRate: 90000, + }, + &format.H265{ + PayloadTyp: 96, + }, + }, + { + "h264", + test.FormatH264, + webrtc.RTPCodecCapability{ + MimeType: "video/H264", + ClockRate: 90000, + SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", + }, + &format.H264{ + PayloadTyp: 96, + PacketizationMode: 1, + }, + }, + { + "opus multichannel", + &format.Opus{ + PayloadTyp: 112, + ChannelCount: 6, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/multiopus", + ClockRate: 48000, + Channels: 6, + SDPFmtpLine: "channel_mapping=0,4,1,2,3,5;num_streams=4;coupled_streams=2", + }, + &format.Opus{ + PayloadTyp: 96, + ChannelCount: 6, + }, + }, + { + "opus stereo", + &format.Opus{ + PayloadTyp: 111, + ChannelCount: 2, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/opus", + ClockRate: 48000, + Channels: 2, + SDPFmtpLine: "minptime=10;useinbandfec=1;stereo=1;sprop-stereo=1", + }, + &format.Opus{ + PayloadTyp: 96, + ChannelCount: 2, + }, + }, + { + "opus mono", + &format.Opus{ + PayloadTyp: 111, + ChannelCount: 1, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/opus", + ClockRate: 48000, + Channels: 2, + SDPFmtpLine: "minptime=10;useinbandfec=1", + }, + &format.Opus{ + PayloadTyp: 96, + ChannelCount: 1, + }, + }, + { + "g722", + &format.G722{}, + webrtc.RTPCodecCapability{ + MimeType: "audio/G722", + ClockRate: 8000, + }, + &format.G722{}, + }, + { + "g711 pcma 8khz mono", + &format.G711{ + PayloadTyp: 8, + SampleRate: 8000, + ChannelCount: 1, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/PCMA", + ClockRate: 8000, + }, + &format.G711{ + PayloadTyp: 8, + SampleRate: 8000, + ChannelCount: 1, + }, + }, + { + "g711 pcmu 8khz mono", + &format.G711{ + MULaw: true, + PayloadTyp: 0, + SampleRate: 8000, + ChannelCount: 1, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/PCMU", + ClockRate: 8000, + }, + &format.G711{ + MULaw: true, + PayloadTyp: 0, + SampleRate: 8000, + ChannelCount: 1, + }, + }, + { + "g711 pcma 8khz stereo", + &format.G711{ + PayloadTyp: 96, + SampleRate: 8000, + ChannelCount: 2, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/PCMA", + ClockRate: 8000, + Channels: 2, + }, + &format.G711{ + PayloadTyp: 119, + SampleRate: 8000, + ChannelCount: 2, + }, + }, + { + "g711 pcmu 8khz stereo", + &format.G711{ + MULaw: true, + PayloadTyp: 96, + SampleRate: 8000, + ChannelCount: 2, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/PCMU", + ClockRate: 8000, + Channels: 2, + }, + &format.G711{ + MULaw: true, + PayloadTyp: 118, + SampleRate: 8000, + ChannelCount: 2, + }, + }, + { + "g711 pcma 16khz stereo", + &format.G711{ + PayloadTyp: 96, + SampleRate: 16000, + ChannelCount: 2, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/L16", + ClockRate: 16000, + Channels: 2, + }, + &format.LPCM{ + PayloadTyp: 96, + BitDepth: 16, + SampleRate: 16000, + ChannelCount: 2, + }, + }, + { + "g711 pcmu 16khz stereo", + &format.G711{ + MULaw: true, + PayloadTyp: 96, + SampleRate: 16000, + ChannelCount: 2, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/L16", + ClockRate: 16000, + Channels: 2, + }, + &format.LPCM{ + PayloadTyp: 96, + BitDepth: 16, + SampleRate: 16000, + ChannelCount: 2, + }, + }, + { + "l16 8khz stereo", + &format.LPCM{ + PayloadTyp: 96, + BitDepth: 16, + SampleRate: 8000, + ChannelCount: 2, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/L16", + ClockRate: 8000, + Channels: 2, + }, + &format.LPCM{ + PayloadTyp: 96, + BitDepth: 16, + SampleRate: 8000, + ChannelCount: 2, + }, + }, + { + "l16 16khz stereo", + &format.LPCM{ + PayloadTyp: 96, + BitDepth: 16, + SampleRate: 16000, + ChannelCount: 2, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/L16", + ClockRate: 16000, + Channels: 2, + }, + &format.LPCM{ + PayloadTyp: 96, + BitDepth: 16, + SampleRate: 16000, + ChannelCount: 2, + }, + }, + { + "l16 48khz stereo", + &format.LPCM{ + PayloadTyp: 96, + BitDepth: 16, + SampleRate: 48000, + ChannelCount: 2, + }, + webrtc.RTPCodecCapability{ + MimeType: "audio/L16", + ClockRate: 48000, + Channels: 2, + }, + &format.LPCM{ + PayloadTyp: 96, + BitDepth: 16, + SampleRate: 48000, + ChannelCount: 2, + }, + }, +} + +func TestToStream(t *testing.T) { + for _, ca := range toFromStreamCases { + t.Run(ca.name, func(t *testing.T) { + pc1 := &PeerConnection{ + HandshakeTimeout: conf.StringDuration(10 * time.Second), + TrackGatherTimeout: conf.StringDuration(2 * time.Second), + LocalRandomUDP: true, + IPsFromInterfaces: true, + Publish: true, + OutgoingTracks: []*OutgoingTrack{{ + Caps: ca.webrtcCaps, + }}, + Log: test.NilLogger, + } + err := pc1.Start() + require.NoError(t, err) + defer pc1.Close() + + pc2 := &PeerConnection{ + HandshakeTimeout: conf.StringDuration(10 * time.Second), + TrackGatherTimeout: conf.StringDuration(2 * time.Second), + LocalRandomUDP: true, + IPsFromInterfaces: true, + Publish: false, + Log: test.NilLogger, + } + err = pc2.Start() + require.NoError(t, err) + defer pc2.Close() + + offer, err := pc1.CreatePartialOffer() + require.NoError(t, err) + + answer, err := pc2.CreateFullAnswer(context.Background(), offer) + require.NoError(t, err) + + err = pc1.SetAnswer(answer) + require.NoError(t, err) + + go func() { + for { + select { + case cnd := <-pc1.NewLocalCandidate(): + err2 := pc2.AddRemoteCandidate(cnd) + require.NoError(t, err2) + + case <-pc1.Connected(): + return + } + } + }() + + err = pc1.WaitUntilConnected(context.Background()) + require.NoError(t, err) + + err = pc2.WaitUntilConnected(context.Background()) + require.NoError(t, err) + + err = pc1.OutgoingTracks[0].WriteRTP(&rtp.Packet{ + Header: rtp.Header{ + Version: 2, + Marker: true, + PayloadType: 111, + SequenceNumber: 1123, + Timestamp: 45343, + SSRC: 563424, + }, + Payload: []byte{5, 2}, + }) + require.NoError(t, err) + + _, err = pc2.GatherIncomingTracks(context.Background()) + require.NoError(t, err) + + /*exp := ca.webrtcOut + exp.RTCPFeedback = inc[0].track.Codec().RTPCodecCapability.RTCPFeedback + require.Equal(t, exp, inc[0].track.Codec().RTPCodecCapability)*/ + + var stream *stream.Stream + medias, err := ToStream(pc2, &stream) + require.NoError(t, err) + require.Equal(t, ca.out, medias[0].Formats[0]) + }) + } +} diff --git a/internal/protocols/webrtc/track_wrapper.go b/internal/protocols/webrtc/track_wrapper.go deleted file mode 100644 index f336f30d..00000000 --- a/internal/protocols/webrtc/track_wrapper.go +++ /dev/null @@ -1,20 +0,0 @@ -package webrtc - -import ( - "github.com/pion/rtp" -) - -// TrackWrapper provides ClockRate() and PTSEqualsDTS() to WebRTC tracks. -type TrackWrapper struct { - ClockRat int -} - -// ClockRate returns the clock rate. -func (w TrackWrapper) ClockRate() int { - return w.ClockRat -} - -// PTSEqualsDTS returns whether PTS equals DTS. -func (TrackWrapper) PTSEqualsDTS(*rtp.Packet) bool { - return true -} diff --git a/internal/protocols/webrtc/tracks_to_medias.go b/internal/protocols/webrtc/tracks_to_medias.go deleted file mode 100644 index 81ea0850..00000000 --- a/internal/protocols/webrtc/tracks_to_medias.go +++ /dev/null @@ -1,20 +0,0 @@ -package webrtc - -import ( - "github.com/bluenviron/gortsplib/v4/pkg/description" - "github.com/bluenviron/gortsplib/v4/pkg/format" -) - -// TracksToMedias converts WebRTC tracks into a media description. -func TracksToMedias(tracks []*IncomingTrack) []*description.Media { - ret := make([]*description.Media, len(tracks)) - - for i, track := range tracks { - ret[i] = &description.Media{ - Type: track.typ, - Formats: []format.Format{track.format}, - } - } - - return ret -} diff --git a/internal/protocols/webrtc/webrtc.go b/internal/protocols/webrtc/webrtc.go deleted file mode 100644 index 53d485f9..00000000 --- a/internal/protocols/webrtc/webrtc.go +++ /dev/null @@ -1,2 +0,0 @@ -// Package webrtc contains WebRTC utilities. -package webrtc diff --git a/internal/protocols/webrtc/whip_client.go b/internal/protocols/whip/client.go similarity index 78% rename from internal/protocols/webrtc/whip_client.go rename to internal/protocols/whip/client.go index 5c1154b8..1af6cdd4 100644 --- a/internal/protocols/webrtc/whip_client.go +++ b/internal/protocols/whip/client.go @@ -1,4 +1,5 @@ -package webrtc +// Package whip contains a WHIP/WHEP client. +package whip import ( "bytes" @@ -9,51 +10,41 @@ import ( "net/url" "time" - "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/pion/sdp/v3" - "github.com/pion/webrtc/v3" + pwebrtc "github.com/pion/webrtc/v3" "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/httpp" + "github.com/bluenviron/mediamtx/internal/protocols/webrtc" ) const ( - webrtcHandshakeTimeout = 10 * time.Second - webrtcTrackGatherTimeout = 2 * time.Second + handshakeTimeout = 10 * time.Second + trackGatherTimeout = 2 * time.Second ) -// WHIPClient is a WHIP client. -type WHIPClient struct { +// Client is a WHIP client. +type Client struct { HTTPClient *http.Client URL *url.URL Log logger.Writer - pc *PeerConnection + pc *webrtc.PeerConnection patchIsSupported bool } // Publish publishes tracks. -func (c *WHIPClient) Publish( +func (c *Client) Publish( ctx context.Context, - videoTrack format.Format, - audioTrack format.Format, -) ([]*OutgoingTrack, error) { + outgoingTracks []*webrtc.OutgoingTrack, +) error { iceServers, err := c.optionsICEServers(ctx) if err != nil { - return nil, err + return err } - var outgoingTracks []*OutgoingTrack - - if videoTrack != nil { - outgoingTracks = append(outgoingTracks, &OutgoingTrack{Format: videoTrack}) - } - if audioTrack != nil { - outgoingTracks = append(outgoingTracks, &OutgoingTrack{Format: audioTrack}) - } - - c.pc = &PeerConnection{ + c.pc = &webrtc.PeerConnection{ ICEServers: iceServers, HandshakeTimeout: conf.StringDuration(10 * time.Second), TrackGatherTimeout: conf.StringDuration(2 * time.Second), @@ -65,35 +56,35 @@ func (c *WHIPClient) Publish( } err = c.pc.Start() if err != nil { - return nil, err + return err } offer, err := c.pc.CreatePartialOffer() if err != nil { c.pc.Close() - return nil, err + return err } res, err := c.postOffer(ctx, offer) if err != nil { c.pc.Close() - return nil, err + return err } c.URL, err = c.URL.Parse(res.Location) if err != nil { c.pc.Close() - return nil, err + return err } err = c.pc.SetAnswer(res.Answer) if err != nil { c.deleteSession(context.Background()) //nolint:errcheck c.pc.Close() - return nil, err + return err } - t := time.NewTimer(webrtcHandshakeTimeout) + t := time.NewTimer(handshakeTimeout) defer t.Stop() outer: @@ -104,7 +95,7 @@ outer: if err != nil { c.deleteSession(context.Background()) //nolint:errcheck c.pc.Close() - return nil, err + return err } case <-c.pc.GatheringDone(): @@ -115,21 +106,21 @@ outer: case <-t.C: c.deleteSession(context.Background()) //nolint:errcheck c.pc.Close() - return nil, fmt.Errorf("deadline exceeded while waiting connection") + return fmt.Errorf("deadline exceeded while waiting connection") } } - return outgoingTracks, nil + return nil } // Read reads tracks. -func (c *WHIPClient) Read(ctx context.Context) ([]*IncomingTrack, error) { +func (c *Client) Read(ctx context.Context) ([]*webrtc.IncomingTrack, error) { iceServers, err := c.optionsICEServers(ctx) if err != nil { return nil, err } - c.pc = &PeerConnection{ + c.pc = &webrtc.PeerConnection{ ICEServers: iceServers, HandshakeTimeout: conf.StringDuration(10 * time.Second), TrackGatherTimeout: conf.StringDuration(2 * time.Second), @@ -169,7 +160,7 @@ func (c *WHIPClient) Read(ctx context.Context) ([]*IncomingTrack, error) { return nil, err } - err = TracksAreValid(sdp.MediaDescriptions) + err = webrtc.TracksAreValid(sdp.MediaDescriptions) if err != nil { c.deleteSession(context.Background()) //nolint:errcheck c.pc.Close() @@ -183,7 +174,7 @@ func (c *WHIPClient) Read(ctx context.Context) ([]*IncomingTrack, error) { return nil, err } - t := time.NewTimer(webrtcHandshakeTimeout) + t := time.NewTimer(handshakeTimeout) defer t.Stop() outer: @@ -219,15 +210,25 @@ outer: return tracks, nil } +// PeerConnection returns the underlying peer connection. +func (c *Client) PeerConnection() *webrtc.PeerConnection { + return c.pc +} + +// StartReading starts reading all incoming tracks. +func (c *Client) StartReading() { + c.pc.StartReading() +} + // Close closes the client. -func (c *WHIPClient) Close() error { +func (c *Client) Close() error { err := c.deleteSession(context.Background()) c.pc.Close() return err } // Wait waits for client errors. -func (c *WHIPClient) Wait(ctx context.Context) error { +func (c *Client) Wait(ctx context.Context) error { select { case <-c.pc.Disconnected(): return fmt.Errorf("peer connection closed") @@ -237,9 +238,9 @@ func (c *WHIPClient) Wait(ctx context.Context) error { } } -func (c *WHIPClient) optionsICEServers( +func (c *Client) optionsICEServers( ctx context.Context, -) ([]webrtc.ICEServer, error) { +) ([]pwebrtc.ICEServer, error) { req, err := http.NewRequestWithContext(ctx, http.MethodOptions, c.URL.String(), nil) if err != nil { return nil, err @@ -259,14 +260,14 @@ func (c *WHIPClient) optionsICEServers( } type whipPostOfferResponse struct { - Answer *webrtc.SessionDescription + Answer *pwebrtc.SessionDescription Location string ETag string } -func (c *WHIPClient) postOffer( +func (c *Client) postOffer( ctx context.Context, - offer *webrtc.SessionDescription, + offer *pwebrtc.SessionDescription, ) (*whipPostOfferResponse, error) { req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.URL.String(), bytes.NewReader([]byte(offer.SDP))) if err != nil { @@ -304,8 +305,8 @@ func (c *WHIPClient) postOffer( return nil, err } - answer := &webrtc.SessionDescription{ - Type: webrtc.SDPTypeAnswer, + answer := &pwebrtc.SessionDescription{ + Type: pwebrtc.SDPTypeAnswer, SDP: string(sdp), } @@ -316,17 +317,17 @@ func (c *WHIPClient) postOffer( }, nil } -func (c *WHIPClient) patchCandidate( +func (c *Client) patchCandidate( ctx context.Context, - offer *webrtc.SessionDescription, + offer *pwebrtc.SessionDescription, etag string, - candidate *webrtc.ICECandidateInit, + candidate *pwebrtc.ICECandidateInit, ) error { if !c.patchIsSupported { return nil } - frag, err := ICEFragmentMarshal(offer.SDP, []*webrtc.ICECandidateInit{candidate}) + frag, err := ICEFragmentMarshal(offer.SDP, []*pwebrtc.ICECandidateInit{candidate}) if err != nil { return err } @@ -352,7 +353,7 @@ func (c *WHIPClient) patchCandidate( return nil } -func (c *WHIPClient) deleteSession( +func (c *Client) deleteSession( ctx context.Context, ) error { req, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.URL.String(), nil) diff --git a/internal/protocols/webrtc/ice_fragment.go b/internal/protocols/whip/ice_fragment.go similarity index 99% rename from internal/protocols/webrtc/ice_fragment.go rename to internal/protocols/whip/ice_fragment.go index 6866cd5f..516f6572 100644 --- a/internal/protocols/webrtc/ice_fragment.go +++ b/internal/protocols/whip/ice_fragment.go @@ -1,4 +1,4 @@ -package webrtc +package whip import ( "fmt" diff --git a/internal/protocols/webrtc/ice_fragment_test.go b/internal/protocols/whip/ice_fragment_test.go similarity index 99% rename from internal/protocols/webrtc/ice_fragment_test.go rename to internal/protocols/whip/ice_fragment_test.go index 42554723..0b170086 100644 --- a/internal/protocols/webrtc/ice_fragment_test.go +++ b/internal/protocols/whip/ice_fragment_test.go @@ -1,4 +1,4 @@ -package webrtc +package whip import ( "testing" @@ -7,11 +7,11 @@ import ( "github.com/stretchr/testify/require" ) -func stringPtr(v string) *string { +func uint16Ptr(v uint16) *uint16 { return &v } -func uint16Ptr(v uint16) *uint16 { +func stringPtr(v string) *string { return &v } diff --git a/internal/protocols/webrtc/link_header.go b/internal/protocols/whip/link_header.go similarity index 99% rename from internal/protocols/webrtc/link_header.go rename to internal/protocols/whip/link_header.go index a7fbfafb..4fb338b8 100644 --- a/internal/protocols/webrtc/link_header.go +++ b/internal/protocols/whip/link_header.go @@ -1,4 +1,4 @@ -package webrtc +package whip import ( "encoding/json" diff --git a/internal/protocols/webrtc/link_header_test.go b/internal/protocols/whip/link_header_test.go similarity index 98% rename from internal/protocols/webrtc/link_header_test.go rename to internal/protocols/whip/link_header_test.go index 4f59ef9c..80bcdc35 100644 --- a/internal/protocols/webrtc/link_header_test.go +++ b/internal/protocols/whip/link_header_test.go @@ -1,4 +1,4 @@ -package webrtc +package whip import ( "testing" diff --git a/internal/servers/hls/muxer.go b/internal/servers/hls/muxer.go index 6c93d97d..59f125ee 100644 --- a/internal/servers/hls/muxer.go +++ b/internal/servers/hls/muxer.go @@ -12,6 +12,7 @@ import ( "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/protocols/hls" ) const ( @@ -154,7 +155,7 @@ func (m *muxer) runInner() error { } err = mi.initialize() if err != nil { - if m.remoteAddr != "" || errors.Is(err, errNoSupportedCodecs) { + if m.remoteAddr != "" || errors.Is(err, hls.ErrNoSupportedCodecs) { return err } diff --git a/internal/servers/hls/muxer_instance.go b/internal/servers/hls/muxer_instance.go index 94f4d184..5ed2379d 100644 --- a/internal/servers/hls/muxer_instance.go +++ b/internal/servers/hls/muxer_instance.go @@ -1,27 +1,20 @@ package hls import ( - "errors" - "fmt" "os" "path/filepath" "time" "github.com/bluenviron/gohlslib" - "github.com/bluenviron/gohlslib/pkg/codecs" - "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/mediamtx/internal/asyncwriter" "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/protocols/hls" "github.com/bluenviron/mediamtx/internal/stream" - "github.com/bluenviron/mediamtx/internal/unit" "github.com/gin-gonic/gin" ) -var errNoSupportedCodecs = errors.New( - "the stream doesn't contain any supported codec, which are currently H265, H264, Opus, MPEG-4 Audio") - type muxerInstance struct { variant conf.HLSVariant segmentCount int @@ -42,14 +35,6 @@ type muxerInstance struct { func (mi *muxerInstance) initialize() error { mi.writer = asyncwriter.New(mi.writeQueueSize, mi) - videoTrack := mi.createVideoTrack() - audioTrack := mi.createAudioTrack() - - if videoTrack == nil && audioTrack == nil { - mi.stream.RemoveReader(mi.writer) - return errNoSupportedCodecs - } - var muxerDirectory string if mi.directory != "" { muxerDirectory = filepath.Join(mi.directory, mi.pathName) @@ -62,12 +47,16 @@ func (mi *muxerInstance) initialize() error { SegmentDuration: time.Duration(mi.segmentDuration), PartDuration: time.Duration(mi.partDuration), SegmentMaxSize: uint64(mi.segmentMaxSize), - VideoTrack: videoTrack, - AudioTrack: audioTrack, Directory: muxerDirectory, } - err := mi.hmuxer.Start() + err := hls.FromStream(mi.stream, mi.writer, mi.hmuxer, mi) + if err != nil { + mi.stream.RemoveReader(mi.writer) + return err + } + + err = mi.hmuxer.Start() if err != nil { mi.stream.RemoveReader(mi.writer) return err @@ -95,180 +84,6 @@ func (mi *muxerInstance) close() { } } -func (mi *muxerInstance) createVideoTrack() *gohlslib.Track { - var videoFormatAV1 *format.AV1 - videoMedia := mi.stream.Desc().FindFormat(&videoFormatAV1) - - if videoFormatAV1 != nil { - mi.stream.AddReader(mi.writer, videoMedia, videoFormatAV1, func(u unit.Unit) error { - tunit := u.(*unit.AV1) - - if tunit.TU == nil { - return nil - } - - err := mi.hmuxer.WriteAV1(tunit.NTP, tunit.PTS, tunit.TU) - if err != nil { - return fmt.Errorf("muxer error: %w", err) - } - - return nil - }) - - return &gohlslib.Track{ - Codec: &codecs.AV1{}, - } - } - - var videoFormatVP9 *format.VP9 - videoMedia = mi.stream.Desc().FindFormat(&videoFormatVP9) - - if videoFormatVP9 != nil { - mi.stream.AddReader(mi.writer, videoMedia, videoFormatVP9, func(u unit.Unit) error { - tunit := u.(*unit.VP9) - - if tunit.Frame == nil { - return nil - } - - err := mi.hmuxer.WriteVP9(tunit.NTP, tunit.PTS, tunit.Frame) - if err != nil { - return fmt.Errorf("muxer error: %w", err) - } - - return nil - }) - - return &gohlslib.Track{ - Codec: &codecs.VP9{}, - } - } - - var videoFormatH265 *format.H265 - videoMedia = mi.stream.Desc().FindFormat(&videoFormatH265) - - if videoFormatH265 != nil { - mi.stream.AddReader(mi.writer, videoMedia, videoFormatH265, func(u unit.Unit) error { - tunit := u.(*unit.H265) - - if tunit.AU == nil { - return nil - } - - err := mi.hmuxer.WriteH265(tunit.NTP, tunit.PTS, tunit.AU) - if err != nil { - return fmt.Errorf("muxer error: %w", err) - } - - return nil - }) - - vps, sps, pps := videoFormatH265.SafeParams() - - return &gohlslib.Track{ - Codec: &codecs.H265{ - VPS: vps, - SPS: sps, - PPS: pps, - }, - } - } - - var videoFormatH264 *format.H264 - videoMedia = mi.stream.Desc().FindFormat(&videoFormatH264) - - if videoFormatH264 != nil { - mi.stream.AddReader(mi.writer, videoMedia, videoFormatH264, func(u unit.Unit) error { - tunit := u.(*unit.H264) - - if tunit.AU == nil { - return nil - } - - err := mi.hmuxer.WriteH264(tunit.NTP, tunit.PTS, tunit.AU) - if err != nil { - return fmt.Errorf("muxer error: %w", err) - } - - return nil - }) - - sps, pps := videoFormatH264.SafeParams() - - return &gohlslib.Track{ - Codec: &codecs.H264{ - SPS: sps, - PPS: pps, - }, - } - } - - return nil -} - -func (mi *muxerInstance) createAudioTrack() *gohlslib.Track { - var audioFormatOpus *format.Opus - audioMedia := mi.stream.Desc().FindFormat(&audioFormatOpus) - - if audioMedia != nil { - mi.stream.AddReader(mi.writer, audioMedia, audioFormatOpus, func(u unit.Unit) error { - tunit := u.(*unit.Opus) - - err := mi.hmuxer.WriteOpus( - tunit.NTP, - tunit.PTS, - tunit.Packets) - if err != nil { - return fmt.Errorf("muxer error: %w", err) - } - - return nil - }) - - return &gohlslib.Track{ - Codec: &codecs.Opus{ - ChannelCount: audioFormatOpus.ChannelCount, - }, - } - } - - var audioFormatMPEG4Audio *format.MPEG4Audio - audioMedia = mi.stream.Desc().FindFormat(&audioFormatMPEG4Audio) - - if audioMedia != nil { - co := audioFormatMPEG4Audio.GetConfig() - if co == nil { - mi.Log(logger.Warn, "skipping MPEG-4 audio track: tracks without explicit configuration are not supported") - } else { - mi.stream.AddReader(mi.writer, audioMedia, audioFormatMPEG4Audio, func(u unit.Unit) error { - tunit := u.(*unit.MPEG4Audio) - - if tunit.AUs == nil { - return nil - } - - err := mi.hmuxer.WriteMPEG4Audio( - tunit.NTP, - tunit.PTS, - tunit.AUs) - if err != nil { - return fmt.Errorf("muxer error: %w", err) - } - - return nil - }) - - return &gohlslib.Track{ - Codec: &codecs.MPEG4Audio{ - Config: *co, - }, - } - } - } - - return nil -} - func (mi *muxerInstance) errorChan() chan error { return mi.writer.Error() } diff --git a/internal/servers/rtmp/conn.go b/internal/servers/rtmp/conn.go index 0ed3c84a..6bf2553d 100644 --- a/internal/servers/rtmp/conn.go +++ b/internal/servers/rtmp/conn.go @@ -188,7 +188,6 @@ func (c *conn) runRead(conn *rtmp.Conn, u *url.URL) error { c.mutex.Unlock() writer := asyncwriter.New(c.writeQueueSize, c) - defer stream.RemoveReader(writer) err = rtmp.FromStream(stream, writer, conn, c.nconn, time.Duration(c.writeTimeout)) diff --git a/internal/servers/srt/conn.go b/internal/servers/srt/conn.go index 453ad35e..6d41bc32 100644 --- a/internal/servers/srt/conn.go +++ b/internal/servers/srt/conn.go @@ -285,7 +285,6 @@ func (c *conn) runRead(streamID *streamID) error { c.mutex.Unlock() writer := asyncwriter.New(c.writeQueueSize, c) - defer stream.RemoveReader(writer) bw := bufio.NewWriterSize(sconn, srtMaxPayloadSize(c.udpMaxPayloadSize)) diff --git a/internal/servers/webrtc/http_server.go b/internal/servers/webrtc/http_server.go index 4255b907..7872de05 100644 --- a/internal/servers/webrtc/http_server.go +++ b/internal/servers/webrtc/http_server.go @@ -20,7 +20,7 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/httpp" - "github.com/bluenviron/mediamtx/internal/protocols/webrtc" + "github.com/bluenviron/mediamtx/internal/protocols/whip" "github.com/bluenviron/mediamtx/internal/restrictnetwork" ) @@ -184,7 +184,7 @@ func (s *httpServer) onWHIPOptions(ctx *gin.Context, pathName string, publish bo ctx.Writer.Header().Set("Access-Control-Allow-Methods", "OPTIONS, GET, POST, PATCH, DELETE") ctx.Writer.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type, If-Match") ctx.Writer.Header().Set("Access-Control-Expose-Headers", "Link") - ctx.Writer.Header()["Link"] = webrtc.LinkHeaderMarshal(servers) + ctx.Writer.Header()["Link"] = whip.LinkHeaderMarshal(servers) ctx.Writer.WriteHeader(http.StatusNoContent) } @@ -239,7 +239,7 @@ func (s *httpServer) onWHIPPost(ctx *gin.Context, pathName string, publish bool) ctx.Writer.Header().Set("ETag", "*") ctx.Writer.Header().Set("ID", res.sx.uuid.String()) ctx.Writer.Header().Set("Accept-Patch", "application/trickle-ice-sdpfrag") - ctx.Writer.Header()["Link"] = webrtc.LinkHeaderMarshal(servers) + ctx.Writer.Header()["Link"] = whip.LinkHeaderMarshal(servers) ctx.Writer.Header().Set("Location", sessionLocation(publish, pathName, res.sx.secret)) ctx.Writer.WriteHeader(http.StatusCreated) ctx.Writer.Write(res.answer) @@ -263,7 +263,7 @@ func (s *httpServer) onWHIPPatch(ctx *gin.Context, pathName string, rawSecret st return } - candidates, err := webrtc.ICEFragmentUnmarshal(byts) + candidates, err := whip.ICEFragmentUnmarshal(byts) if err != nil { writeError(ctx, http.StatusBadRequest, err) return diff --git a/internal/servers/webrtc/server.go b/internal/servers/webrtc/server.go index a5392032..16342509 100644 --- a/internal/servers/webrtc/server.go +++ b/internal/servers/webrtc/server.go @@ -31,7 +31,6 @@ import ( const ( webrtcTurnSecretExpiration = 24 * 3600 * time.Second - webrtcPayloadMaxSize = 1188 // 1200 - 12 (RTP header) ) // ErrSessionNotFound is returned when a session is not found. diff --git a/internal/servers/webrtc/server_test.go b/internal/servers/webrtc/server_test.go index 6a05433c..73bde323 100644 --- a/internal/servers/webrtc/server_test.go +++ b/internal/servers/webrtc/server_test.go @@ -17,6 +17,7 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/externalcmd" "github.com/bluenviron/mediamtx/internal/protocols/webrtc" + "github.com/bluenviron/mediamtx/internal/protocols/whip" "github.com/bluenviron/mediamtx/internal/stream" "github.com/bluenviron/mediamtx/internal/test" "github.com/bluenviron/mediamtx/internal/unit" @@ -26,6 +27,10 @@ import ( "github.com/stretchr/testify/require" ) +func uint16Ptr(v uint16) *uint16 { + return &v +} + func checkClose(t *testing.T, closeFunc func() error) { require.NoError(t, closeFunc()) } @@ -228,7 +233,7 @@ func TestServerOptionsICEServer(t *testing.T) { require.Equal(t, http.StatusNoContent, res.StatusCode) - iceServers, err := webrtc.LinkHeaderUnmarshal(res.Header["Link"]) + iceServers, err := whip.LinkHeaderUnmarshal(res.Header["Link"]) require.NoError(t, err) require.Equal(t, []pwebrtc.ICEServer{{ @@ -290,17 +295,25 @@ func TestServerPublish(t *testing.T) { su, err := url.Parse("http://myuser:mypass@localhost:8886/teststream/whip?param=value") require.NoError(t, err) - wc := &webrtc.WHIPClient{ + wc := &whip.Client{ HTTPClient: hc, URL: su, Log: test.NilLogger, } - tracks, err := wc.Publish(context.Background(), test.FormatH264, nil) + track := &webrtc.OutgoingTrack{ + Caps: pwebrtc.RTPCodecCapability{ + MimeType: pwebrtc.MimeTypeH264, + ClockRate: 90000, + SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", + }, + } + + err = wc.Publish(context.Background(), []*webrtc.OutgoingTrack{track}) require.NoError(t, err) defer checkClose(t, wc.Close) - err = tracks[0].WriteRTP(&rtp.Packet{ + err = track.WriteRTP(&rtp.Packet{ Header: rtp.Header{ Version: 2, Marker: true, @@ -337,7 +350,7 @@ func TestServerPublish(t *testing.T) { return nil }) - err = tracks[0].WriteRTP(&rtp.Packet{ + err = track.WriteRTP(&rtp.Packet{ Header: rtp.Header{ Version: 2, Marker: true, @@ -565,7 +578,7 @@ func TestServerRead(t *testing.T) { defer tr.CloseIdleConnections() hc := &http.Client{Transport: tr} - wc := &webrtc.WHIPClient{ + wc := &whip.Client{ HTTPClient: hc, URL: u, Log: test.NilLogger, @@ -602,9 +615,20 @@ func TestServerRead(t *testing.T) { require.NoError(t, err) defer checkClose(t, wc.Close) - pkt, err := tracks[0].ReadRTP() - require.NoError(t, err) - require.Equal(t, ca.outRTPPayload, pkt.Payload) + done := make(chan struct{}) + + tracks[0].OnPacketRTP = func(pkt *rtp.Packet) { + select { + case <-done: + default: + require.Equal(t, ca.outRTPPayload, pkt.Payload) + close(done) + } + } + + wc.StartReading() + + <-done }) } } @@ -847,7 +871,7 @@ func TestServerPatchNotFound(t *testing.T) { offer, err := pc.CreateOffer(nil) require.NoError(t, err) - frag, err := webrtc.ICEFragmentMarshal(offer.SDP, []*pwebrtc.ICECandidateInit{{ + frag, err := whip.ICEFragmentMarshal(offer.SDP, []*pwebrtc.ICECandidateInit{{ Candidate: "mycandidate", SDPMLineIndex: uint16Ptr(0), }}) diff --git a/internal/servers/webrtc/session.go b/internal/servers/webrtc/session.go index 421b4a18..eafc4358 100644 --- a/internal/servers/webrtc/session.go +++ b/internal/servers/webrtc/session.go @@ -2,7 +2,6 @@ package webrtc import ( "context" - "crypto/rand" "encoding/hex" "errors" "fmt" @@ -12,14 +11,6 @@ import ( "time" "github.com/bluenviron/gortsplib/v4/pkg/description" - "github.com/bluenviron/gortsplib/v4/pkg/format" - "github.com/bluenviron/gortsplib/v4/pkg/format/rtpav1" - "github.com/bluenviron/gortsplib/v4/pkg/format/rtph264" - "github.com/bluenviron/gortsplib/v4/pkg/format/rtplpcm" - "github.com/bluenviron/gortsplib/v4/pkg/format/rtpvp8" - "github.com/bluenviron/gortsplib/v4/pkg/format/rtpvp9" - "github.com/bluenviron/gortsplib/v4/pkg/rtptime" - "github.com/bluenviron/mediacommon/pkg/codecs/g711" "github.com/google/uuid" "github.com/pion/ice/v2" "github.com/pion/sdp/v3" @@ -33,359 +24,8 @@ import ( "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/webrtc" "github.com/bluenviron/mediamtx/internal/stream" - "github.com/bluenviron/mediamtx/internal/unit" ) -var errNoSupportedCodecs = errors.New( - "the stream doesn't contain any supported codec, which are currently AV1, VP9, VP8, H264, Opus, G722, G711, LPCM") - -type setupStreamFunc func(*webrtc.OutgoingTrack) error - -func uint16Ptr(v uint16) *uint16 { - return &v -} - -func randUint32() (uint32, error) { - var b [4]byte - _, err := rand.Read(b[:]) - if err != nil { - return 0, err - } - return uint32(b[0])<<24 | uint32(b[1])<<16 | uint32(b[2])<<8 | uint32(b[3]), nil -} - -func findVideoTrack( - stream *stream.Stream, - writer *asyncwriter.Writer, -) (format.Format, setupStreamFunc) { - var av1Format *format.AV1 - media := stream.Desc().FindFormat(&av1Format) - - if av1Format != nil { - return av1Format, func(track *webrtc.OutgoingTrack) error { - encoder := &rtpav1.Encoder{ - PayloadType: 105, - PayloadMaxSize: webrtcPayloadMaxSize, - } - err := encoder.Init() - if err != nil { - return err - } - - stream.AddReader(writer, media, av1Format, func(u unit.Unit) error { - tunit := u.(*unit.AV1) - - if tunit.TU == nil { - return nil - } - - packets, err := encoder.Encode(tunit.TU) - if err != nil { - return nil //nolint:nilerr - } - - for _, pkt := range packets { - pkt.Timestamp += tunit.RTPPackets[0].Timestamp - track.WriteRTP(pkt) //nolint:errcheck - } - - return nil - }) - - return nil - } - } - - var vp9Format *format.VP9 - media = stream.Desc().FindFormat(&vp9Format) - - if vp9Format != nil { - return vp9Format, func(track *webrtc.OutgoingTrack) error { - encoder := &rtpvp9.Encoder{ - PayloadType: 96, - PayloadMaxSize: webrtcPayloadMaxSize, - InitialPictureID: uint16Ptr(8445), - } - err := encoder.Init() - if err != nil { - return err - } - - stream.AddReader(writer, media, vp9Format, func(u unit.Unit) error { - tunit := u.(*unit.VP9) - - if tunit.Frame == nil { - return nil - } - - packets, err := encoder.Encode(tunit.Frame) - if err != nil { - return nil //nolint:nilerr - } - - for _, pkt := range packets { - pkt.Timestamp += tunit.RTPPackets[0].Timestamp - track.WriteRTP(pkt) //nolint:errcheck - } - - return nil - }) - - return nil - } - } - - var vp8Format *format.VP8 - media = stream.Desc().FindFormat(&vp8Format) - - if vp8Format != nil { - return vp8Format, func(track *webrtc.OutgoingTrack) error { - encoder := &rtpvp8.Encoder{ - PayloadType: 96, - PayloadMaxSize: webrtcPayloadMaxSize, - } - err := encoder.Init() - if err != nil { - return err - } - - stream.AddReader(writer, media, vp8Format, func(u unit.Unit) error { - tunit := u.(*unit.VP8) - - if tunit.Frame == nil { - return nil - } - - packets, err := encoder.Encode(tunit.Frame) - if err != nil { - return nil //nolint:nilerr - } - - for _, pkt := range packets { - pkt.Timestamp += tunit.RTPPackets[0].Timestamp - track.WriteRTP(pkt) //nolint:errcheck - } - - return nil - }) - - return nil - } - } - - var h264Format *format.H264 - media = stream.Desc().FindFormat(&h264Format) - - if h264Format != nil { - return h264Format, func(track *webrtc.OutgoingTrack) error { - encoder := &rtph264.Encoder{ - PayloadType: 96, - PayloadMaxSize: webrtcPayloadMaxSize, - } - err := encoder.Init() - if err != nil { - return err - } - - firstReceived := false - var lastPTS time.Duration - - stream.AddReader(writer, media, h264Format, func(u unit.Unit) error { - tunit := u.(*unit.H264) - - if tunit.AU == nil { - return nil - } - - if !firstReceived { - firstReceived = true - } else if tunit.PTS < lastPTS { - return fmt.Errorf("WebRTC doesn't support H264 streams with B-frames") - } - lastPTS = tunit.PTS - - packets, err := encoder.Encode(tunit.AU) - if err != nil { - return nil //nolint:nilerr - } - - for _, pkt := range packets { - pkt.Timestamp += tunit.RTPPackets[0].Timestamp - track.WriteRTP(pkt) //nolint:errcheck - } - - return nil - }) - - return nil - } - } - - return nil, nil -} - -func findAudioTrack( - stream *stream.Stream, - writer *asyncwriter.Writer, -) (format.Format, setupStreamFunc) { - var opusFormat *format.Opus - media := stream.Desc().FindFormat(&opusFormat) - - if opusFormat != nil { - return opusFormat, func(track *webrtc.OutgoingTrack) error { - stream.AddReader(writer, media, opusFormat, func(u unit.Unit) error { - for _, pkt := range u.GetRTPPackets() { - track.WriteRTP(pkt) //nolint:errcheck - } - - return nil - }) - return nil - } - } - - var g722Format *format.G722 - media = stream.Desc().FindFormat(&g722Format) - - if g722Format != nil { - return g722Format, func(track *webrtc.OutgoingTrack) error { - stream.AddReader(writer, media, g722Format, func(u unit.Unit) error { - for _, pkt := range u.GetRTPPackets() { - track.WriteRTP(pkt) //nolint:errcheck - } - - return nil - }) - return nil - } - } - - var g711Format *format.G711 - media = stream.Desc().FindFormat(&g711Format) - - if g711Format != nil { - return g711Format, func(track *webrtc.OutgoingTrack) error { - if g711Format.SampleRate == 8000 { - curTimestamp, err := randUint32() - if err != nil { - return err - } - - stream.AddReader(writer, media, g711Format, func(u unit.Unit) error { - for _, pkt := range u.GetRTPPackets() { - // 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) - - track.WriteRTP(pkt) //nolint:errcheck - } - - return nil - }) - } else { - encoder := &rtplpcm.Encoder{ - PayloadType: 96, - PayloadMaxSize: webrtcPayloadMaxSize, - BitDepth: 16, - ChannelCount: g711Format.ChannelCount, - } - err := encoder.Init() - if err != nil { - return err - } - - curTimestamp, err := randUint32() - if err != nil { - return err - } - - stream.AddReader(writer, media, g711Format, func(u unit.Unit) error { - tunit := u.(*unit.G711) - - if tunit.Samples == nil { - return nil - } - - var lpcmSamples []byte - if g711Format.MULaw { - lpcmSamples = g711.DecodeMulaw(tunit.Samples) - } else { - lpcmSamples = g711.DecodeAlaw(tunit.Samples) - } - - packets, err := encoder.Encode(lpcmSamples) - if err != nil { - return nil //nolint:nilerr - } - - 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) - - track.WriteRTP(pkt) //nolint:errcheck - } - - return nil - }) - } - return nil - } - } - - var lpcmFormat *format.LPCM - media = stream.Desc().FindFormat(&lpcmFormat) - - if lpcmFormat != nil { - return lpcmFormat, func(track *webrtc.OutgoingTrack) error { - encoder := &rtplpcm.Encoder{ - PayloadType: 96, - BitDepth: 16, - ChannelCount: lpcmFormat.ChannelCount, - PayloadMaxSize: webrtcPayloadMaxSize, - } - err := encoder.Init() - if err != nil { - return err - } - - curTimestamp, err := randUint32() - if err != nil { - return err - } - - stream.AddReader(writer, media, lpcmFormat, func(u unit.Unit) error { - tunit := u.(*unit.LPCM) - - if tunit.Samples == nil { - return nil - } - - packets, err := encoder.Encode(tunit.Samples) - if err != nil { - return nil //nolint:nilerr - } - - 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) - - track.WriteRTP(pkt) //nolint:errcheck - } - - return nil - }) - return nil - } - } - - return nil, nil -} - func whipOffer(body []byte) *pwebrtc.SessionDescription { return &pwebrtc.SessionDescription{ Type: pwebrtc.SDPTypeOffer, @@ -573,14 +213,19 @@ func (s *session) runPublish() (int, error) { s.pc = pc s.mutex.Unlock() - tracks, err := pc.GatherIncomingTracks(s.ctx) + _, err = pc.GatherIncomingTracks(s.ctx) if err != nil { return 0, err } - medias := webrtc.TracksToMedias(tracks) + var stream *stream.Stream - stream, err := path.StartPublisher(defs.PathStartPublisherReq{ + medias, err := webrtc.ToStream(pc, &stream) + if err != nil { + return 0, err + } + + stream, err = path.StartPublisher(defs.PathStartPublisherReq{ Author: s, Desc: &description.Session{Medias: medias}, GenerateRTPPackets: false, @@ -589,29 +234,7 @@ func (s *session) runPublish() (int, error) { return 0, err } - timeDecoder := rtptime.NewGlobalDecoder() - - for i, media := range medias { - ci := i - cmedia := media - trackWrapper := &webrtc.TrackWrapper{ClockRat: cmedia.Formats[0].ClockRate()} - - go func() { - for { - pkt, err := tracks[ci].ReadRTP() - if err != nil { - return - } - - pts, ok := timeDecoder.Decode(trackWrapper, pkt) - if !ok { - continue - } - - stream.WriteRTPPacket(cmedia, cmedia.Formats[0], pkt, time.Now(), pts) - } - }() - } + pc.StartReading() select { case <-pc.Disconnected(): @@ -661,22 +284,7 @@ func (s *session) runRead() (int, error) { } writer := asyncwriter.New(s.writeQueueSize, s) - - videoTrack, videoSetup := findVideoTrack(stream, writer) - audioTrack, audioSetup := findAudioTrack(stream, writer) - - if videoTrack == nil && audioTrack == nil { - return http.StatusBadRequest, errNoSupportedCodecs - } - - var outgoingTracks []*webrtc.OutgoingTrack - - if videoTrack != nil { - outgoingTracks = append(outgoingTracks, &webrtc.OutgoingTrack{Format: videoTrack}) - } - if audioTrack != nil { - outgoingTracks = append(outgoingTracks, &webrtc.OutgoingTrack{Format: audioTrack}) - } + defer stream.RemoveReader(writer) pc := &webrtc.PeerConnection{ ICEServers: iceServers, @@ -688,9 +296,14 @@ func (s *session) runRead() (int, error) { ICEUDPMux: s.iceUDPMux, ICETCPMux: s.iceTCPMux, Publish: true, - OutgoingTracks: outgoingTracks, Log: s, } + + err = webrtc.FromStream(stream, writer, pc) + if err != nil { + return http.StatusBadRequest, err + } + err = pc.Start() if err != nil { return http.StatusBadRequest, err @@ -717,25 +330,6 @@ func (s *session) runRead() (int, error) { s.pc = pc s.mutex.Unlock() - defer stream.RemoveReader(writer) - - n := 0 - - if videoTrack != nil { - err := videoSetup(outgoingTracks[n]) - if err != nil { - return 0, err - } - n++ - } - - if audioTrack != nil { - err := audioSetup(outgoingTracks[n]) - if err != nil { - return 0, err - } - } - s.Log(logger.Info, "is reading from path '%s', %s", path.Name(), defs.FormatsInfo(stream.FormatsForReader(writer))) diff --git a/internal/staticsources/hls/source.go b/internal/staticsources/hls/source.go index 7e333f72..095ed7a7 100644 --- a/internal/staticsources/hls/source.go +++ b/internal/staticsources/hls/source.go @@ -6,16 +6,14 @@ import ( "time" "github.com/bluenviron/gohlslib" - "github.com/bluenviron/gohlslib/pkg/codecs" "github.com/bluenviron/gortsplib/v4/pkg/description" - "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/protocols/hls" "github.com/bluenviron/mediamtx/internal/protocols/tls" "github.com/bluenviron/mediamtx/internal/stream" - "github.com/bluenviron/mediamtx/internal/unit" ) // Source is a HLS static source. @@ -69,133 +67,9 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { decodeErrLogger.Log(logger.Warn, err.Error()) }, OnTracks: func(tracks []*gohlslib.Track) error { - var medias []*description.Media - - for _, track := range tracks { - var medi *description.Media - - switch tcodec := track.Codec.(type) { - case *codecs.AV1: - medi = &description.Media{ - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.AV1{ - PayloadTyp: 96, - }}, - } - - c.OnDataAV1(track, func(pts time.Duration, tu [][]byte) { - stream.WriteUnit(medi, medi.Formats[0], &unit.AV1{ - Base: unit.Base{ - NTP: time.Now(), - PTS: pts, - }, - TU: tu, - }) - }) - - case *codecs.VP9: - medi = &description.Media{ - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.VP9{ - PayloadTyp: 96, - }}, - } - - c.OnDataVP9(track, func(pts time.Duration, frame []byte) { - stream.WriteUnit(medi, medi.Formats[0], &unit.VP9{ - Base: unit.Base{ - NTP: time.Now(), - PTS: pts, - }, - Frame: frame, - }) - }) - - case *codecs.H264: - medi = &description.Media{ - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.H264{ - PayloadTyp: 96, - PacketizationMode: 1, - SPS: tcodec.SPS, - PPS: tcodec.PPS, - }}, - } - - c.OnDataH26x(track, func(pts time.Duration, _ time.Duration, au [][]byte) { - stream.WriteUnit(medi, medi.Formats[0], &unit.H264{ - Base: unit.Base{ - NTP: time.Now(), - PTS: pts, - }, - AU: au, - }) - }) - - case *codecs.H265: - medi = &description.Media{ - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.H265{ - PayloadTyp: 96, - VPS: tcodec.VPS, - SPS: tcodec.SPS, - PPS: tcodec.PPS, - }}, - } - - c.OnDataH26x(track, func(pts time.Duration, _ time.Duration, au [][]byte) { - stream.WriteUnit(medi, medi.Formats[0], &unit.H265{ - Base: unit.Base{ - NTP: time.Now(), - PTS: pts, - }, - AU: au, - }) - }) - - case *codecs.MPEG4Audio: - medi = &description.Media{ - Type: description.MediaTypeAudio, - Formats: []format.Format{&format.MPEG4Audio{ - PayloadTyp: 96, - SizeLength: 13, - IndexLength: 3, - IndexDeltaLength: 3, - Config: &tcodec.Config, - }}, - } - - c.OnDataMPEG4Audio(track, func(pts time.Duration, aus [][]byte) { - stream.WriteUnit(medi, medi.Formats[0], &unit.MPEG4Audio{ - Base: unit.Base{ - NTP: time.Now(), - PTS: pts, - }, - AUs: aus, - }) - }) - - case *codecs.Opus: - medi = &description.Media{ - Type: description.MediaTypeAudio, - Formats: []format.Format{&format.Opus{ - PayloadTyp: 96, - ChannelCount: tcodec.ChannelCount, - }}, - } - - c.OnDataOpus(track, func(pts time.Duration, packets [][]byte) { - stream.WriteUnit(medi, medi.Formats[0], &unit.Opus{ - Base: unit.Base{ - NTP: time.Now(), - PTS: pts, - }, - Packets: packets, - }) - }) - } - - medias = append(medias, medi) + medias, err := hls.ToStream(c, tracks, &stream) + if err != nil { + return err } res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{ diff --git a/internal/staticsources/webrtc/source.go b/internal/staticsources/webrtc/source.go index f2004a7d..18256a44 100644 --- a/internal/staticsources/webrtc/source.go +++ b/internal/staticsources/webrtc/source.go @@ -8,13 +8,14 @@ import ( "time" "github.com/bluenviron/gortsplib/v4/pkg/description" - "github.com/bluenviron/gortsplib/v4/pkg/rtptime" "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/tls" "github.com/bluenviron/mediamtx/internal/protocols/webrtc" + "github.com/bluenviron/mediamtx/internal/protocols/whip" + "github.com/bluenviron/mediamtx/internal/stream" ) // Source is a WebRTC static source. @@ -44,7 +45,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { } defer tr.CloseIdleConnections() - client := webrtc.WHIPClient{ + client := whip.Client{ HTTPClient: &http.Client{ Timeout: time.Duration(s.ReadTimeout), Transport: tr, @@ -53,13 +54,18 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { Log: s, } - tracks, err := client.Read(params.Context) + _, err = client.Read(params.Context) if err != nil { return err } defer client.Close() //nolint:errcheck - medias := webrtc.TracksToMedias(tracks) + var stream *stream.Stream + + medias, err := webrtc.ToStream(client.PeerConnection(), &stream) + if err != nil { + return err + } rres := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{ Desc: &description.Session{Medias: medias}, @@ -69,31 +75,11 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { return rres.Err } + stream = rres.Stream + defer s.Parent.SetNotReady(defs.PathSourceStaticSetNotReadyReq{}) - timeDecoder := rtptime.NewGlobalDecoder() - - for i, media := range medias { - ci := i - cmedia := media - trackWrapper := &webrtc.TrackWrapper{ClockRat: cmedia.Formats[0].ClockRate()} - - go func() { - for { - pkt, err := tracks[ci].ReadRTP() - if err != nil { - return - } - - pts, ok := timeDecoder.Decode(trackWrapper, pkt) - if !ok { - continue - } - - rres.Stream.WriteRTPPacket(cmedia, cmedia.Formats[0], pkt, time.Now(), pts) - } - }() - } + client.StartReading() return client.Wait(params.Context) } diff --git a/internal/staticsources/webrtc/source_test.go b/internal/staticsources/webrtc/source_test.go index c8b2842f..96d2c140 100644 --- a/internal/staticsources/webrtc/source_test.go +++ b/internal/staticsources/webrtc/source_test.go @@ -8,7 +8,6 @@ import ( "testing" "time" - "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/pion/rtp" pwebrtc "github.com/pion/webrtc/v3" "github.com/stretchr/testify/require" @@ -27,10 +26,15 @@ func whipOffer(body []byte) *pwebrtc.SessionDescription { } func TestSource(t *testing.T) { - outgoingTracks := []*webrtc.OutgoingTrack{{Format: &format.Opus{ - PayloadTyp: 111, - ChannelCount: 2, - }}} + outgoingTracks := []*webrtc.OutgoingTrack{{ + Caps: pwebrtc.RTPCodecCapability{ + MimeType: "audio/opus", + ClockRate: 48000, + Channels: 2, + SDPFmtpLine: "minptime=10;useinbandfec=1;stereo=1;sprop-stereo=1", + }, + }} + pc := &webrtc.PeerConnection{ LocalRandomUDP: true, IPsFromInterfaces: true,