diff --git a/api/openapi.yaml b/api/openapi.yaml index 0e054326..ca288533 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -151,6 +151,8 @@ components: - rtsp - rtsps - srt + - whip + - whips ForwardDestState: type: string @@ -1267,6 +1269,8 @@ components: properties: dest: type: string + whipBearerToken: + type: string PathList: type: object diff --git a/docs/2-features/11-forward.md b/docs/2-features/11-forward.md index 87290cb3..ea0d22b5 100644 --- a/docs/2-features/11-forward.md +++ b/docs/2-features/11-forward.md @@ -2,12 +2,38 @@ Incoming streams can be natively forwarded to other servers with the following protocols: +- [SRT](#srt) +- [WebRTC](#webrtc) - [RTSP](#rtsp) - [RTMP](#rtmp) -- [SRT](#srt) It is also possible to use [FFmpeg](#ffmpeg) to perform the forwarding. +## SRT + +Add the target URL inside `dest` of a `forward` entry: + +```yml +paths: + mypath: + forward: + - dest: srt://host:port?streamid=streamid +``` + +## WebRTC + +We support forwarding streams by using the WebRTC protocol and the WHIP extension. Add the target URL inside `dest` of a `forward` entry. Use `whip://` for HTTP and `whips://` for HTTPS: + +```yml +paths: + mypath: + forward: + - dest: whip://host:port/mystream/whip + whipBearerToken: mytoken +``` + +If the remote server is a _MediaMTX_ instance, remember to add a `/whip` suffix after the stream name, since in _MediaMTX_ [it's part of the WHIP URL](../3-publish/05-webrtc-clients.md). + ## RTSP Add the target URL inside `dest` of a `forward` entry: @@ -30,17 +56,6 @@ paths: - dest: rtmp://user:pass@host:port/path#streamKey ``` -## SRT - -Add the target URL inside `dest` of a `forward` entry: - -```yml -paths: - mypath: - forward: - - dest: srt://host:port?streamid=streamid -``` - ## FFmpeg When the destination requires transcoding, filtering or a protocol that is not supported by `forward`, use _FFmpeg_ inside the `runOnAvailable` parameter instead: diff --git a/internal/api/api_forward_test.go b/internal/api/api_forward_test.go index 2265be72..d36e544a 100644 --- a/internal/api/api_forward_test.go +++ b/internal/api/api_forward_test.go @@ -53,11 +53,12 @@ func (m *testForwardPathManager) APIForwardDestGet(path string, id uuid.UUID) (* } func TestForward(t *testing.T) { - id := uuid.New() + rtmpID := uuid.New() + whipID := uuid.New() pathManager := &testForwardPathManager{ items: map[uuid.UUID]*defs.APIForwardDest{ - id: { - ID: id, + rtmpID: { + ID: rtmpID, Pos: 1, Created: time.Date(2026, 6, 18, 9, 0, 0, 0, time.UTC), Conf: conf.ForwardDest{Dest: "rtmp://localhost/live/stream"}, @@ -66,6 +67,15 @@ func TestForward(t *testing.T) { LastError: "connection refused", OutboundBytes: 123, }, + whipID: { + ID: whipID, + Pos: 2, + Created: time.Date(2026, 6, 18, 9, 1, 0, 0, time.UTC), + Conf: conf.ForwardDest{Dest: "whip://localhost/live/stream/whip", WhipBearerToken: "mytoken"}, + Protocol: defs.APIForwardDestProtocolWHIP, + State: defs.APIForwardDestStateForwarding, + OutboundBytes: 456, + }, }, } @@ -88,17 +98,38 @@ func TestForward(t *testing.T) { var list defs.APIForwardDestList httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/paths/forward/list?path=my%2Fnested%2Fstream", nil, &list) - require.Equal(t, 1, list.ItemCount) + require.Equal(t, 2, list.ItemCount) require.Equal(t, 1, list.PageCount) - require.Equal(t, id, list.Items[0].ID) - require.Equal(t, 1, list.Items[0].Pos) + require.Len(t, list.Items, 2) + + require.ElementsMatch(t, []defs.APIForwardDest{ + { + ID: rtmpID, + Pos: 1, + Created: time.Date(2026, 6, 18, 9, 0, 0, 0, time.UTC), + Conf: conf.ForwardDest{Dest: "rtmp://localhost/live/stream"}, + Protocol: defs.APIForwardDestProtocolRTMP, + State: defs.APIForwardDestStateError, + LastError: "connection refused", + OutboundBytes: 123, + }, + { + ID: whipID, + Pos: 2, + Created: time.Date(2026, 6, 18, 9, 1, 0, 0, time.UTC), + Conf: conf.ForwardDest{Dest: "whip://localhost/live/stream/whip", WhipBearerToken: "mytoken"}, + Protocol: defs.APIForwardDestProtocolWHIP, + State: defs.APIForwardDestStateForwarding, + OutboundBytes: 456, + }, + }, list.Items) var item defs.APIForwardDest httpRequest(t, hc, http.MethodGet, - "http://localhost:9997/v3/paths/forward/get?path=my%2Fnested%2Fstream&id="+id.String(), nil, &item) - require.Equal(t, "rtmp://localhost/live/stream", item.Conf.Dest) - require.Equal(t, defs.APIForwardDestProtocolRTMP, item.Protocol) - require.Equal(t, defs.APIForwardDestStateError, item.State) - require.Equal(t, "connection refused", item.LastError) - require.Equal(t, uint64(123), item.OutboundBytes) + "http://localhost:9997/v3/paths/forward/get?path=my%2Fnested%2Fstream&id="+whipID.String(), nil, &item) + require.Equal(t, "whip://localhost/live/stream/whip", item.Conf.Dest) + require.Equal(t, "mytoken", item.Conf.WhipBearerToken) + require.Equal(t, defs.APIForwardDestProtocolWHIP, item.Protocol) + require.Equal(t, defs.APIForwardDestStateForwarding, item.State) + require.Equal(t, uint64(456), item.OutboundBytes) } diff --git a/internal/conf/conf_test.go b/internal/conf/conf_test.go index 62bb66bf..bf3937aa 100644 --- a/internal/conf/conf_test.go +++ b/internal/conf/conf_test.go @@ -884,20 +884,34 @@ func TestConfErrors(t *testing.T) { " source: rtsp://user@localhost/stream\n", "username and password must be both provided", }, + { + "valid whip forward destination", + "paths:\n" + + " mypath:\n" + + " forward:\n" + + " - dest: whip://localhost/stream/whip\n" + + " whipBearerToken: mytoken\n", + "", + }, { "invalid forward destination", "paths:\n" + " mypath:\n" + " forward:\n" + " - dest: http://localhost/stream\n", - "invalid 'forward': entry 0: unsupported scheme 'http', supported ones are rtmp, rtmps, rtsp, rtsps and srt", + "invalid 'forward': entry 0: unsupported scheme 'http', supported ones are " + + "rtmp, rtmps, rtsp, rtsps, srt, whip and whips", }, } { t.Run(ca.name, func(t *testing.T) { tmpf := createTempFile(t, []byte(ca.conf)) _, _, err := Load(tmpf, nil, nil) - require.EqualError(t, err, ca.err) + if ca.err == "" { + require.NoError(t, err) + } else { + require.EqualError(t, err, ca.err) + } }) } } diff --git a/internal/conf/forward_dest.go b/internal/conf/forward_dest.go index 1fd8d1ff..711f8ac2 100644 --- a/internal/conf/forward_dest.go +++ b/internal/conf/forward_dest.go @@ -8,7 +8,8 @@ import ( // ForwardDest is a destination to which a path is forwarded. type ForwardDest struct { - Dest string `json:"dest"` + Dest string `json:"dest"` + WhipBearerToken string `json:"whipBearerToken"` } func validateForwardDest(dest string) (*url.URL, error) { @@ -29,10 +30,10 @@ func (p *ForwardDest) Validate() error { } switch u.Scheme { - case "rtmp", "rtmps", "rtsp", "rtsps", "srt": + case "rtmp", "rtmps", "rtsp", "rtsps", "srt", "whip", "whips": default: return fmt.Errorf( - "unsupported scheme '%s', supported ones are rtmp, rtmps, rtsp, rtsps and srt", + "unsupported scheme '%s', supported ones are rtmp, rtmps, rtsp, rtsps, srt, whip and whips", u.Scheme) } diff --git a/internal/core/forward_test.go b/internal/core/forward_test.go index 6406b213..bae28784 100644 --- a/internal/core/forward_test.go +++ b/internal/core/forward_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "io" "net" "net/http" "net/url" @@ -14,9 +15,12 @@ import ( "github.com/bluenviron/gortmplib" rtmpcodecs "github.com/bluenviron/gortmplib/pkg/codecs" "github.com/google/uuid" + "github.com/pion/rtp" + pwebrtc "github.com/pion/webrtc/v4" "github.com/stretchr/testify/require" "github.com/bluenviron/mediamtx/internal/defs" + mtxwebrtc "github.com/bluenviron/mediamtx/internal/protocols/webrtc" "github.com/bluenviron/mediamtx/internal/test" ) @@ -204,10 +208,146 @@ func waitRTMPForwardFrame( } } +func startWHIPForwardServer( + t *testing.T, + expectedBearerToken string, +) (string, <-chan struct{}, <-chan error) { + pc := &mtxwebrtc.PeerConnection{ + LocalRandomUDP: true, + IPsFromInterfaces: true, + Log: test.NilLogger, + } + err := pc.Start() + require.NoError(t, err) + t.Cleanup(func() { + pc.Close() + }) + + received := make(chan struct{}, 16) + serverErr := make(chan error, 16) + + httpServ := &http.Server{ + Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if expectedBearerToken != "" { + require.Equal(t, "Bearer "+expectedBearerToken, r.Header.Get("Authorization")) + } + + switch { + case r.Method == http.MethodOptions && r.URL.Path == "/teststream/whip": + w.Header().Set("Access-Control-Allow-Methods", "OPTIONS, GET, POST, PATCH, DELETE") + w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type, If-Match") + w.WriteHeader(http.StatusNoContent) + + case r.Method == http.MethodPost && r.URL.Path == "/teststream/whip": + require.Equal(t, "application/sdp", r.Header.Get("Content-Type")) + + body, err2 := io.ReadAll(r.Body) + require.NoError(t, err2) + offer := &pwebrtc.SessionDescription{ + Type: pwebrtc.SDPTypeOffer, + SDP: string(body), + } + + answer, err2 := pc.CreateFullAnswer(offer, false) + require.NoError(t, err2) + + w.Header().Set("Content-Type", "application/sdp") + w.Header().Set("ETag", "test_etag") + w.Header().Set("Location", "/teststream/whip/sessionid") + w.WriteHeader(http.StatusCreated) + _, err2 = w.Write([]byte(answer.SDP)) + require.NoError(t, err2) + + go func() { + err3 := pc.WaitUntilConnected(10 * time.Second) + if err3 != nil { + serverErr <- err3 + return + } + + err3 = pc.GatherInboundTracks(2 * time.Second) + if err3 != nil { + serverErr <- err3 + return + } + + if len(pc.InboundTracks()) != 1 { + serverErr <- fmt.Errorf("unexpected track count: %d", len(pc.InboundTracks())) + return + } + + pc.InboundTracks()[0].OnPacketRTP = func(_ *rtp.Packet) { + select { + case received <- struct{}{}: + default: + } + } + + pc.StartReading() + }() + + case r.URL.Path == "/teststream/whip/sessionid" && r.Method == http.MethodPatch: + w.WriteHeader(http.StatusNoContent) + + case r.URL.Path == "/teststream/whip/sessionid" && r.Method == http.MethodDelete: + w.WriteHeader(http.StatusOK) + + default: + serverErr <- fmt.Errorf("unexpected request: %s %s", r.Method, r.URL.Path) + w.WriteHeader(http.StatusBadRequest) + } + }), + } + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + go httpServ.Serve(ln) + + t.Cleanup(func() { + httpServ.Shutdown(context.Background()) + }) + + return "whip://" + ln.Addr().String() + "/teststream/whip", received, serverErr +} + +func waitWHIPForwardFrame( + t *testing.T, + w *gortmplib.Writer, + track *gortmplib.Track, + received <-chan struct{}, + serverErr <-chan error, +) { + t.Helper() + + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + timer := time.NewTimer(10 * time.Second) + defer timer.Stop() + + for { + select { + case <-received: + return + + case err := <-serverErr: + require.NoError(t, err) + + case <-ticker.C: + err := w.WriteH264(track, 2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) + require.NoError(t, err) + + case <-timer.C: + t.Fatal("timed out waiting for WHIP forwarded frame") + } + } +} + func TestPathForwardRTMP(t *testing.T) { dest, received, serverErr := startRTMPForwardServer(t) p, ok := newInstance(t, "api: yes\n"+ + "moq: no\n"+ "paths:\n"+ " source:\n"+ " forward:\n"+ @@ -258,6 +398,7 @@ func TestPathForwardRTMPReconnectsAfterSourceUnavailable(t *testing.T) { dest, received, serverErr := startRTMPForwardServer(t) p, ok := newInstance(t, "api: yes\n"+ + "moq: no\n"+ "paths:\n"+ " source:\n"+ " forward:\n"+ @@ -319,6 +460,7 @@ func TestPathForwardRTMPReconnectsAfterDestinationUnavailable(t *testing.T) { dest, received, _, serverErr := startRTMPForwardServerControlled(t, ready) p, ok := newInstance(t, "api: yes\n"+ + "moq: no\n"+ "paths:\n"+ " source:\n"+ " forward:\n"+ @@ -356,3 +498,59 @@ func TestPathForwardRTMPReconnectsAfterDestinationUnavailable(t *testing.T) { require.Equal(t, defs.APIForwardDestProtocolRTMP, item.Protocol) require.Greater(t, item.OutboundBytes, uint64(0)) } + +func TestPathForwardWHIP(t *testing.T) { + const bearerToken = "mytoken" + + dest, received, serverErr := startWHIPForwardServer(t, bearerToken) + + p, ok := newInstance(t, "api: yes\n"+ + "moq: no\n"+ + "paths:\n"+ + " source:\n"+ + " forward:\n"+ + " - dest: "+dest+"\n"+ + " whipBearerToken: "+bearerToken+"\n") + require.Equal(t, true, ok) + defer p.Close() + + source, w, track := startRTMPPublisher(t, "source") + defer source.Close() + + tr := &http.Transport{} + defer tr.CloseIdleConnections() + hc := &http.Client{Transport: tr} + + err := w.WriteH264(track, 2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) + require.NoError(t, err) + + require.Eventually(t, func() bool { + var path struct { + Ready bool `json:"ready"` + } + httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/paths/get/source", nil, &path) + return path.Ready + }, 5*time.Second, 100*time.Millisecond) + + var list defs.APIForwardDestList + httpRequest(t, hc, http.MethodGet, + "http://localhost:9997/v3/paths/forward/list?path=source", nil, &list) + require.Len(t, list.Items, 1) + added := list.Items[0] + require.Equal(t, dest, added.Conf.Dest) + require.Equal(t, bearerToken, added.Conf.WhipBearerToken) + require.Equal(t, defs.APIForwardDestProtocolWHIP, added.Protocol) + require.Equal(t, 1, added.Pos) + + waitWHIPForwardFrame(t, w, track, received, serverErr) + + require.Eventually(t, func() bool { + var item defs.APIForwardDest + httpRequest(t, hc, http.MethodGet, + "http://localhost:9997/v3/paths/forward/get?path=source&id="+added.ID.String(), nil, &item) + return item.State == defs.APIForwardDestStateForwarding && + item.Protocol == defs.APIForwardDestProtocolWHIP && + item.Conf.WhipBearerToken == bearerToken && + item.OutboundBytes > 0 + }, 5*time.Second, 100*time.Millisecond) +} diff --git a/internal/defs/api_forward.go b/internal/defs/api_forward.go index 1b40ced1..34072d04 100644 --- a/internal/defs/api_forward.go +++ b/internal/defs/api_forward.go @@ -28,6 +28,8 @@ const ( APIForwardDestProtocolRTSP APIForwardDestProtocol = "rtsp" APIForwardDestProtocolRTSPS APIForwardDestProtocol = "rtsps" APIForwardDestProtocolSRT APIForwardDestProtocol = "srt" + APIForwardDestProtocolWHIP APIForwardDestProtocol = "whip" + APIForwardDestProtocolWHIPS APIForwardDestProtocol = "whips" ) // APIForwardDest is a forward destination. diff --git a/internal/forward/dest_handler.go b/internal/forward/dest_handler.go index 95cf76ee..e936e73c 100644 --- a/internal/forward/dest_handler.go +++ b/internal/forward/dest_handler.go @@ -17,6 +17,7 @@ import ( forwardrtmp "github.com/bluenviron/mediamtx/internal/forward/rtmp" forwardrtsp "github.com/bluenviron/mediamtx/internal/forward/rtsp" forwardsrt "github.com/bluenviron/mediamtx/internal/forward/srt" + forwardwebrtc "github.com/bluenviron/mediamtx/internal/forward/webrtc" "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/stream" ) @@ -128,6 +129,12 @@ func destProtocol(dest string) defs.APIForwardDestProtocol { case strings.HasPrefix(dest, "srt://"): return defs.APIForwardDestProtocolSRT + case strings.HasPrefix(dest, "whip://"): + return defs.APIForwardDestProtocolWHIP + + case strings.HasPrefix(dest, "whips://"): + return defs.APIForwardDestProtocolWHIPS + default: panic("should not happen") } @@ -202,6 +209,15 @@ func (h *DestHandler) runOnce(strm *stream.Stream) error { Parent: h, } + case defs.APIForwardDestProtocolWHIP, defs.APIForwardDestProtocolWHIPS: + dest = &forwardwebrtc.Dest{ + Stream: strm, + Dest: resolvedDest, + ReadTimeout: h.ReadTimeout, + WhipBearerToken: h.Conf.WhipBearerToken, + Parent: h, + } + default: panic("should not happen") } diff --git a/internal/forward/dest_handler_test.go b/internal/forward/dest_handler_test.go index 7cf5566e..a22e59c3 100644 --- a/internal/forward/dest_handler_test.go +++ b/internal/forward/dest_handler_test.go @@ -6,6 +6,54 @@ import ( "github.com/stretchr/testify/require" ) +func TestDestProtocol(t *testing.T) { + for _, ca := range []struct { + name string + dest string + expected string + }{ + { + name: "rtmp", + dest: "rtmp://example.com/live/stream", + expected: "rtmp", + }, + { + name: "rtmps", + dest: "rtmps://example.com/live/stream", + expected: "rtmps", + }, + { + name: "rtsp", + dest: "rtsp://example.com/live/stream", + expected: "rtsp", + }, + { + name: "rtsps", + dest: "rtsps://example.com/live/stream", + expected: "rtsps", + }, + { + name: "srt", + dest: "srt://example.com:9000?streamid=publish:test", + expected: "srt", + }, + { + name: "whip", + dest: "whip://example.com/live/stream/whip", + expected: "whip", + }, + { + name: "whips", + dest: "whips://example.com/live/stream/whip", + expected: "whips", + }, + } { + t.Run(ca.name, func(t *testing.T) { + require.Equal(t, ca.expected, string(destProtocol(ca.dest))) + }) + } +} + func TestResolveDest(t *testing.T) { for _, ca := range []struct { name string diff --git a/internal/forward/manager_test.go b/internal/forward/manager_test.go index 3c3f8d7b..1c4e737a 100644 --- a/internal/forward/manager_test.go +++ b/internal/forward/manager_test.go @@ -131,7 +131,7 @@ func TestManagerReloadConf(t *testing.T) { m.ReloadConf(conf.Forward{ {Dest: "rtmp://localhost:5788/app/stream"}, // unchanged - {Dest: "srt://localhost:5790?streamid=publish:test"}, + {Dest: "whip://localhost:5790/teststream/whip", WhipBearerToken: "mytoken"}, {Dest: "rtsp://localhost:5789/stream"}, }) @@ -148,11 +148,14 @@ func TestManagerReloadConf(t *testing.T) { LastError: list2.Items[0].LastError, }, { - ID: list2.Items[1].ID, - Pos: 2, - Created: list2.Items[1].Created, - Conf: conf.ForwardDest{Dest: "srt://localhost:5790?streamid=publish:test"}, - Protocol: "srt", + ID: list2.Items[1].ID, + Pos: 2, + Created: list2.Items[1].Created, + Conf: conf.ForwardDest{ + Dest: "whip://localhost:5790/teststream/whip", + WhipBearerToken: "mytoken", + }, + Protocol: "whip", State: list2.Items[1].State, LastError: list2.Items[1].LastError, }, diff --git a/internal/forward/webrtc/dest.go b/internal/forward/webrtc/dest.go new file mode 100644 index 00000000..79ae2d5a --- /dev/null +++ b/internal/forward/webrtc/dest.go @@ -0,0 +1,110 @@ +// Package webrtc contains the WebRTC/WHIP forward destination. +package webrtc + +import ( + "context" + "fmt" + "net/http" + "net/url" + "strings" + "sync" + "time" + + "github.com/bluenviron/mediamtx/internal/conf" + "github.com/bluenviron/mediamtx/internal/logger" + pwebrtc "github.com/bluenviron/mediamtx/internal/protocols/webrtc" + "github.com/bluenviron/mediamtx/internal/protocols/whip" + "github.com/bluenviron/mediamtx/internal/stream" +) + +// Dest is a WebRTC/WHIP forward destination. +type Dest struct { + Stream *stream.Stream + Dest string + ReadTimeout conf.Duration + WhipBearerToken string + Parent logger.Writer + + mutex sync.RWMutex + client *whip.Client +} + +// Log implements logger.Writer. +func (d *Dest) Log(level logger.Level, format string, args ...any) { + d.Parent.Log(level, format, args...) +} + +// OutboundBytes returns the number of bytes sent by the destination. +func (d *Dest) OutboundBytes() uint64 { + d.mutex.RLock() + client := d.client + d.mutex.RUnlock() + + if client == nil || client.PeerConnection() == nil { + return 0 + } + + return client.PeerConnection().Stats().BytesSent +} + +// Run runs the destination. +func (d *Dest) Run(ctx context.Context) error { + u, err := url.Parse(d.Dest) + if err != nil { + return err + } + + u.Scheme = strings.Replace(u.Scheme, "whip", "http", 1) + + hc := &http.Client{Timeout: time.Duration(d.ReadTimeout)} + + r := &stream.Reader{Parent: d} + pc := &pwebrtc.PeerConnection{} + + err = pwebrtc.FromStream(d.Stream.OrigDesc, r, pc) + if err != nil { + return err + } + + client := &whip.Client{ + URL: u, + Publish: true, + OutboundTracks: pc.OutboundTracks, + OutboundDataChannels: pc.OutboundDataChannels, + HTTPClient: hc, + BearerToken: d.WhipBearerToken, + Log: d, + } + if err = client.Initialize(ctx); err != nil { + return err + } + defer client.Close() //nolint:errcheck + + d.mutex.Lock() + d.client = client + d.mutex.Unlock() + defer func() { + d.mutex.Lock() + d.client = nil + d.mutex.Unlock() + }() + + d.Stream.AddReader(r) + defer d.Stream.RemoveReader(r) + + clientErr := make(chan error, 1) + go func() { + clientErr <- client.Wait() + }() + + select { + case err = <-r.Error(): + return err + + case err = <-clientErr: + return err + + case <-ctx.Done(): + return fmt.Errorf("terminated") + } +} diff --git a/internal/forward/webrtc/dest_test.go b/internal/forward/webrtc/dest_test.go new file mode 100644 index 00000000..a68d4f02 --- /dev/null +++ b/internal/forward/webrtc/dest_test.go @@ -0,0 +1,194 @@ +package webrtc_test + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "testing" + "time" + + "github.com/bluenviron/gortsplib/v5/pkg/description" + "github.com/bluenviron/gortsplib/v5/pkg/format" + "github.com/pion/rtp" + pwebrtc "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" + + "github.com/bluenviron/mediamtx/internal/conf" + forwardwebrtc "github.com/bluenviron/mediamtx/internal/forward/webrtc" + mtxwebrtc "github.com/bluenviron/mediamtx/internal/protocols/webrtc" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/test" + "github.com/bluenviron/mediamtx/internal/unit" +) + +func startWHIPServer( + t *testing.T, + expectedBearerToken string, +) (string, <-chan struct{}, <-chan error) { + pc := &mtxwebrtc.PeerConnection{ + LocalRandomUDP: true, + IPsFromInterfaces: true, + Log: test.NilLogger, + } + err := pc.Start() + require.NoError(t, err) + t.Cleanup(func() { + pc.Close() + }) + + received := make(chan struct{}, 1) + serverErr := make(chan error, 1) + + httpServ := &http.Server{ + Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if expectedBearerToken != "" { + require.Equal(t, "Bearer "+expectedBearerToken, r.Header.Get("Authorization")) + } + + switch { + case r.Method == http.MethodOptions && r.URL.Path == "/stream/whip": + w.Header().Set("Access-Control-Allow-Methods", "OPTIONS, GET, POST, DELETE") + w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type") + w.WriteHeader(http.StatusNoContent) + + case r.Method == http.MethodPost && r.URL.Path == "/stream/whip": + require.Equal(t, "application/sdp", r.Header.Get("Content-Type")) + + body, err2 := io.ReadAll(r.Body) + require.NoError(t, err2) + + offer := &pwebrtc.SessionDescription{ + Type: pwebrtc.SDPTypeOffer, + SDP: string(body), + } + + answer, err2 := pc.CreateFullAnswer(offer, false) + require.NoError(t, err2) + + w.Header().Set("Content-Type", "application/sdp") + w.Header().Set("Location", "/stream/whip/sessionid") + w.WriteHeader(http.StatusCreated) + _, err2 = w.Write([]byte(answer.SDP)) + require.NoError(t, err2) + + go func() { + err3 := pc.WaitUntilConnected(10 * time.Second) + if err3 != nil { + serverErr <- err3 + return + } + + err3 = pc.GatherInboundTracks(2 * time.Second) + if err3 != nil { + serverErr <- err3 + return + } + + if len(pc.InboundTracks()) != 1 { + serverErr <- fmt.Errorf("unexpected track count: %d", len(pc.InboundTracks())) + return + } + + pc.InboundTracks()[0].OnPacketRTP = func(_ *rtp.Packet) { + select { + case received <- struct{}{}: + default: + } + } + + pc.StartReading() + }() + + case r.Method == http.MethodDelete && r.URL.Path == "/stream/whip/sessionid": + w.WriteHeader(http.StatusOK) + + default: + serverErr <- fmt.Errorf("unexpected request: %s %s", r.Method, r.URL.Path) + w.WriteHeader(http.StatusBadRequest) + } + }), + } + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + go httpServ.Serve(ln) + + t.Cleanup(func() { + httpServ.Shutdown(context.Background()) + }) + + return "whip://" + ln.Addr().String() + "/stream/whip", received, serverErr +} + +func TestDest(t *testing.T) { + const bearerToken = "mytoken" + + destURL, received, serverErr := startWHIPServer(t, bearerToken) + + desc := &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{test.FormatH264}, + }}} + strm := &stream.Stream{ + OrigDesc: desc, + WriteQueueSize: 512, + RTPMaxPayloadSize: 1450, + Parent: test.NilLogger, + } + require.NoError(t, strm.Initialize()) + defer strm.Close() + + subStream := &stream.SubStream{ + Stream: strm, + UseRTPPackets: false, + } + require.NoError(t, subStream.Initialize()) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + dest := &forwardwebrtc.Dest{ + Stream: strm, + Dest: destURL, + ReadTimeout: conf.Duration(10 * time.Second), + WhipBearerToken: bearerToken, + Parent: test.NilLogger, + } + + done := make(chan error, 1) + go func() { + done <- dest.Run(ctx) + }() + + strm.WaitForReaders() + for i := range 2 { + subStream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.Unit{ + PTS: int64(i) * 2 * 90000, + Payload: unit.PayloadH264{{5, 2}}, + }) + } + + select { + case <-received: + case err := <-serverErr: + require.NoError(t, err) + case runErr := <-done: + t.Fatalf("WHIP destination stopped before forwarding a frame: %v", runErr) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for WHIP frame") + } + + require.Eventually(t, func() bool { + return dest.OutboundBytes() > 0 + }, 5*time.Second, 10*time.Millisecond) + + cancel() + select { + case runErr := <-done: + require.Error(t, runErr) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for WHIP destination to stop") + } +} diff --git a/internal/protocols/whip/client.go b/internal/protocols/whip/client.go index ab9f2e03..a9d7b9e2 100644 --- a/internal/protocols/whip/client.go +++ b/internal/protocols/whip/client.go @@ -75,17 +75,18 @@ func offerAndCandidateToSDPFragment( // Client is a WHIP client. type Client struct { - URL *url.URL - Publish bool - OutboundTracks []*webrtc.OutboundTrack - HTTPClient *http.Client - BearerToken string - UDPReadBufferSize uint - SupportsIPv6 bool - STUNGatherTimeout time.Duration - HandshakeTimeout time.Duration - TrackGatherTimeout time.Duration - Log logger.Writer + URL *url.URL + Publish bool + OutboundTracks []*webrtc.OutboundTrack + OutboundDataChannels []*webrtc.OutboundDataChannel + HTTPClient *http.Client + BearerToken string + UDPReadBufferSize uint + SupportsIPv6 bool + STUNGatherTimeout time.Duration + HandshakeTimeout time.Duration + TrackGatherTimeout time.Duration + Log logger.Writer pc *webrtc.PeerConnection useTrickleICE bool @@ -109,15 +110,16 @@ func (c *Client) Initialize(ctx context.Context) error { } c.pc = &webrtc.PeerConnection{ - Net: &webrtc.Net{UDPReadBufferSize: int(c.UDPReadBufferSize)}, - LocalRandomUDP: true, - SupportsIPv6: c.SupportsIPv6, - ICEServers: iceServers, - IPsFromInterfaces: true, - Publish: c.Publish, - STUNGatherTimeout: c.STUNGatherTimeout, - OutboundTracks: c.OutboundTracks, - Log: c.Log, + Net: &webrtc.Net{UDPReadBufferSize: int(c.UDPReadBufferSize)}, + LocalRandomUDP: true, + SupportsIPv6: c.SupportsIPv6, + ICEServers: iceServers, + IPsFromInterfaces: true, + Publish: c.Publish, + STUNGatherTimeout: c.STUNGatherTimeout, + OutboundTracks: c.OutboundTracks, + OutboundDataChannels: c.OutboundDataChannels, + Log: c.Log, } err = c.pc.Start() if err != nil { diff --git a/internal/staticsources/webrtc/source.go b/internal/staticsources/webrtc/source.go index 80231a1b..78c5540a 100644 --- a/internal/staticsources/webrtc/source.go +++ b/internal/staticsources/webrtc/source.go @@ -67,7 +67,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { tr.TLSClientConfig = tlsConfig } - u.Scheme = strings.ReplaceAll(u.Scheme, "whep", "http") + u.Scheme = strings.Replace(u.Scheme, "whep", "http", 1) client := whip.Client{ URL: u, diff --git a/mediamtx.yml b/mediamtx.yml index 923d15ee..0ea85782 100644 --- a/mediamtx.yml +++ b/mediamtx.yml @@ -547,10 +547,14 @@ pathDefaults: # * rtmp://user:pass@host:port/path#streamKey -> the stream is forwarded to another RTMP server # * rtmps://user:pass@host:port/path#streamKey -> the stream is forwarded to another RTMP server with RTMPS # * srt://host:port?streamid=streamid -> the stream is forwarded to another SRT server + # * whip://host:port/path/whip -> the stream is forwarded to another WebRTC server with WHIP over HTTP + # * whips://host:port/path/whip -> the stream is forwarded to another WebRTC server with WHIP over HTTPS # The following variables can be used: # * $MTX_PATH: path name # * $G1, $G2, ...: regular expression groups, if path name is a regular expression. # - dest: + # # Token to insert in the Authorization: Bearer header when using WHIP. + # whipBearerToken: "" ############################################### # Default path settings -> Record