Use InputVideo/AudioState from WHIPRTCConnectionNotify. (#384)

Also terminate the session if we haven't received any Notify in 3 min

Author: biglittlebigben

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/livekit/go-rtmp v0.0.0-20230829211117-1c4f5a5c81ed
 	github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731
 	github.com/livekit/mediatransportutil v0.0.0-20241220010243-a2bdee945564
-	github.com/livekit/protocol v1.41.1-0.20250923010807-539194425c60
+	github.com/livekit/protocol v1.42.1-0.20251006173230-a7bc52ad9724
 	github.com/livekit/psrpc v0.7.0
 	github.com/livekit/server-sdk-go/v2 v2.4.2
 	github.com/pion/dtls/v3 v3.0.7

go.sum

@@ -116,8 +116,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT
 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20241220010243-a2bdee945564 h1:GX7KF/V9ExmcfT/2Bdia8aROjkxrgx7WpyH7w9MB4J4=
 github.com/livekit/mediatransportutil v0.0.0-20241220010243-a2bdee945564/go.mod h1:36s+wwmU3O40IAhE+MjBWP3W71QRiEE9SfooSBvtBqY=
-github.com/livekit/protocol v1.41.1-0.20250923010807-539194425c60 h1:rdWGS5IIQwb/dX94mrBF0LsCxHiwHZruJxqxoP34WNg=
-github.com/livekit/protocol v1.41.1-0.20250923010807-539194425c60/go.mod h1:vhMS30QoEyH2p34vi6X1eWkC4EMV72ZGZwQb74ajY7A=
+github.com/livekit/protocol v1.42.1-0.20251006173230-a7bc52ad9724 h1:oz2RQahwZ8LPaGX+2RzAR08s+zqJJSep+OCvWuD90I4=
+github.com/livekit/protocol v1.42.1-0.20251006173230-a7bc52ad9724/go.mod h1:vhMS30QoEyH2p34vi6X1eWkC4EMV72ZGZwQb74ajY7A=
 github.com/livekit/psrpc v0.7.0 h1:rtfqfjYN06WJYloE/S0nmkJ/Y04x4pxLQLe8kQ4FVHU=
 github.com/livekit/psrpc v0.7.0/go.mod h1:AuDC5uOoEjQJEc69v4Li3t77Ocz0e0NdjQEuFfO+vfk=
 github.com/livekit/server-sdk-go/v2 v2.4.2 h1:q6ioBWpwLaLNj41f96eLQHi11kRyiY9MfEb5D3zi5AI=

pkg/whip/whip_proxy_handler.go

@@ -22,6 +22,7 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/frostbyte73/core"
@@ -38,6 +39,10 @@ import (
 	"github.com/livekit/ingress/pkg/utils"
 )
 
+const (
+	sessionNotifyTimeout = 3 * time.Minute
+)
+
 type proxyWhipHandler struct {
 	logger        logger.Logger
 	params        *params.Params
@@ -47,6 +52,9 @@ type proxyWhipHandler struct {
 
 	done      core.Fuse
 	rpcServer rpc.IngressHandlerServer
+
+	mu       sync.Mutex
+	watchdog *time.Timer
 }
 
 func NewProxyWHIPHandler(p *params.Params, bus psrpc.MessageBus, ua string) (WHIPHandler, error) {
@@ -183,6 +191,13 @@ func (h *proxyWhipHandler) close(isRTCClosed bool) {
 		return
 	}
 
+	h.mu.Lock()
+	if h.watchdog != nil {
+		h.watchdog.Stop()
+		h.watchdog = nil
+	}
+	h.mu.Unlock()
+
 	utils.DeregisterIngressRpcHandlers(h.rpcServer, h.params.IngressInfo)
 	if h.participantID != "" {
 		if isRTCClosed {
@@ -225,6 +240,8 @@ func (h *proxyWhipHandler) close(isRTCClosed bool) {
 }
 
 func (h *proxyWhipHandler) WaitForSessionEnd(ctx context.Context) error {
+	h.resetWatchDog()
+
 	select {
 	case <-h.done.Watch():
 	case <-ctx.Done():
@@ -313,11 +330,35 @@ func (h *proxyWhipHandler) ICERestartWHIPResource(ctx context.Context, req *rpc.
 }
 
 func (h *proxyWhipHandler) WHIPRTCConnectionNotify(ctx context.Context, req *rpc.WHIPRTCConnectionNotifyRequest) (*google_protobuf2.Empty, error) {
-	h.logger.Infow("WHIPRTCConnectionNotify", "participantID", h.participantID, "req", req)
+	tctx, done := context.WithTimeout(context.Background(), 10*time.Second)
+	defer done()
+
+	h.resetWatchDog()
+
+	if req.Video != nil {
+		h.params.SetInputVideoState(tctx, req.Video, true)
+	}
+
+	if req.Audio != nil {
+		h.params.SetInputAudioState(tctx, req.Audio, true)
+	}
 
 	if req.Closed {
 		h.close(true)
 	}
 
 	return &google_protobuf2.Empty{}, nil
 }
+
+func (h *proxyWhipHandler) resetWatchDog() {
+	h.mu.Lock()
+	if h.watchdog != nil {
+		h.watchdog.Stop()
+	}
+
+	h.watchdog = time.AfterFunc(sessionNotifyTimeout, func() {
+		h.logger.Infow("no Notify call from the SFU, terminating ingress")
+		h.done.Break()
+	})
+	h.mu.Unlock()
+}