diff --git a/internal/protocols/webrtc/peer_connection.go b/internal/protocols/webrtc/peer_connection.go index 0050f0a5..c105c5aa 100644 --- a/internal/protocols/webrtc/peer_connection.go +++ b/internal/protocols/webrtc/peer_connection.go @@ -36,9 +36,12 @@ type PeerConnection struct { newLocalCandidate chan *webrtc.ICECandidateInit connected chan struct{} disconnected chan struct{} - closed chan struct{} + done chan struct{} gatheringDone chan struct{} incomingTrack chan trackRecvPair + + ctx context.Context + ctxCancel context.CancelFunc } // Start starts the peer connection. @@ -56,10 +59,12 @@ func (co *PeerConnection) Start() error { co.newLocalCandidate = make(chan *webrtc.ICECandidateInit) co.connected = make(chan struct{}) co.disconnected = make(chan struct{}) - co.closed = make(chan struct{}) + co.done = make(chan struct{}) co.gatheringDone = make(chan struct{}) co.incomingTrack = make(chan trackRecvPair) + co.ctx, co.ctxCancel = context.WithCancel(context.Background()) + if !co.Publish { _, err = co.wr.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RtpTransceiverInit{ Direction: webrtc.RTPTransceiverDirectionRecvonly, @@ -80,7 +85,7 @@ func (co *PeerConnection) Start() error { co.wr.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) { select { case co.incomingTrack <- trackRecvPair{track, receiver}: - case <-co.closed: + case <-co.ctx.Done(): } }) } @@ -90,7 +95,7 @@ func (co *PeerConnection) Start() error { defer co.stateChangeMutex.Unlock() select { - case <-co.closed: + case <-co.done: return default: } @@ -108,7 +113,7 @@ func (co *PeerConnection) Start() error { close(co.disconnected) case webrtc.PeerConnectionStateClosed: - close(co.closed) + close(co.done) } }) @@ -118,7 +123,7 @@ func (co *PeerConnection) Start() error { select { case co.newLocalCandidate <- &v: case <-co.connected: - case <-co.closed: + case <-co.ctx.Done(): } } else { close(co.gatheringDone) @@ -130,8 +135,9 @@ func (co *PeerConnection) Start() error { // Close closes the connection. func (co *PeerConnection) Close() { + co.ctxCancel() co.wr.Close() //nolint:errcheck - <-co.closed + <-co.done } // CreatePartialOffer creates a partial offer. diff --git a/internal/protocols/webrtc/peer_connection_test.go b/internal/protocols/webrtc/peer_connection_test.go new file mode 100644 index 00000000..f00e1fcb --- /dev/null +++ b/internal/protocols/webrtc/peer_connection_test.go @@ -0,0 +1,33 @@ +package webrtc + +import ( + "testing" + "time" + + "github.com/bluenviron/mediamtx/internal/test" + "github.com/stretchr/testify/require" +) + +func TestPeerConnectionCloseAfterError(t *testing.T) { + api, err := NewAPI(APIConf{ + LocalRandomUDP: true, + IPsFromInterfaces: true, + }) + require.NoError(t, err) + + pc := &PeerConnection{ + API: api, + Publish: false, + Log: test.NilLogger{}, + } + err = pc.Start() + require.NoError(t, err) + + _, err = pc.CreatePartialOffer() + require.NoError(t, err) + + // wait for ICE candidates to be generated + time.Sleep(500 * time.Millisecond) + + pc.Close() +} diff --git a/internal/protocols/webrtc/whip_client.go b/internal/protocols/webrtc/whip_client.go index eb129403..2c4d5a01 100644 --- a/internal/protocols/webrtc/whip_client.go +++ b/internal/protocols/webrtc/whip_client.go @@ -89,7 +89,7 @@ outer: for { select { case ca := <-c.pc.NewLocalCandidate(): - err := WHIPPatchCandidate(context.Background(), c.HTTPClient, c.URL.String(), offer, res.ETag, ca) + err := WHIPPatchCandidate(ctx, c.HTTPClient, c.URL.String(), offer, res.ETag, ca) if err != nil { c.pc.Close() return nil, err @@ -180,7 +180,7 @@ outer: for { select { case ca := <-c.pc.NewLocalCandidate(): - err := WHIPPatchCandidate(context.Background(), c.HTTPClient, c.URL.String(), offer, res.ETag, ca) + err := WHIPPatchCandidate(ctx, c.HTTPClient, c.URL.String(), offer, res.ETag, ca) if err != nil { c.pc.Close() return nil, err