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"
 )
@@ -47,29 +46,16 @@ type ExternalCapability struct {
 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()
 
 	//Orchestrator fields
 	Sender         ethcommon.Address
+	StreamRequest  []byte
 	pubChannel     *trickle.TrickleLocalPublisher
 	subChannel     *trickle.TrickleLocalPublisher
 	controlChannel *trickle.TrickleLocalPublisher
 	eventsChannel  *trickle.TrickleLocalPublisher
 	dataChannel    *trickle.TrickleLocalPublisher
 	//Stream fields
-	Params       interface{}
-	DataWriter   *media.SegmentWriter
 	JobParams    string
 	StreamCtx    context.Context
 	CancelStream context.CancelFunc
@@ -82,19 +68,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 +103,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 +115,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 +126,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

@@ -3,7 +3,6 @@ package server
 import (
 	"bytes"
 	"context"
-	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -1042,7 +1041,7 @@ func (h *lphttp) StartStream(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("X-Data-Url", dataUrl)
 	}
 
-	reqBodyForRunner["request"] = base64.StdEncoding.EncodeToString(body)
+	reqBodyForRunner["request"] = string(body)
 	reqBodyBytes, err := json.Marshal(reqBodyForRunner)
 	if err != nil {
 		clog.Errorf(ctx, "Failed to marshal request body err=%v", err)
@@ -1089,7 +1088,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, reqBodyBytes)
 	if err != nil {
 		clog.Errorf(ctx, "Error adding stream to external capabilities: %v", err)
 		respondWithError(w, "Error adding stream to external capabilities", http.StatusInternalServerError)