whip: support DELETE on the resource URL

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")

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)
+}

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