BYOC: add streaming

common/testutil.go

@@ -93,11 +93,23 @@ func IgnoreRoutines() []goleak.Option {
 		"github.com/livepeer/go-livepeer/core.(*Balances).StartCleanup",
 		"internal/synctest.Run",
 		"testing/synctest.testingSynctestTest",
+		"github.com/livepeer/go-livepeer/core.(*Balances).StartCleanup", "github.com/livepeer/go-livepeer/server.startTrickleSubscribe.func2", "github.com/livepeer/go-livepeer/server.startTrickleSubscribe",
+		"net/http.(*persistConn).writeLoop", "net/http.(*persistConn).readLoop", "io.(*pipe).read",
+		"github.com/livepeer/go-livepeer/media.gatherIncomingTracks",
 	}
 
-	res := make([]goleak.Option, 0, len(funcs2ignore))
+	// Functions that might have other functions on top of their stack (like time.Sleep)
+	// These need to be ignored with IgnoreAnyFunction instead of IgnoreTopFunction
+	funcsAnyIgnore := []string{
+		"github.com/livepeer/go-livepeer/server.ffmpegOutput",
+	}
+
+	res := make([]goleak.Option, 0, len(funcs2ignore)+len(funcsAnyIgnore))
 	for _, f := range funcs2ignore {
 		res = append(res, goleak.IgnoreTopFunction(f))
 	}
+	for _, f := range funcsAnyIgnore {
+		res = append(res, goleak.IgnoreAnyFunction(f))
+	}
 	return res
 }

core/ai_orchestrator.go

@@ -1200,6 +1200,12 @@ func (orch *orchestrator) JobPriceInfo(sender ethcommon.Address, jobCapability s
 		return nil, err
 	}
 
+	//ensure price numerator and denominator can be int64
+	jobPrice, err = common.PriceToInt64(jobPrice)
+	if err != nil {
+		return nil, fmt.Errorf("invalid job price: %w", err)
+	}
+
 	return &net.PriceInfo{
 		PricePerUnit:  jobPrice.Num().Int64(),
 		PixelsPerUnit: jobPrice.Denom().Int64(),

core/external_capabilities.go

@@ -1,15 +1,33 @@
 package core
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"math/big"
 
 	"sync"
 
+	ethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/golang/glog"
+	"github.com/livepeer/go-livepeer/net"
+	"github.com/livepeer/go-livepeer/trickle"
 )
 
+type JobToken struct {
+	SenderAddress *JobSender        `json:"sender_address,omitempty"`
+	TicketParams  *net.TicketParams `json:"ticket_params,omitempty"`
+	Balance       int64             `json:"balance,omitempty"`
+	Price         *net.PriceInfo    `json:"price,omitempty"`
+	ServiceAddr   string            `json:"service_addr,omitempty"`
+
+	LastNonce uint32
+}
+type JobSender struct {
+	Addr string `json:"addr"`
+	Sig  string `json:"sig"`
+}
+
 type ExternalCapability struct {
 	Name          string `json:"name"`
 	Description   string `json:"description"`
@@ -25,13 +43,144 @@ type ExternalCapability struct {
 	Load int
 }
 
+type StreamInfo struct {
+	StreamID   string
+	Capability string
+
+	//Orchestrator fields
+	Sender         ethcommon.Address
+	StreamRequest  []byte
+	pubChannel     *trickle.TrickleLocalPublisher
+	subChannel     *trickle.TrickleLocalPublisher
+	controlChannel *trickle.TrickleLocalPublisher
+	eventsChannel  *trickle.TrickleLocalPublisher
+	dataChannel    *trickle.TrickleLocalPublisher
+	//Stream fields
+	JobParams    string
+	StreamCtx    context.Context
+	CancelStream context.CancelFunc
+
+	sdm sync.Mutex
+}
+
+func (sd *StreamInfo) IsActive() bool {
+	sd.sdm.Lock()
+	defer sd.sdm.Unlock()
+	if sd.StreamCtx.Err() != nil {
+		return false
+	}
+
+	if sd.controlChannel == nil {
+		return false
+	}
+
+	return true
+}
+
+func (sd *StreamInfo) UpdateParams(params string) {
+	sd.sdm.Lock()
+	defer sd.sdm.Unlock()
+	sd.JobParams = params
+}
+
+func (sd *StreamInfo) SetChannels(pub, sub, control, events, data *trickle.TrickleLocalPublisher) {
+	sd.sdm.Lock()
+	defer sd.sdm.Unlock()
+	sd.pubChannel = pub
+	sd.subChannel = sub
+	sd.controlChannel = control
+	sd.eventsChannel = events
+	sd.dataChannel = data
+}
+
 type ExternalCapabilities struct {
 	capm         sync.Mutex
 	Capabilities map[string]*ExternalCapability
+	Streams      map[string]*StreamInfo
 }
 
 func NewExternalCapabilities() *ExternalCapabilities {
-	return &ExternalCapabilities{Capabilities: make(map[string]*ExternalCapability)}
+	return &ExternalCapabilities{Capabilities: make(map[string]*ExternalCapability),
+		Streams: make(map[string]*StreamInfo),
+	}
+}
+
+func (extCaps *ExternalCapabilities) AddStream(streamID string, capability string, streamReq []byte) (*StreamInfo, error) {
+	extCaps.capm.Lock()
+	defer extCaps.capm.Unlock()
+	_, ok := extCaps.Streams[streamID]
+	if ok {
+		return nil, fmt.Errorf("stream already exists: %s", streamID)
+	}
+
+	//add to streams
+	ctx, cancel := context.WithCancel(context.Background())
+	stream := StreamInfo{
+		StreamID:      streamID,
+		Capability:    capability,
+		StreamRequest: streamReq,
+		StreamCtx:     ctx,
+		CancelStream:  cancel,
+	}
+	extCaps.Streams[streamID] = &stream
+
+	//clean up when stream ends
+	go func() {
+		<-ctx.Done()
+
+		//orchestrator channels shutdown
+		if stream.pubChannel != nil {
+			if err := stream.pubChannel.Close(); err != nil {
+				glog.Errorf("error closing pubChannel for stream=%s: %v", streamID, err)
+			}
+		}
+		if stream.subChannel != nil {
+			if err := stream.subChannel.Close(); err != nil {
+				glog.Errorf("error closing subChannel for stream=%s: %v", streamID, err)
+			}
+		}
+		if stream.controlChannel != nil {
+			if err := stream.controlChannel.Close(); err != nil {
+				glog.Errorf("error closing controlChannel for stream=%s: %v", streamID, err)
+			}
+		}
+		if stream.eventsChannel != nil {
+			if err := stream.eventsChannel.Close(); err != nil {
+				glog.Errorf("error closing eventsChannel for stream=%s: %v", streamID, err)
+			}
+		}
+		if stream.dataChannel != nil {
+			if err := stream.dataChannel.Close(); err != nil {
+				glog.Errorf("error closing dataChannel for stream=%s: %v", streamID, err)
+			}
+		}
+		return
+	}()
+
+	return &stream, nil
+}
+
+func (extCaps *ExternalCapabilities) RemoveStream(streamID string) {
+	extCaps.capm.Lock()
+	defer extCaps.capm.Unlock()
+
+	streamInfo, ok := extCaps.Streams[streamID]
+	if ok {
+		//confirm stream context is canceled before deleting
+		if streamInfo.StreamCtx.Err() == nil {
+			streamInfo.CancelStream()
+		}
+	}
+
+	delete(extCaps.Streams, streamID)
+}
+
+func (extCaps *ExternalCapabilities) StreamExists(streamID string) bool {
+	extCaps.capm.Lock()
+	defer extCaps.capm.Unlock()
+
+	_, ok := extCaps.Streams[streamID]
+	return ok
 }
 
 func (extCaps *ExternalCapabilities) RemoveCapability(extCap string) {

core/livepeernode.go

@@ -10,6 +10,7 @@ orchestrator.go: Code that is called only when the node is in orchestrator mode.
 package core
 
 import (
+	"context"
 	"errors"
 	"math/big"
 	"math/rand"
@@ -19,6 +20,7 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	"github.com/livepeer/go-livepeer/media"
 	"github.com/livepeer/go-livepeer/pm"
 	"github.com/livepeer/go-livepeer/trickle"
 
@@ -183,6 +185,54 @@ type LivePipeline struct {
 	ControlPub   *trickle.TricklePublisher
 	StopControl  func()
 	ReportUpdate func([]byte)
+	DataWriter   *media.SegmentWriter
+
+	StreamCtx     context.Context
+	streamCancel  context.CancelCauseFunc
+	streamParams  interface{}
+	streamRequest []byte
+}
+
+func (n *LivepeerNode) NewLivePipeline(requestID, streamID, pipeline string, streamParams interface{}, streamRequest []byte) *LivePipeline {
+	streamCtx, streamCancel := context.WithCancelCause(context.Background())
+	n.LiveMu.Lock()
+	defer n.LiveMu.Unlock()
+	n.LivePipelines[streamID] = &LivePipeline{
+		RequestID:     requestID,
+		StreamID:      streamID,
+		Pipeline:      pipeline,
+		StreamCtx:     streamCtx,
+		streamParams:  streamParams,
+		streamCancel:  streamCancel,
+		streamRequest: streamRequest,
+	}
+	return n.LivePipelines[streamID]
+}
+
+func (n *LivepeerNode) RemoveLivePipeline(streamID string) {
+	n.LiveMu.Lock()
+	defer n.LiveMu.Unlock()
+	delete(n.LivePipelines, streamID)
+}
+
+func (p *LivePipeline) StreamParams() interface{} {
+	return p.streamParams
+}
+
+func (p *LivePipeline) UpdateStreamParams(newParams interface{}) {
+	p.streamParams = newParams
+}
+
+func (p *LivePipeline) StreamRequest() []byte {
+	return p.streamRequest
+}
+
+func (p *LivePipeline) StopStream(err error) {
+	if p.StopControl != nil {
+		p.StopControl()
+	}
+
+	p.streamCancel(err)
 }
 
 // NewLivepeerNode creates a new Livepeer Node. Eth can be nil.