diff --git a/media/whip_server.go b/media/whip_server.go index bf608587f5..b2207914da 100644 --- a/media/whip_server.go +++ b/media/whip_server.go @@ -31,7 +31,7 @@ import ( "github.com/pion/webrtc/v4" ) -// TODO handle PATCH/PUT for ICE restarts (new Offers) and DELETE +// TODO handle PATCH/PUT for ICE restarts (new Offers) const ( keyframeInterval = 2 * time.Second // TODO make configurable? @@ -65,6 +65,45 @@ var WebrtcConfig = webrtc.Configuration{ type WHIPServer struct { mediaEngine *webrtc.MediaEngine settings func(*webrtc.API) + + // Live sessions by resource ID, so DELETE can find the connection it + // is asked to terminate. Entries are removed when the session ends, + // whether that is by DELETE, by the peer, or by an error. + mu sync.Mutex + resources map[string]*MediaState +} + +func (s *WHIPServer) track(resourceID string, ms *MediaState) { + s.mu.Lock() + if s.resources == nil { + s.resources = map[string]*MediaState{} + } + s.resources[resourceID] = ms + s.mu.Unlock() + + // Drop the entry once the session ends by any route; MediaState.Close + // is idempotent, so a DELETE that races this is harmless. + go func() { + _ = ms.AwaitClose() + s.mu.Lock() + delete(s.resources, resourceID) + s.mu.Unlock() + }() +} + +// DeleteWHIP terminates the session identified by resourceID, per the WHIP +// spec's DELETE on the resource URL returned in Location. Reports whether +// the resource existed. +func (s *WHIPServer) DeleteWHIP(ctx context.Context, resourceID string) bool { + s.mu.Lock() + ms, ok := s.resources[resourceID] + s.mu.Unlock() + if !ok { + return false + } + clog.Infof(ctx, "deleting whip resource=%s", resourceID) + ms.Close() + return true } // handleCreate implements the POST that creates a new resource. @@ -158,6 +197,8 @@ func (s *WHIPServer) CreateWHIP(ctx context.Context, ssr *SwitchableSegmentReade resourceID := generateID() etag := generateETag() + s.track(resourceID, mediaState) + // Respond with 201 Created resourceURL := fmt.Sprintf("%s/%s", getRequestURL(r), resourceID) w.Header().Set("Content-Type", "application/sdp") diff --git a/media/whip_server_test.go b/media/whip_server_test.go new file mode 100644 index 0000000000..0bcc5df54e --- /dev/null +++ b/media/whip_server_test.go @@ -0,0 +1,40 @@ +package media + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestWHIPServerDeleteResource(t *testing.T) { + ctx := context.Background() + s := &WHIPServer{} + ms := NewMediaState(&MockPC{}) + s.track("res-1", ms) + + require.False(t, s.DeleteWHIP(ctx, "unknown"), "unknown resource") + require.True(t, s.DeleteWHIP(ctx, "res-1"), "known resource") + require.True(t, ms.IsClosed(), "connection closed") + + // The entry is dropped once the session ends, so the resource is gone + // for any later request. + require.Eventually(t, func() bool { + return !s.DeleteWHIP(ctx, "res-1") + }, time.Second, 10*time.Millisecond) +} + +// A session that ends on its own must not leave its resource behind. +func TestWHIPServerUntracksOnClose(t *testing.T) { + ctx := context.Background() + s := &WHIPServer{} + ms := NewMediaState(&MockPC{}) + s.track("res-2", ms) + + ms.Close() + + require.Eventually(t, func() bool { + return !s.DeleteWHIP(ctx, "res-2") + }, time.Second, 10*time.Millisecond) +} diff --git a/server/ai_mediaserver.go b/server/ai_mediaserver.go index 8b80e4e8e7..168aecd932 100644 --- a/server/ai_mediaserver.go +++ b/server/ai_mediaserver.go @@ -106,6 +106,9 @@ func startAIMediaServer(ctx context.Context, ls *LivepeerServer) error { ls.HTTPMux.Handle("POST /live/video-to-video/{stream}/whip", ls.CreateWhip(whipServer)) ls.HTTPMux.Handle("HEAD /live/video-to-video/{stream}/whip", ls.WithCode(http.StatusMethodNotAllowed)) ls.HTTPMux.Handle("OPTIONS /live/video-to-video/{stream}/whip", ls.WithCode(http.StatusNoContent)) + // Resource URL handed to the client in Location on create. + ls.HTTPMux.Handle("DELETE /live/video-to-video/{stream}/whip/{resource}", ls.DeleteWhip(whipServer)) + ls.HTTPMux.Handle("OPTIONS /live/video-to-video/{stream}/whip/{resource}", ls.WithCode(http.StatusNoContent)) } var whepServer *media.WHEPServer @@ -950,6 +953,26 @@ func (ls *LivepeerServer) GetLiveVideoToVideoStatus() http.Handler { }) } +// DeleteWhip terminates a WHIP session via DELETE on the resource URL +// returned in the Location header at create time (WHIP spec, ingest +// session termination). 404 if the resource is unknown or already gone. +func (ls *LivepeerServer) DeleteWhip(server *media.WHIPServer) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + corsHeaders(w, r.Method) + resource := r.PathValue("resource") + if resource == "" { + http.Error(w, "Missing resource", http.StatusBadRequest) + return + } + ctx := clog.AddVal(r.Context(), "stream", r.PathValue("stream")) + if !server.DeleteWHIP(ctx, resource) { + http.Error(w, "Resource not found", http.StatusNotFound) + return + } + w.WriteHeader(http.StatusOK) + }) +} + func (ls *LivepeerServer) CreateWhip(server *media.WHIPServer) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // two main sequential parts here