move LivePipeline StreamCtx to not exported and add function to get Context

Author: ad-astra-video

core/livepeernode.go

@@ -191,7 +191,7 @@ type LivePipeline struct {
 
 	DataWriter *media.SegmentWriter
 
-	StreamCtx     context.Context
+	streamCtx     context.Context
 	streamCancel  context.CancelCauseFunc
 	streamParams  interface{}
 	streamRequest []byte
@@ -202,7 +202,8 @@ func (n *LivepeerNode) NewLivePipeline(requestID, streamID, pipeline string, str
 	n.LiveMu.Lock()
 	defer n.LiveMu.Unlock()
 
-	//ensure streamRequest is not nil or empty
+	//ensure streamRequest is not nil or empty to avoid json unmarshal issues on Orchestrator failover
+	//sends the request bytes to next Orchestrator
 	if streamRequest == nil || len(streamRequest) == 0 {
 		streamRequest = []byte("{}")
 	}
@@ -211,7 +212,7 @@ func (n *LivepeerNode) NewLivePipeline(requestID, streamID, pipeline string, str
 		RequestID:     requestID,
 		StreamID:      streamID,
 		Pipeline:      pipeline,
-		StreamCtx:     streamCtx,
+		streamCtx:     streamCtx,
 		streamParams:  streamParams,
 		streamCancel:  streamCancel,
 		streamRequest: streamRequest,
@@ -226,6 +227,10 @@ func (n *LivepeerNode) RemoveLivePipeline(streamID string) {
 	delete(n.LivePipelines, streamID)
 }
 
+func (n *LivePipeline) GetContext() context.Context {
+	return n.streamCtx
+}
+
 func (p *LivePipeline) StreamParams() interface{} {
 	return p.streamParams
 }

server/job_stream.go

@@ -152,7 +152,7 @@ func (ls *LivepeerServer) runStream(gatewayJob *gatewayJob) {
 		return
 	}
 	//this context passes to all channels that will close when stream is canceled
-	ctx := stream.StreamCtx
+	ctx := stream.GetContext()
 	ctx = clog.AddVal(ctx, "stream_id", streamID)
 
 	params, err := getStreamRequestParams(stream)
@@ -290,10 +290,10 @@ func (ls *LivepeerServer) monitorStream(streamId string) {
 
 	//ensure live pipeline is cleaned up if monitoring ends
 	defer ls.LivepeerNode.RemoveLivePipeline(streamId)
-
+	streamCtx := stream.GetContext()
 	for {
 		select {
-		case <-stream.StreamCtx.Done():
+		case <-streamCtx.Done():
 			clog.Infof(ctx, "Stream %s stopped, ending monitoring", streamId)
 			return
 		case <-pmtTicker.C: