diff --git a/internal/protocols/webrtc/peer_connection.go b/internal/protocols/webrtc/peer_connection.go index c3ebc82f..c64510f0 100644 --- a/internal/protocols/webrtc/peer_connection.go +++ b/internal/protocols/webrtc/peer_connection.go @@ -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) diff --git a/internal/protocols/whip/client.go b/internal/protocols/whip/client.go index 2f865521..1b1a1f44 100644 --- a/internal/protocols/whip/client.go +++ b/internal/protocols/whip/client.go @@ -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 diff --git a/internal/protocols/whip/client_test.go b/internal/protocols/whip/client_test.go index 19d20fb2..3f3a6b12 100644 --- a/internal/protocols/whip/client_test.go +++ b/internal/protocols/whip/client_test.go @@ -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 +} diff --git a/internal/servers/webrtc/http_server.go b/internal/servers/webrtc/http_server.go index c8a0616d..5a2dbaaf 100644 --- a/internal/servers/webrtc/http_server.go +++ b/internal/servers/webrtc/http_server.go @@ -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) diff --git a/internal/staticsources/webrtc/source_test.go b/internal/staticsources/webrtc/source_test.go index eb03c5f2..baf329da 100644 --- a/internal/staticsources/webrtc/source_test.go +++ b/internal/staticsources/webrtc/source_test.go @@ -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)