Merge pull request #185 from livepeer/eli/mist-stream-source

mapic: allow customization of mist stream source
livepeer · Jul 23, 2022 · 0959ca3 · 0959ca3
2 parents 2ce44d6 + 9a4e5d1
commit 0959ca3
Show file tree

Hide file tree

Showing 2 changed files with 63 additions and 58 deletions.
diff --git a/cmd/mist-api-connector/mist-api-connector.go b/cmd/mist-api-connector/mist-api-connector.go
@@ -11,11 +11,11 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	"github.com/livepeer/livepeer-data/pkg/mistconnector"
 	"github.com/livepeer/stream-tester/apis/livepeer"
 	mistapi "github.com/livepeer/stream-tester/apis/mist"
 	"github.com/livepeer/stream-tester/internal/app/mistapiconnector"
 	"github.com/livepeer/stream-tester/internal/metrics"
-	"github.com/livepeer/livepeer-data/pkg/mistconnector"
 	"github.com/livepeer/stream-tester/internal/utils"
 	"github.com/livepeer/stream-tester/model"
 	"github.com/peterbourgon/ff"
@@ -39,6 +39,7 @@ func main() {
 	mistPort := fs.Uint("mist-port", 4242, "Port of the Mist server")
 	mistCreds := fs.String("mist-creds", "", "login:password of the Mist server")
 	mistConnectTimeout := fs.Duration("mist-connect-timeout", 5*time.Minute, "Max time to wait attempting to connect to Mist server")
+	mistStreamSource := fs.String("mist-stream-source", "push://", "Stream source we should use for created Mist stream")
 	sendAudio := fs.String("send-audio", "record", "when should we send audio?  {always|never|record}")
 	apiToken := fs.String("api-token", "", "Token of the Livepeer API to be used by the Mist server")
 	apiServer := fs.String("api-server", livepeer.ACServer, "Livepeer API server to use")
@@ -101,23 +102,24 @@ func main() {
 	}
 
 	opts := mistapiconnector.MacOptions{
-		NodeID:         hostName,
-		MistHost:       *mistHost,
-		MistAPI:        mapi,
-		LivepeerAPI:    lapi,
-		BalancerHost:   *balancerHost,
-		RoutePrefix:    *routePrefix,
-		PlaybackDomain: *playbackDomain,
-		MistURL:        *mistURL,
-		BaseStreamName: *baseStreamName,
-		CheckBandwidth: false,
-		SendAudio:      *sendAudio,
-		EtcdEndpoints:  etcdEndpoints,
-		EtcdCaCert:     *etcdCaCert,
-		EtcdCert:       *etcdCert,
-		EtcdKey:        *etcdKey,
-		AMQPUrl:        *amqpUrl,
-		OwnRegion:      *ownRegion,
+		NodeID:           hostName,
+		MistHost:         *mistHost,
+		MistAPI:          mapi,
+		LivepeerAPI:      lapi,
+		BalancerHost:     *balancerHost,
+		RoutePrefix:      *routePrefix,
+		PlaybackDomain:   *playbackDomain,
+		MistURL:          *mistURL,
+		BaseStreamName:   *baseStreamName,
+		CheckBandwidth:   false,
+		SendAudio:        *sendAudio,
+		EtcdEndpoints:    etcdEndpoints,
+		EtcdCaCert:       *etcdCaCert,
+		EtcdCert:         *etcdCert,
+		EtcdKey:          *etcdKey,
+		AMQPUrl:          *amqpUrl,
+		OwnRegion:        *ownRegion,
+		MistStreamSource: *mistStreamSource,
 	}
 	mc, err := mistapiconnector.NewMac(opts)
 	if err != nil {

diff --git a/internal/app/mistapiconnector/mistapiconnector_app.go b/internal/app/mistapiconnector/mistapiconnector_app.go
@@ -122,34 +122,36 @@ type (
 		EtcdEndpoints                 []string
 		EtcdCaCert, EtcdCert, EtcdKey string
 		AMQPUrl, OwnRegion            string
+		MistStreamSource              string
 	}
 
 	trackList map[string]*trackListDesc
 
 	mac struct {
-		ctx            context.Context
-		cancel         context.CancelFunc
-		mapi           *mist.API
-		lapi           *livepeer.API
-		balancerHost   string
-		srv            *http.Server
-		srvShutCh      chan error
-		mu             sync.RWMutex
-		mistHot        string
-		checkBandwidth bool
-		routePrefix    string
-		mistURL        string
-		playbackDomain string
-		sendAudio      string
-		baseStreamName string
-		useEtcd        bool
-		etcdClient     *clientv3.Client
-		etcdSession    *concurrency.Session
-		etcdPub2rev    map[string]etcdRevData // public key to revision of etcd keys
-		pub2info       map[string]*streamInfo // public key to info
-		producer       *event.AMQPProducer
-		nodeID         string
-		ownRegion      string
+		ctx              context.Context
+		cancel           context.CancelFunc
+		mapi             *mist.API
+		lapi             *livepeer.API
+		balancerHost     string
+		srv              *http.Server
+		srvShutCh        chan error
+		mu               sync.RWMutex
+		mistHot          string
+		checkBandwidth   bool
+		routePrefix      string
+		mistURL          string
+		playbackDomain   string
+		sendAudio        string
+		baseStreamName   string
+		useEtcd          bool
+		etcdClient       *clientv3.Client
+		etcdSession      *concurrency.Session
+		etcdPub2rev      map[string]etcdRevData // public key to revision of etcd keys
+		pub2info         map[string]*streamInfo // public key to info
+		producer         *event.AMQPProducer
+		nodeID           string
+		ownRegion        string
+		mistStreamSource string
 		// pub2id         map[string]string // public key to stream id
 	}
 )
@@ -239,21 +241,22 @@ func NewMac(opts MacOptions) (IMac, error) {
 		checkBandwidth: opts.CheckBandwidth,
 		balancerHost:   opts.BalancerHost,
 		// pub2id:         make(map[string]string), // public key to stream id
-		pub2info:       make(map[string]*streamInfo), // public key to info
-		routePrefix:    opts.RoutePrefix,
-		mistURL:        opts.MistURL,
-		playbackDomain: opts.PlaybackDomain,
-		sendAudio:      opts.SendAudio,
-		baseStreamName: opts.BaseStreamName,
-		useEtcd:        useEtcd,
-		etcdClient:     cli,
-		etcdSession:    sess,
-		etcdPub2rev:    make(map[string]etcdRevData), // public key to revision of etcd keys
-		srvShutCh:      make(chan error),
-		ctx:            ctx,
-		cancel:         cancel,
-		producer:       producer,
-		ownRegion:      opts.OwnRegion,
+		pub2info:         make(map[string]*streamInfo), // public key to info
+		routePrefix:      opts.RoutePrefix,
+		mistURL:          opts.MistURL,
+		playbackDomain:   opts.PlaybackDomain,
+		sendAudio:        opts.SendAudio,
+		baseStreamName:   opts.BaseStreamName,
+		useEtcd:          useEtcd,
+		etcdClient:       cli,
+		etcdSession:      sess,
+		etcdPub2rev:      make(map[string]etcdRevData), // public key to revision of etcd keys
+		srvShutCh:        make(chan error),
+		ctx:              ctx,
+		cancel:           cancel,
+		producer:         producer,
+		ownRegion:        opts.OwnRegion,
+		mistStreamSource: opts.MistStreamSource,
 	}
 	go mc.recoverSessionLoop()
 	if producer != nil {
@@ -1120,13 +1123,13 @@ func (mc *mac) SetupTriggers(ownURI string) error {
 		apiURL := mc.lapi.GetServer() + "/api/stream/" + mc.baseStreamName
 		presets := []string{"P144p30fps16x9"}
 		// base stream created with audio disabled
-		err = mc.mapi.CreateStream(mc.baseStreamName, presets, nil, "1", apiURL, "", false, false)
+		err = mc.mapi.CreateStream(mc.baseStreamName, presets, nil, "1", apiURL, mc.mistStreamSource, false, false)
 		if err != nil {
 			glog.Error(err)
 			return err
 		}
 		// create second stream with audio enabled - used for stream with recording enabled
-		err = mc.mapi.CreateStream(mc.baseStreamName+audioEnabledStreamSuffix, presets, nil, "1", apiURL, "", false, true)
+		err = mc.mapi.CreateStream(mc.baseStreamName+audioEnabledStreamSuffix, presets, nil, "1", apiURL, mc.mistStreamSource, false, true)
 	}
 	return err
 }