OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

webrtc: support forwarding streams (#6099)

Browse Source
This commit is contained in:
Alessandro Ros
2026-08-16 21:09:59 +02:00
committed by GitHub
parent c9f003f247
commit 831ed9564b
15 changed files with 698 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/openapi.yaml
+4
View File
@@ -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
docs/2-features/11-forward.md
+27 -12
View File
@@ -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:
internal/api/api_forward_test.go
+43 -12
View File
@@ -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)
}
internal/conf/conf_test.go
+16 -2
View File
@@ -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)
}
})
}
}
internal/conf/forward_dest.go
+4 -3
View File
@@ -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)
}
internal/core/forward_test.go
+198
View File
@@ -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)
}
internal/defs/api_forward.go
+2
View File
@@ -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.
internal/forward/dest_handler.go
+16
View File
@@ -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")
}
internal/forward/dest_handler_test.go
+48
View File
@@ -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
internal/forward/manager_test.go
+9 -6
View File
@@ -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,
},
internal/forward/webrtc/dest.go
+110
View File
@@ -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")
}
}
internal/forward/webrtc/dest_test.go
+194
View File
@@ -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")
}
}
internal/protocols/whip/client.go
+22 -20
View File
@@ -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 {
internal/staticsources/webrtc/source.go
+1 -1
View File
@@ -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,
mediamtx.yml
+4
View File
@@ -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