update to make data channel optional

Author: ad-astra-video

server/ai_http.go

@@ -148,6 +148,26 @@ func (h *lphttp) StartLiveVideoToVideo() http.Handler {
 			dataUrl    = pubUrl + "-data"
 		)
 
+		//if data is not enabled remove the url and do not start the data channel
+		if enableData, ok := (*req.Params)["enableData"]; ok {
+			if val, ok := enableData.(bool); ok {
+				//turn off data channel if request sets to false
+				if !val {
+					dataUrl = ""
+				} else {
+					clog.Infof(ctx, "data channel is enabled")
+				}
+			} else {
+				clog.Warningf(ctx, "enableData is not a bool, got type %T", enableData)
+			}
+
+			//delete the param used for go-livepeer signaling
+			delete((*req.Params), "enableData")
+		} else {
+			//default to no data channel
+			dataUrl = ""
+		}
+
 		// Handle initial payment, the rest of the payments are done separately from the stream processing
 		// Note that this payment is debit from the balance and acts as a buffer for the AI Realtime Video processing
 		payment, err := getPayment(r.Header.Get(paymentHeader))
@@ -181,8 +201,13 @@ func (h *lphttp) StartLiveVideoToVideo() http.Handler {
 		controlPubCh.CreateChannel()
 		eventsCh := trickle.NewLocalPublisher(h.trickleSrv, mid+"-events", "application/json")
 		eventsCh.CreateChannel()
-		dataCh := trickle.NewLocalPublisher(h.trickleSrv, mid+"-data", "application/jsonl")
-		dataCh.CreateChannel()
+
+		//optional channels
+		var dataCh *trickle.TrickleLocalPublisher
+		if dataUrl != "" {
+			dataCh = trickle.NewLocalPublisher(h.trickleSrv, mid+"-data", "application/jsonl")
+			dataCh.CreateChannel()
+		}
 
 		// Start payment receiver which accounts the payments and stops the stream if the payment is insufficient
 		priceInfo := payment.GetExpectedPrice()
@@ -203,6 +228,9 @@ func (h *lphttp) StartLiveVideoToVideo() http.Handler {
 					subCh.Close()
 					eventsCh.Close()
 					controlPubCh.Close()
+					if dataCh != nil {
+						dataCh.Close()
+					}
 					cancel()
 				}
 				return err
@@ -230,11 +258,17 @@ func (h *lphttp) StartLiveVideoToVideo() http.Handler {
 		}()
 
 		// Prepare request to worker
+		// required channels
 		controlUrlOverwrite := overwriteHost(h.node.LiveAITrickleHostForRunner, controlUrl)
 		eventsUrlOverwrite := overwriteHost(h.node.LiveAITrickleHostForRunner, eventsUrl)
 		subscribeUrlOverwrite := overwriteHost(h.node.LiveAITrickleHostForRunner, pubUrl)
 		publishUrlOverwrite := overwriteHost(h.node.LiveAITrickleHostForRunner, subUrl)
-		dataUrlOverwrite := overwriteHost(h.node.LiveAITrickleHostForRunner, dataUrl)
+
+		// optional channels
+		var dataUrlOverwrite string
+		if dataCh != nil {
+			dataUrlOverwrite = overwriteHost(h.node.LiveAITrickleHostForRunner, dataUrl)
+		}
 
 		workerReq := worker.LiveVideoToVideoParams{
 			ModelId:          req.ModelId,
@@ -260,7 +294,9 @@ func (h *lphttp) StartLiveVideoToVideo() http.Handler {
 			subCh.Close()
 			controlPubCh.Close()
 			eventsCh.Close()
-			dataCh.Close()
+			if dataCh != nil {
+				dataCh.Close()
+			}
 			cancel()
 			respondWithError(w, err.Error(), http.StatusInternalServerError)
 			return
@@ -272,6 +308,7 @@ func (h *lphttp) StartLiveVideoToVideo() http.Handler {
 			SubscribeUrl: subUrl,
 			ControlUrl:   &controlUrl,
 			EventsUrl:    &eventsUrl,
+			DataUrl:      &dataUrl,
 			RequestId:    &requestID,
 			ManifestId:   &mid,
 		})

server/ai_live_video.go

@@ -763,6 +763,11 @@ func startEventsSubscribe(ctx context.Context, url *url.URL, params aiRequestPar
 }
 
 func startDataSubscribe(ctx context.Context, url *url.URL, params aiRequestParams, sess *AISession) {
+	//only start DataSubscribe if enabled
+	if params.liveParams.dataWriter == nil {
+		return
+	}
+
 	// subscribe to the outputs
 	subscriber, err := trickle.NewTrickleSubscriber(trickle.TrickleSubscriberConfig{
 		URL: url.String(),

server/ai_mediaserver.go

@@ -623,7 +623,6 @@ func (ls *LivepeerServer) StartLiveVideo() http.Handler {
 
 			liveParams: &liveRequestParams{
 				segmentReader:          ssr,
-				dataWriter:             media.NewSegmentWriter(5),
 				rtmpOutputs:            rtmpOutputs,
 				localRTMPPrefix:        mediaMTXInputURL,
 				stream:                 streamName,
@@ -638,6 +637,15 @@ func (ls *LivepeerServer) StartLiveVideo() http.Handler {
 			},
 		}
 
+		//create a dataWriter for data channel if enabled
+		if enableData, ok := pipelineParams["enableData"]; ok {
+			if enableData == true || enableData == "true" {
+				params.liveParams.dataWriter = media.NewSegmentWriter(5)
+				pipelineParams["enableData"] = true
+				clog.Infof(ctx, "Data channel enabled for stream %s", streamName)
+			}
+		}
+
 		registerControl(ctx, params)
 
 		// Create a special parent context for orchestrator cancellation
@@ -763,6 +771,8 @@ func startProcessing(ctx context.Context, params aiRequestParams, res interface{
 	resp := res.(*worker.GenLiveVideoToVideoResponse)
 
 	host := params.liveParams.sess.Transcoder()
+
+	//required channels
 	pub, err := common.AppendHostname(resp.JSON200.PublishUrl, host)
 	if err != nil {
 		return fmt.Errorf("invalid publish URL: %w", err)
@@ -779,21 +789,30 @@ func startProcessing(ctx context.Context, params aiRequestParams, res interface{
 	if err != nil {
 		return fmt.Errorf("invalid events URL: %w", err)
 	}
-	data, err := common.AppendHostname(*resp.JSON200.DataUrl, host)
-	if err != nil {
-		return fmt.Errorf("invalid data URL: %w", err)
-	}
+
 	if resp.JSON200.ManifestId != nil {
 		ctx = clog.AddVal(ctx, "manifest_id", *resp.JSON200.ManifestId)
 		params.liveParams.manifestID = *resp.JSON200.ManifestId
 	}
-	clog.V(common.VERBOSE).Infof(ctx, "pub %s sub %s control %s events %s data %s", pub, sub, control, events, data)
+
+	clog.V(common.VERBOSE).Infof(ctx, "pub %s sub %s control %s events %s", pub, sub, control, events)
 
 	startControlPublish(ctx, control, params)
 	startTricklePublish(ctx, pub, params, params.liveParams.sess)
 	startTrickleSubscribe(ctx, sub, params, params.liveParams.sess)
 	startEventsSubscribe(ctx, events, params, params.liveParams.sess)
-	startDataSubscribe(ctx, data, params, params.liveParams.sess)
+
+	//optional channels
+	var data *url.URL
+	if *resp.JSON200.DataUrl != "" {
+		data, err = common.AppendHostname(*resp.JSON200.DataUrl, host)
+		if err != nil {
+			return fmt.Errorf("invalid data URL: %w", err)
+		}
+		clog.V(common.VERBOSE).Infof(ctx, "data %s", data)
+		startDataSubscribe(ctx, data, params, params.liveParams.sess)
+	}
+
 	return nil
 }
 
@@ -1090,7 +1109,6 @@ func (ls *LivepeerServer) CreateWhip(server *media.WHIPServer) http.Handler {
 
 				liveParams: &liveRequestParams{
 					segmentReader:          ssr,
-					dataWriter:             media.NewSegmentWriter(5),
 					rtmpOutputs:            rtmpOutputs,
 					localRTMPPrefix:        internalOutputHost,
 					stream:                 streamName,
@@ -1106,6 +1124,15 @@ func (ls *LivepeerServer) CreateWhip(server *media.WHIPServer) http.Handler {
 				},
 			}
 
+			//create a dataWriter for data channel if enabled
+			if enableData, ok := pipelineParams["enableData"]; ok {
+				if enableData == true || enableData == "true" {
+					params.liveParams.dataWriter = media.NewSegmentWriter(5)
+					pipelineParams["enableData"] = true
+					clog.Infof(ctx, "Data channel enabled for stream %s", streamName)
+				}
+			}
+
 			registerControl(ctx, params)
 
 			req := worker.GenLiveVideoToVideoJSONRequestBody{