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

webrtc: support interacting with servers with no trickle ICE (#5273) (#5757)

Browse Source
This commit is contained in:
Alessandro Ros
2026-05-12 21:07:16 +02:00
committed by GitHub
parent f542ccaf19
commit cdea2dc167
5 changed files with 220 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/protocols/webrtc/peer_connection.go
+28
View File
@@ -629,6 +629,34 @@ func (co *PeerConnection) CreatePartialOffer() (*webrtc.SessionDescription, erro
return offer, nil
}
// CreateFullOffer creates a full offer.
func (co *PeerConnection) CreateFullOffer() (*webrtc.SessionDescription, error) {
tmp, err := co.wr.CreateOffer(nil)
if err != nil {
return nil, err
}
offer := &tmp
err = co.wr.SetLocalDescription(*offer)
if err != nil {
return nil, err
}
err = co.waitGatheringDone()
if err != nil {
return nil, err
}
offer = co.wr.LocalDescription()
offer, err = co.filterLocalDescription(offer)
if err != nil {
return nil, err
}
return offer, nil
}
// SetAnswer sets the answer.
func (co *PeerConnection) SetAnswer(answer *webrtc.SessionDescription) error {
return co.wr.SetRemoteDescription(*answer)
internal/protocols/whip/client.go
+66 -36
View File
@@ -8,6 +8,7 @@ import (
"io"
"net/http"
"net/url"
"strings"
"time"
"github.com/pion/sdp/v3"
@@ -35,8 +36,8 @@ type Client struct {
TrackGatherTimeout time.Duration
Log logger.Writer
pc *webrtc.PeerConnection
patchIsSupported bool
pc *webrtc.PeerConnection
useTrickleICE bool
}
// Initialize initializes the Client.
@@ -95,9 +96,19 @@ func (c *Client) Initialize(ctx context.Context) error {
}
func (c *Client) initializeInner(ctx context.Context) error {
offer, err := c.pc.CreatePartialOffer()
if err != nil {
return err
var offer *pwebrtc.SessionDescription
if c.useTrickleICE {
var err error
offer, err = c.pc.CreatePartialOffer()
if err != nil {
return err
}
} else {
var err error
offer, err = c.pc.CreateFullOffer()
if err != nil {
return err
}
}
res, err := c.postOffer(ctx, offer)
@@ -131,28 +142,10 @@ func (c *Client) initializeInner(ctx context.Context) error {
return err
}
t := time.NewTimer(c.HandshakeTimeout)
defer t.Stop()
outer:
for {
select {
case ca := <-c.pc.NewLocalCandidate():
err = c.patchCandidate(ctx, offer, res.ETag, ca)
if err != nil {
c.deleteSession(context.Background()) //nolint:errcheck
return err
}
case <-c.pc.GatheringDone():
case <-c.pc.Connected():
break outer
case <-t.C:
c.deleteSession(context.Background()) //nolint:errcheck
return fmt.Errorf("deadline exceeded while waiting connection")
}
err = c.waitConnected(ctx, offer, res.ETag)
if err != nil {
c.deleteSession(context.Background()) //nolint:errcheck
return err
}
if !c.Publish {
@@ -166,6 +159,39 @@ outer:
return nil
}
func (c *Client) waitConnected(ctx context.Context, offer *pwebrtc.SessionDescription, eTag string) error {
t := time.NewTimer(c.HandshakeTimeout)
defer t.Stop()
if c.useTrickleICE {
for {
select {
case ca := <-c.pc.NewLocalCandidate():
err := c.patchCandidate(ctx, offer, eTag, ca)
if err != nil {
return err
}
case <-c.pc.Connected():
return nil
case <-t.C:
return fmt.Errorf("deadline exceeded while waiting connection")
}
}
}
for {
select {
case <-c.pc.Connected():
return nil
case <-t.C:
return fmt.Errorf("deadline exceeded while waiting connection")
}
}
}
// PeerConnection returns the underlying peer connection.
func (c *Client) PeerConnection() *webrtc.PeerConnection {
return c.pc
@@ -216,6 +242,13 @@ func (c *Client) optionsICEServers(
return nil, fmt.Errorf("bad status code: %v", res.StatusCode)
}
for m := range strings.SplitSeq(res.Header.Get("Access-Control-Allow-Methods"), ",") {
if strings.TrimSpace(m) == "PATCH" {
c.useTrickleICE = true
break
}
}
return LinkHeaderUnmarshal(res.Header["Link"])
}
@@ -255,13 +288,14 @@ func (c *Client) postOffer(
return nil, fmt.Errorf("bad Content-Type: expected 'application/sdp', got '%s'", contentType)
}
c.patchIsSupported = (res.Header.Get("Accept-Patch") == "application/trickle-ice-sdpfrag")
Location := res.Header.Get("Location")
etag := res.Header.Get("ETag")
if etag == "" {
return nil, fmt.Errorf("ETag is missing")
var etag string
if c.useTrickleICE {
etag = res.Header.Get("ETag")
if etag == "" {
return nil, fmt.Errorf("ETag is missing")
}
}
sdp, err := io.ReadAll(&customLimitReader{res.Body, maxInboundSDPSize})
@@ -287,10 +321,6 @@ func (c *Client) patchCandidate(
etag string,
candidate *pwebrtc.ICECandidateInit,
) error {
if !c.patchIsSupported {
return nil
}
frag, err := ICEFragmentMarshal(offer.SDP, []*pwebrtc.ICECandidateInit{candidate})
if err != nil {
return err
internal/protocols/whip/client_test.go
+120 -2
View File
@@ -101,11 +101,12 @@ func TestClientRead(t *testing.T) {
require.NoError(t, err2)
offer := whipOffer(body)
require.NotContains(t, offer.SDP, "a=candidate:")
answer, err2 := pc.CreateFullAnswer(offer)
require.NoError(t, err2)
w.Header().Set("Content-Type", "application/sdp")
w.Header().Set("Accept-Patch", "application/trickle-ice-sdpfrag")
w.Header().Set("ETag", "test_etag")
w.Header().Set("Location", "/my/resource/sessionid")
w.WriteHeader(http.StatusCreated)
@@ -270,11 +271,12 @@ func TestClientPublish(t *testing.T) {
require.NoError(t, err2)
offer := whipOffer(body)
require.NotContains(t, offer.SDP, "a=candidate:")
answer, err2 := pc.CreateFullAnswer(offer)
require.NoError(t, err2)
w.Header().Set("Content-Type", "application/sdp")
w.Header().Set("Accept-Patch", "application/trickle-ice-sdpfrag")
w.Header().Set("ETag", "test_etag")
w.Header().Set("Location", "/my/resource/sessionid")
w.WriteHeader(http.StatusCreated)
@@ -499,3 +501,119 @@ func TestClientBearerToken(t *testing.T) {
require.NoError(t, err)
defer cl.Close() //nolint:errcheck
}
func TestClientNoTrickleICE(t *testing.T) {
pc := &webrtc.PeerConnection{
LocalRandomUDP: true,
IPsFromInterfaces: true,
Log: test.NilLogger,
}
err := pc.Start()
require.NoError(t, err)
defer pc.Close()
state := 0
recv := make(chan struct{})
httpServ := &http.Server{
Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch state {
case 0:
require.Equal(t, http.MethodOptions, r.Method)
require.Equal(t, "/my/resource", r.URL.Path)
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 1:
require.Equal(t, http.MethodPost, r.Method)
require.Equal(t, "/my/resource", r.URL.Path)
require.Equal(t, "application/sdp", r.Header.Get("Content-Type"))
body, err2 := io.ReadAll(r.Body)
require.NoError(t, err2)
offer := whipOffer(body)
require.Contains(t, offer.SDP, "a=candidate:")
answer, err2 := pc.CreateFullAnswer(offer)
require.NoError(t, err2)
w.Header().Set("Content-Type", "application/sdp")
w.Header().Set("Location", "/my/resource/sessionid")
w.WriteHeader(http.StatusCreated)
w.Write([]byte(answer.SDP))
go func() {
err3 := pc.WaitUntilConnected(10 * time.Second)
require.NoError(t, err3)
err3 = pc.GatherIncomingTracks(2 * time.Second)
require.NoError(t, err3)
pc.IncomingTracks()[0].OnPacketRTP = func(_ *rtp.Packet) {
close(recv)
}
pc.StartReading()
}()
default:
require.Equal(t, "/my/resource/sessionid", r.URL.Path)
switch r.Method {
case http.MethodDelete:
w.WriteHeader(http.StatusOK)
default:
t.Errorf("unexpected method: %s", r.Method)
}
}
state++
}),
}
ln, err := net.Listen("tcp", "localhost:9005")
require.NoError(t, err)
go httpServ.Serve(ln)
defer httpServ.Shutdown(context.Background())
u, err := url.Parse("http://localhost:9005/my/resource")
require.NoError(t, err)
outgoingTracks := []*webrtc.OutgoingTrack{{
Caps: pwebrtc.RTPCodecCapability{
MimeType: "audio/opus",
ClockRate: 48000,
Channels: 2,
},
}}
cl := &Client{
URL: u,
Publish: true,
OutgoingTracks: outgoingTracks,
HTTPClient: &http.Client{},
Log: test.NilLogger,
}
err = cl.Initialize(context.Background())
require.NoError(t, err)
defer cl.Close() //nolint:errcheck
err = 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)
<-recv
}
internal/servers/webrtc/http_server.go
+6 -1
View File
@@ -187,7 +187,8 @@ func (s *httpServer) onWHIPOptions(ctx *gin.Context, pathName string, publish bo
ctx.Header("Access-Control-Allow-Methods", "OPTIONS, GET, POST, PATCH, DELETE")
ctx.Header("Access-Control-Allow-Headers", "Authorization, Content-Type, If-Match")
ctx.Header("Access-Control-Expose-Headers", "Link")
ctx.Header("Access-Control-Expose-Headers", "Accept-Post, Link")
ctx.Header("Accept-Post", "application/sdp")
ctx.Writer.Header()["Link"] = whip.LinkHeaderMarshal(servers)
ctx.Writer.WriteHeader(http.StatusNoContent)
}
@@ -243,7 +244,11 @@ func (s *httpServer) onWHIPPost(ctx *gin.Context, pathName string, publish bool)
ctx.Header("Access-Control-Expose-Headers", "ETag, ID, Accept-Patch, Link, Location")
ctx.Header("ETag", "*")
ctx.Header("ID", res.sx.uuid.String())
// Accept-Patch has been removed from WHIP/WHEP specifications
// but is kept here for compatibility reasons.
ctx.Header("Accept-Patch", "application/trickle-ice-sdpfrag")
ctx.Writer.Header()["Link"] = whip.LinkHeaderMarshal(servers)
ctx.Header("Location", sessionLocation(publish, pathName, ctx.Request.URL.RawQuery, res.sx.secret))
ctx.Writer.WriteHeader(http.StatusCreated)
internal/staticsources/webrtc/source_test.go
-1
View File
@@ -72,7 +72,6 @@ func TestSource(t *testing.T) {
require.NoError(t, err2)
w.Header().Set("Content-Type", "application/sdp")
w.Header().Set("Accept-Patch", "application/trickle-ice-sdpfrag")
w.Header().Set("ETag", "test_etag")
w.Header().Set("Location", "/my/resource/sessionid")
w.WriteHeader(http.StatusCreated)