slim down StreamInfo in external capabilities used by Orchestrator

Author: ad-astra-video

core/external_capabilities.go

@@ -10,7 +10,6 @@ import (
 
 	ethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/golang/glog"
-	"github.com/livepeer/go-livepeer/media"
 	"github.com/livepeer/go-livepeer/net"
 	"github.com/livepeer/go-livepeer/trickle"
 )
@@ -48,17 +47,7 @@ type StreamInfo struct {
 	StreamID   string
 	Capability string
 	//Gateway fields
-	StreamRequest    []byte
-	ExcludeOrchs     []string
-	OrchToken        *JobToken
-	OrchUrl          string
-	OrchPublishUrl   string
-	OrchSubscribeUrl string
-	OrchControlUrl   string
-	OrchEventsUrl    string
-	OrchDataUrl      string
-	ControlPub       *trickle.TricklePublisher
-	StopControl      func()
+	StreamRequest []byte
 
 	//Orchestrator fields
 	Sender         ethcommon.Address
@@ -68,8 +57,6 @@ type StreamInfo struct {
 	eventsChannel  *trickle.TrickleLocalPublisher
 	dataChannel    *trickle.TrickleLocalPublisher
 	//Stream fields
-	Params       interface{}
-	DataWriter   *media.SegmentWriter
 	JobParams    string
 	StreamCtx    context.Context
 	CancelStream context.CancelFunc
@@ -82,19 +69,13 @@ func (sd *StreamInfo) IsActive() bool {
 		return false
 	}
 
-	if sd.controlChannel == nil && sd.ControlPub == nil {
+	if sd.controlChannel == nil {
 		return false
 	}
 
 	return true
 }
 
-func (sd *StreamInfo) ExcludeOrch(orchUrl string) {
-	sd.sdm.Lock()
-	defer sd.sdm.Unlock()
-	sd.ExcludeOrchs = append(sd.ExcludeOrchs, orchUrl)
-}
-
 func (sd *StreamInfo) UpdateParams(params string) {
 	sd.sdm.Lock()
 	defer sd.sdm.Unlock()
@@ -123,7 +104,7 @@ func NewExternalCapabilities() *ExternalCapabilities {
 	}
 }
 
-func (extCaps *ExternalCapabilities) AddStream(streamID string, pipeline string, params interface{}, streamReq []byte) (*StreamInfo, error) {
+func (extCaps *ExternalCapabilities) AddStream(streamID string, capability string, streamReq []byte) (*StreamInfo, error) {
 	extCaps.capm.Lock()
 	defer extCaps.capm.Unlock()
 	_, ok := extCaps.Streams[streamID]
@@ -135,8 +116,7 @@ func (extCaps *ExternalCapabilities) AddStream(streamID string, pipeline string,
 	ctx, cancel := context.WithCancel(context.Background())
 	stream := StreamInfo{
 		StreamID:      streamID,
-		Capability:    pipeline,
-		Params:        params, // Store the interface value directly, not a pointer to it
+		Capability:    capability,
 		StreamRequest: streamReq,
 		StreamCtx:     ctx,
 		CancelStream:  cancel,
@@ -147,15 +127,6 @@ func (extCaps *ExternalCapabilities) AddStream(streamID string, pipeline string,
 	go func() {
 		<-ctx.Done()
 
-		//gateway channels shutdown
-		if stream.DataWriter != nil {
-			stream.DataWriter.Close()
-		}
-		if stream.ControlPub != nil {
-			stream.StopControl()
-			stream.ControlPub.Close()
-		}
-
 		//orchestrator channels shutdown
 		if stream.pubChannel != nil {
 			stream.pubChannel.Close()

server/job_stream.go

@@ -1089,7 +1089,7 @@ func (h *lphttp) StartStream(w http.ResponseWriter, r *http.Request) {
 	clog.V(common.SHORT).Infof(ctx, "stream start processed successfully took=%v balance=%v", time.Since(start), getPaymentBalance(orch, orchJob.Sender, orchJob.Req.Capability).FloatString(0))
 
 	//setup the stream
-	stream, err := h.node.ExternalCapabilities.AddStream(orchJob.Req.ID, orchJob.Req.Capability, orchJob.Req, respBody)
+	stream, err := h.node.ExternalCapabilities.AddStream(orchJob.Req.ID, orchJob.Req.Capability, respBody)
 	if err != nil {
 		clog.Errorf(ctx, "Error adding stream to external capabilities: %v", err)
 		respondWithError(w, "Error adding stream to external capabilities", http.StatusInternalServerError)