+3 -1

go.mod

··· 8 9 replace github.com/AxisCommunications/go-dpop => github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4 10 0 0 11 tool github.com/bluesky-social/indigo/cmd/lexgen 12 13 require ( ··· 48 github.com/multiformats/go-multihash v0.2.3 49 github.com/orandin/slog-gorm v1.4.0 50 github.com/patrickmn/go-cache v2.1.0+incompatible 51 - github.com/peterbourgon/ff/v3 v3.4.0 52 github.com/pion/interceptor v0.1.37 53 github.com/pion/rtcp v1.2.16 54 github.com/pion/webrtc/v4 v4.0.11 ··· 64 github.com/streamplace/oatproxy v0.0.0-20260130124113-420429019d3b 65 github.com/stretchr/testify v1.11.1 66 github.com/tdewolff/canvas v0.0.0-20250728095813-50d4cb1eee71 0 67 github.com/whyrusleeping/cbor-gen v0.3.1 68 github.com/whyrusleeping/go-did v0.0.0-20230824162731-404d1707d5d6 69 gitlab.com/gitlab-org/release-cli v0.18.0

··· 8 9 replace github.com/AxisCommunications/go-dpop => github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4 10 11 + //replace github.com/livepeer/go-livepeer => ../go-livepeer 12 + 13 tool github.com/bluesky-social/indigo/cmd/lexgen 14 15 require ( ··· 50 github.com/multiformats/go-multihash v0.2.3 51 github.com/orandin/slog-gorm v1.4.0 52 github.com/patrickmn/go-cache v2.1.0+incompatible 0 53 github.com/pion/interceptor v0.1.37 54 github.com/pion/rtcp v1.2.16 55 github.com/pion/webrtc/v4 v4.0.11 ··· 65 github.com/streamplace/oatproxy v0.0.0-20260130124113-420429019d3b 66 github.com/stretchr/testify v1.11.1 67 github.com/tdewolff/canvas v0.0.0-20250728095813-50d4cb1eee71 68 + github.com/urfave/cli/v3 v3.6.2 69 github.com/whyrusleeping/cbor-gen v0.3.1 70 github.com/whyrusleeping/go-did v0.0.0-20230824162731-404d1707d5d6 71 gitlab.com/gitlab-org/release-cli v0.18.0

+2 -4

go.sum

··· 1080 github.com/perimeterx/marshmallow v1.1.5/go.mod h1:dsXbUu8CRzfYP5a87xpp0xq9S3u0Vchtcl8we9tYaXw= 1081 github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9 h1:1/WtZae0yGtPq+TI6+Tv1WTxkukpXeMlviSxvL7SRgk= 1082 github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9/go.mod h1:x3N5drFsm2uilKKuuYo6LdyD8vZAW55sH/9w+pbo1sw= 1083 - github.com/peterbourgon/ff/v3 v3.4.0 h1:QBvM/rizZM1cB0p0lGMdmR7HxZeI/ZrBWB4DqLkMUBc= 1084 - github.com/peterbourgon/ff/v3 v3.4.0/go.mod h1:zjJVUhx+twciwfDl0zBcFzl4dW8axCRyXE/eKY9RztQ= 1085 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 h1:oYW+YCJ1pachXTQmzR3rNLYGGz4g/UgFcjb28p/viDM= 1086 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0= 1087 github.com/pierrec/lz4 v2.6.1+incompatible h1:9UY3+iC23yxF0UfGaYrGplQ+79Rg+h/q9FV9ix19jjM= ··· 1317 github.com/streamplace/atproto-oauth-golang v0.0.0-20250619231223-a9c04fb888ac/go.mod h1:9LlKkqciiO5lRfbX0n4Wn5KNY9nvFb4R3by8FdW2TWc= 1318 github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4 h1:L1fS4HJSaAyNnkwfuZubgfeZy8rkWmA0cMtH5Z0HqNc= 1319 github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4/go.mod h1:bGUXY9Wd4mnd+XUrOYZr358J2f6z9QO/dLhL1SsiD+0= 1320 - github.com/streamplace/oatproxy v0.0.0-20260112011721-d74b4913c93f h1:hhbQ8CtcAZVlLit/r7b9QDK7qEgOth4hgE13xV6ViBI= 1321 - github.com/streamplace/oatproxy v0.0.0-20260112011721-d74b4913c93f/go.mod h1:pXi24hA7xBHj8eEywX6wGqJOR9FaEYlGwQ/72rN6okw= 1322 github.com/streamplace/oatproxy v0.0.0-20260130124113-420429019d3b h1:BB/R1egvkEqZhGeKL3tqAlTn0mkoOaaMY6r6s18XJYA= 1323 github.com/streamplace/oatproxy v0.0.0-20260130124113-420429019d3b/go.mod h1:pXi24hA7xBHj8eEywX6wGqJOR9FaEYlGwQ/72rN6okw= 1324 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= ··· 1389 github.com/urfave/cli v1.22.10/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0= 1390 github.com/urfave/cli/v2 v2.27.7 h1:bH59vdhbjLv3LAvIu6gd0usJHgoTTPhCFib8qqOwXYU= 1391 github.com/urfave/cli/v2 v2.27.7/go.mod h1:CyNAG/xg+iAOg0N4MPGZqVmv2rCoP267496AOXUZjA4= 0 0 1392 github.com/uudashr/gocognit v1.2.0 h1:3BU9aMr1xbhPlvJLSydKwdLN3tEUUrzPSSM8S4hDYRA= 1393 github.com/uudashr/gocognit v1.2.0/go.mod h1:k/DdKPI6XBZO1q7HgoV2juESI2/Ofj9AcHPZhBBdrTU= 1394 github.com/uudashr/iface v1.3.1 h1:bA51vmVx1UIhiIsQFSNq6GZ6VPTk3WNMZgRiCe9R29U=

··· 1080 github.com/perimeterx/marshmallow v1.1.5/go.mod h1:dsXbUu8CRzfYP5a87xpp0xq9S3u0Vchtcl8we9tYaXw= 1081 github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9 h1:1/WtZae0yGtPq+TI6+Tv1WTxkukpXeMlviSxvL7SRgk= 1082 github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9/go.mod h1:x3N5drFsm2uilKKuuYo6LdyD8vZAW55sH/9w+pbo1sw= 0 0 1083 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 h1:oYW+YCJ1pachXTQmzR3rNLYGGz4g/UgFcjb28p/viDM= 1084 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0= 1085 github.com/pierrec/lz4 v2.6.1+incompatible h1:9UY3+iC23yxF0UfGaYrGplQ+79Rg+h/q9FV9ix19jjM= ··· 1315 github.com/streamplace/atproto-oauth-golang v0.0.0-20250619231223-a9c04fb888ac/go.mod h1:9LlKkqciiO5lRfbX0n4Wn5KNY9nvFb4R3by8FdW2TWc= 1316 github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4 h1:L1fS4HJSaAyNnkwfuZubgfeZy8rkWmA0cMtH5Z0HqNc= 1317 github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4/go.mod h1:bGUXY9Wd4mnd+XUrOYZr358J2f6z9QO/dLhL1SsiD+0= 0 0 1318 github.com/streamplace/oatproxy v0.0.0-20260130124113-420429019d3b h1:BB/R1egvkEqZhGeKL3tqAlTn0mkoOaaMY6r6s18XJYA= 1319 github.com/streamplace/oatproxy v0.0.0-20260130124113-420429019d3b/go.mod h1:pXi24hA7xBHj8eEywX6wGqJOR9FaEYlGwQ/72rN6okw= 1320 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= ··· 1385 github.com/urfave/cli v1.22.10/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0= 1386 github.com/urfave/cli/v2 v2.27.7 h1:bH59vdhbjLv3LAvIu6gd0usJHgoTTPhCFib8qqOwXYU= 1387 github.com/urfave/cli/v2 v2.27.7/go.mod h1:CyNAG/xg+iAOg0N4MPGZqVmv2rCoP267496AOXUZjA4= 1388 + github.com/urfave/cli/v3 v3.6.2 h1:lQuqiPrZ1cIz8hz+HcrG0TNZFxU70dPZ3Yl+pSrH9A8= 1389 + github.com/urfave/cli/v3 v3.6.2/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso= 1390 github.com/uudashr/gocognit v1.2.0 h1:3BU9aMr1xbhPlvJLSydKwdLN3tEUUrzPSSM8S4hDYRA= 1391 github.com/uudashr/gocognit v1.2.0/go.mod h1:k/DdKPI6XBZO1q7HgoV2juESI2/Ofj9AcHPZhBBdrTU= 1392 github.com/uudashr/iface v1.3.1 h1:bA51vmVx1UIhiIsQFSNq6GZ6VPTk3WNMZgRiCe9R29U=

+2 -1

js/docs/package.json

··· 23 }, 24 "devDependencies": { 25 "starlight-sidebar-topics": "^0.6.2" 26 - } 0 27 }

··· 23 }, 24 "devDependencies": { 25 "starlight-sidebar-topics": "^0.6.2" 26 + }, 27 + "private": true 28 }

+1 -1

js/docs/src/content/docs/guides/start-contributing/streamplace-dev-setup.md

··· 57 frontend, you can override the pertinent command line argument: 58 59 ```shell 60 - make dev && ./build-darwin-arm64/streamplace --dev-frontend-proxy="" 61 ``` 62 63 If you're using a proxy server, you may want to set your tunnel URL as the

··· 57 frontend, you can override the pertinent command line argument: 58 59 ```shell 60 + make dev && ./build-darwin-arm64/streamplace --dev-frontend-proxy=false 61 ``` 62 63 If you're using a proxy server, you may want to set your tunnel URL as the

+6 -15

pkg/cmd/combine.go

··· 14 "stream.place/streamplace/pkg/media" 15 ) 16 17 - func Combine(ctx context.Context, build *config.BuildFlags, allArgs []string) error { 18 gstinit.InitGST() 19 - cli := &config.CLI{Build: build} 20 - fs := cli.NewFlagSet("streamplace combine") 21 - debugDir := fs.String("debug-dir", "", "directory to write debug files to") 22 23 - err := cli.Parse(fs, allArgs) 24 - if err != nil { 25 - return err 26 - } 27 - if *debugDir != "" { 28 - err := os.MkdirAll(*debugDir, 0755) 29 if err != nil { 30 return fmt.Errorf("failed to create debug directory: %w", err) 31 } 32 } 33 - log.Debug(context.Background(), "combine command: starting", "args", fs.Args()) 34 ctx = log.WithDebugValue(ctx, cli.Debug) 35 cryptoSigner, err := createSigner(ctx, cli) 36 if err != nil { ··· 40 if err != nil { 41 return err 42 } 43 - args := fs.Args() 44 - outFile := args[0] 45 - inputs := args[1:] 46 log.Log(ctx, "combining segments", "outFile", outFile, "inputs", inputs) 47 outFd, err := os.Create(outFile) 48 if err != nil { ··· 62 if err != nil { 63 return err 64 } 65 - err = CheckCombined(ctx, cli, outFd, *debugDir) 66 if err != nil { 67 return err 68 }

··· 14 "stream.place/streamplace/pkg/media" 15 ) 16 17 + func Combine(ctx context.Context, cli *config.CLI, debugDir string, outFile string, inputs []string) error { 18 gstinit.InitGST() 0 0 0 19 20 + if debugDir != "" { 21 + err := os.MkdirAll(debugDir, 0755) 0 0 0 0 22 if err != nil { 23 return fmt.Errorf("failed to create debug directory: %w", err) 24 } 25 } 26 + log.Debug(context.Background(), "combine command: starting", "outFile", outFile, "inputs", inputs) 27 ctx = log.WithDebugValue(ctx, cli.Debug) 28 cryptoSigner, err := createSigner(ctx, cli) 29 if err != nil { ··· 33 if err != nil { 34 return err 35 } 36 + 0 0 37 log.Log(ctx, "combining segments", "outFile", outFile, "inputs", inputs) 38 outFd, err := os.Create(outFile) 39 if err != nil { ··· 53 if err != nil { 54 return err 55 } 56 + err = CheckCombined(ctx, cli, outFd, debugDir) 57 if err != nil { 58 return err 59 }

+2 -31

pkg/cmd/go_livepeer.go

··· 3 import ( 4 "context" 5 "flag" 6 - "strings" 7 8 - "github.com/golang/glog" 9 "github.com/livepeer/go-livepeer/cmd/livepeer/starter" 10 "stream.place/streamplace/pkg/config" 11 ) 12 13 func GoLivepeer(ctx context.Context, fs *flag.FlagSet) error { 14 - lpfs := flag.NewFlagSet("livepeer", flag.ExitOnError) 15 - cfg := starter.NewLivepeerConfig(lpfs) 16 - fs.VisitAll(func(f *flag.Flag) { 17 - if !strings.HasPrefix(f.Name, "livepeer.") { 18 - return 19 - } 20 - name := strings.TrimPrefix(f.Name, "livepeer.") 21 - adapted := config.LivepeerFlags.SnakeToCamel[name] 22 - 23 - if adapted == "" { 24 - panic("unknown livepeer flag: " + name) 25 - } 26 - err := lpfs.Set(adapted, f.Value.String()) 27 - if err != nil { 28 - panic(err) 29 - } 30 - }) 31 - 32 err := flag.Set("logtostderr", "true") 33 if err != nil { 34 return err ··· 39 return err 40 } 41 42 - // Config file 43 - // err = ff.Parse(fs, args, 44 - // ff.WithConfigFileFlag("config"), 45 - // ff.WithEnvVarPrefix("SP_LIVEPEER"), 46 - // ) 47 - if err != nil { 48 - glog.Exit("Error parsing config: ", err) 49 - } 50 - 51 - cfg = starter.UpdateNilsForUnsetFlags(cfg) 52 53 - starter.StartLivepeer(ctx, cfg) 54 55 return nil 56 }

··· 3 import ( 4 "context" 5 "flag" 0 6 0 7 "github.com/livepeer/go-livepeer/cmd/livepeer/starter" 8 "stream.place/streamplace/pkg/config" 9 ) 10 11 func GoLivepeer(ctx context.Context, fs *flag.FlagSet) error { 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 12 err := flag.Set("logtostderr", "true") 13 if err != nil { 14 return err ··· 19 return err 20 } 21 22 + config.LivepeerConfig = starter.UpdateNilsForUnsetFlags(config.LivepeerConfig) 0 0 0 0 0 0 0 0 0 23 24 + starter.StartLivepeer(ctx, config.LivepeerConfig) 25 26 return nil 27 }

+13 -25

pkg/cmd/sign.go

··· 4 "bytes" 5 "context" 6 "crypto/ecdsa" 7 - "flag" 8 "fmt" 9 "io" 10 "os" ··· 16 "stream.place/streamplace/pkg/media" 17 ) 18 19 - func Sign(ctx context.Context) error { 20 - fs := flag.NewFlagSet("streamplace", flag.ExitOnError) 21 - certPath := fs.String("cert", "", "path to the certificate file") 22 - key := fs.String("key", "", "base58-encoded secp256k1 private key") 23 - streamerName := fs.String("streamer", "", "streamer name") 24 - taURL := fs.String("ta-url", "http://timestamp.digicert.com", "timestamp authority server for signing") 25 - startTime := fs.Int64("start-time", 0, "start time of the stream") 26 - manifestJSON := fs.String("manifest", "", "JSON manifest to use for signing") 27 - if err := fs.Parse(os.Args[2:]); err != nil { 28 - return err 29 - } 30 - 31 log.Debug(ctx, "Sign command: starting", 32 - "streamer", *streamerName, 33 - "startTime", *startTime, 34 - "hasManifest", len(*manifestJSON) > 0) 35 36 - keyBs, err := base58.Decode(*key) 37 if err != nil { 38 return err 39 } 40 41 - if *streamerName == "" { 42 return fmt.Errorf("streamer name is required") 43 } 44 ··· 48 } 49 signer := secpSigner.ToECDSA() 50 51 - certBs, err := os.ReadFile(*certPath) 52 if err != nil { 53 return err 54 } ··· 61 ms := &media.MediaSignerLocal{ 62 Signer: signer, 63 Cert: certBs, 64 - StreamerName: *streamerName, 65 - TAURL: *taURL, 66 AQPub: pub, 67 - PrebuiltManifest: []byte(*manifestJSON), // Pass the manifest from parent process 68 } 69 70 - if len(*manifestJSON) > 0 { 71 - log.Debug(ctx, "Sign command: using provided manifest", "manifestLength", len(*manifestJSON)) 72 } 73 74 inputBs, err := io.ReadAll(os.Stdin) ··· 76 return err 77 } 78 79 - mp4, err := ms.SignMP4(ctx, bytes.NewReader(inputBs), *startTime) 80 if err != nil { 81 return err 82 }

··· 4 "bytes" 5 "context" 6 "crypto/ecdsa" 0 7 "fmt" 8 "io" 9 "os" ··· 15 "stream.place/streamplace/pkg/media" 16 ) 17 18 + func Sign(ctx context.Context, certPath string, key string, streamerName string, taURL string, startTime int64, manifestJSON string) error { 0 0 0 0 0 0 0 0 0 0 0 19 log.Debug(ctx, "Sign command: starting", 20 + "streamer", streamerName, 21 + "startTime", startTime, 22 + "hasManifest", len(manifestJSON) > 0) 23 24 + keyBs, err := base58.Decode(key) 25 if err != nil { 26 return err 27 } 28 29 + if streamerName == "" { 30 return fmt.Errorf("streamer name is required") 31 } 32 ··· 36 } 37 signer := secpSigner.ToECDSA() 38 39 + certBs, err := os.ReadFile(certPath) 40 if err != nil { 41 return err 42 } ··· 49 ms := &media.MediaSignerLocal{ 50 Signer: signer, 51 Cert: certBs, 52 + StreamerName: streamerName, 53 + TAURL: taURL, 54 AQPub: pub, 55 + PrebuiltManifest: []byte(manifestJSON), // Pass the manifest from parent process 56 } 57 58 + if len(manifestJSON) > 0 { 59 + log.Debug(ctx, "Sign command: using provided manifest", "manifestLength", len(manifestJSON)) 60 } 61 62 inputBs, err := io.ReadAll(os.Stdin) ··· 64 return err 65 } 66 67 + mp4, err := ms.SignMP4(ctx, bytes.NewReader(inputBs), startTime) 68 if err != nil { 69 return err 70 }

+319 -142

pkg/cmd/streamplace.go

··· 21 "github.com/bluesky-social/indigo/carstore" 22 "github.com/ethereum/go-ethereum/common/hexutil" 23 "github.com/livepeer/go-livepeer/cmd/livepeer/starter" 24 - "github.com/peterbourgon/ff/v3" 25 "github.com/streamplace/oatproxy/pkg/oatproxy" 0 26 "stream.place/streamplace/pkg/aqhttp" 27 "stream.place/streamplace/pkg/atproto" 28 "stream.place/streamplace/pkg/bus" ··· 54 // parse the CLI and fire up an streamplace node! 55 func start(build *config.BuildFlags, platformJobs []jobFunc) error { 56 iroh_streamplace.InitLogging() 57 - selfTest := len(os.Args) > 1 && os.Args[1] == "self-test" 58 - err := media.RunSelfTest(context.Background()) 59 - if err != nil { 60 - if selfTest { 61 - fmt.Println(err.Error()) 62 - os.Exit(1) 63 - } else { 64 - retryCount, _ := strconv.Atoi(os.Getenv("STREAMPLACE_SELFTEST_RETRY")) 65 - if retryCount >= 3 { 66 - log.Error(context.Background(), "gstreamer self-test failed 3 times, giving up", "error", err) 67 - return err 68 - } 69 - log.Log(context.Background(), "error in gstreamer self-test, attempting recovery", "error", err, "retry", retryCount+1) 70 - os.Setenv("STREAMPLACE_SELFTEST_RETRY", strconv.Itoa(retryCount+1)) 71 - err := syscall.Exec(os.Args[0], os.Args[1:], os.Environ()) 72 - if err != nil { 73 - log.Error(context.Background(), "error in gstreamer self-test, could not restart", "error", err) 74 - return err 75 - } 76 - panic("invalid code path: exec succeeded but we're still here???") 77 - } 78 - } 79 - if selfTest { 80 - runtime.GC() 81 - if err := pprof.Lookup("goroutine").WriteTo(os.Stderr, 2); err != nil { 82 - log.Error(context.Background(), "error creating pprof", "error", err) 83 - } 84 - fmt.Println("self-test successful!") 85 - os.Exit(0) 86 - } 87 88 - if len(os.Args) > 1 && os.Args[1] == "stream" { 89 - if len(os.Args) != 3 { 90 - fmt.Println("usage: streamplace stream [user]") 91 - os.Exit(1) 92 - } 93 - return Stream(os.Args[2]) 0 0 0 0 0 0 0 0 0 94 } 95 - 96 - if len(os.Args) > 1 && os.Args[1] == "live" { 97 - cli := config.CLI{Build: build} 98 - fs := cli.NewFlagSet("streamplace live") 99 - 100 - err := cli.Parse(fs, os.Args[2:]) 0 0 0 0 101 if err != nil { 102 - return err 103 - } 104 - 105 - args := fs.Args() 106 - if len(args) != 1 { 107 - fmt.Println("usage: streamplace live [flags] [stream-key]") 108 - os.Exit(1) 0 0 0 0 0 0 0 0 0 0 0 109 } 110 - 111 - return Live(args[0], cli.HTTPInternalAddr) 112 } 113 - 114 - if len(os.Args) > 1 && os.Args[1] == "sign" { 115 - return Sign(context.Background()) 116 } 117 118 - if len(os.Args) > 1 && os.Args[1] == "whep" { 119 - return WHEP(os.Args[2:]) 120 - } 121 - if len(os.Args) > 1 && os.Args[1] == "whip" { 122 - return WHIP(os.Args[2:]) 123 - } 124 - 125 - if len(os.Args) > 1 && os.Args[1] == "combine" { 126 - return Combine(context.Background(), build, os.Args[2:]) 127 - } 128 129 - if len(os.Args) > 1 && os.Args[1] == "split" { 130 - cli := config.CLI{Build: build} 131 - fs := cli.NewFlagSet("streamplace split") 132 - 133 - err := cli.Parse(fs, os.Args[2:]) 134 - if err != nil { 135 - return err 136 - } 137 - ctx := context.Background() 138 - ctx = log.WithDebugValue(ctx, cli.Debug) 139 - if len(fs.Args()) != 2 { 140 - fmt.Println("usage: streamplace split [flags] [input file] [output directory]") 141 - os.Exit(1) 142 - } 143 - gstinit.InitGST() 144 - return Split(ctx, fs.Args()[0], fs.Args()[1]) 145 - } 146 - 147 - if len(os.Args) > 1 && os.Args[1] == "self-test" { 148 - err := media.RunSelfTest(context.Background()) 149 - if err != nil { 150 - fmt.Println(err.Error()) 151 - os.Exit(1) 152 - } 153 - fmt.Println("self-test successful!") 154 - os.Exit(0) 155 - } 156 - 157 - if len(os.Args) > 1 && os.Args[1] == "livepeer" { 158 - lpfs := flag.NewFlagSet("livepeer", flag.ExitOnError) 159 - _ = starter.NewLivepeerConfig(lpfs) 160 - err = ff.Parse(lpfs, os.Args[2:], 161 - ff.WithConfigFileFlag("config"), 162 - ff.WithEnvVarPrefix("LP"), 163 - ) 164 - if err != nil { 165 - return err 166 - } 167 - err = GoLivepeer(context.Background(), lpfs) 168 - if err != nil { 169 - log.Error(context.Background(), "error in livepeer", "error", err) 170 - os.Exit(1) 171 - } 172 - os.Exit(0) 173 - } 174 - 175 _ = flag.Set("logtostderr", "true") 176 vFlag := flag.Lookup("v") 177 - cli := config.CLI{Build: build} 178 - fs := cli.NewFlagSet("streamplace") 179 - verbosity := fs.String("v", "3", "log verbosity level") 180 - version := fs.Bool("version", false, "print version and exit") 181 182 - err = cli.Parse( 183 - fs, os.Args[1:], 184 - ) 185 if err != nil { 186 return err 187 } ··· 190 if err != nil { 191 return err 192 } 193 - _ = vFlag.Value.Set(*verbosity) 0 194 log.SetColorLogger(cli.Color) 195 - ctx := context.Background() 196 ctx = log.WithDebugValue(ctx, cli.Debug) 197 198 log.Log(ctx, ··· 203 "runtime.GOOS", runtime.GOOS, 204 "runtime.GOARCH", runtime.GOARCH, 205 "runtime.Version", runtime.Version()) 206 - if *version { 207 - return nil 208 - } 209 - signer, err := createSigner(ctx, &cli) 210 if err != nil { 211 return err 212 } 213 214 if len(os.Args) > 1 && os.Args[1] == "migrate" { 215 - return statedb.Migrate(&cli) 216 } 217 218 spmetrics.Version.WithLabelValues(build.Version).Inc() ··· 262 if err != nil { 263 return err 264 } 265 - state, err := statedb.MakeDB(ctx, &cli, noter, mod) 266 if err != nil { 267 return err 268 } 269 - handle, err := atproto.MakeLexiconRepo(ctx, &cli, mod, state) 270 if err != nil { 271 return err 272 } ··· 286 287 b := bus.NewBus() 288 atsync := &atproto.ATProtoSynchronizer{ 289 - CLI: &cli, 290 Model: mod, 291 StatefulDB: state, 292 Noter: noter, ··· 297 return fmt.Errorf("failed to migrate: %w", err) 298 } 299 300 - mm, err := media.MakeMediaManager(ctx, &cli, signer, mod, b, atsync, ldb) 301 if err != nil { 302 return err 303 } 304 305 - ms, err := media.MakeMediaSigner(ctx, &cli, cli.StreamerName, signer, mod) 306 if err != nil { 307 return err 308 } ··· 365 return err 366 } 367 } 368 - replicator, err = iroh_replicator.NewSwarm(ctx, &cli, secret, topic, mm, b, mod) 369 if err != nil { 370 return err 371 } ··· 387 Public: cli.PublicOAuth, 388 HTTPClient: &aqhttp.Client, 389 }) 390 - d := director.NewDirector(mm, mod, &cli, b, op, state, replicator, ldb) 391 - a, err := api.MakeStreamplaceAPI(&cli, mod, state, noter, mm, ms, b, atsync, d, op, ldb) 392 if err != nil { 393 return err 394 } ··· 418 }) 419 if cli.RTMPServerAddon != "" { 420 group.Go(func() error { 421 - return rtmps.ServeRTMPSAddon(ctx, &cli) 422 }) 423 } 424 group.Go(func() error { 425 - return a.ServeRTMPS(ctx, &cli) 426 }) 427 } else { 428 group.Go(func() error { ··· 453 }) 454 455 group.Go(func() error { 456 - return storage.StartSegmentCleaner(ctx, ldb, &cli) 457 }) 458 459 group.Go(func() error { ··· 461 }) 462 463 group.Go(func() error { 464 - return replicator.Start(ctx, &cli) 465 }) 466 467 if cli.LivepeerGateway { ··· 475 return err 476 } 477 group.Go(func() error { 478 - err := GoLivepeer(ctx, fs) 479 if err != nil { 480 return err 481 } ··· 497 return err 498 } 499 did := atkey.DIDKey() 500 - testMediaSigner, err := media.MakeMediaSigner(ctx, &cli, did, signer, mod) 501 if err != nil { 502 return err 503 } ··· 524 return err 525 } 526 did2 := atkey2.DIDKey() 527 - intermittentMediaSigner, err := media.MakeMediaSigner(ctx, &cli, did2, signer, mod) 528 if err != nil { 529 return err 530 } ··· 561 562 for _, job := range platformJobs { 563 group.Go(func() error { 564 - return job(ctx, &cli) 565 }) 566 } 567 568 if cli.WHIPTest != "" { 569 group.Go(func() error { 570 - err := WHIP(strings.Split(cli.WHIPTest, " ")) 0 0 0 571 log.Warn(ctx, "WHIP test complete, sleeping for 3 seconds and shutting down gstreamer") 572 time.Sleep(time.Second * 3) 573 // gst.Deinit() ··· 599 } 600 } 601 } 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0

··· 21 "github.com/bluesky-social/indigo/carstore" 22 "github.com/ethereum/go-ethereum/common/hexutil" 23 "github.com/livepeer/go-livepeer/cmd/livepeer/starter" 0 24 "github.com/streamplace/oatproxy/pkg/oatproxy" 25 + urfavecli "github.com/urfave/cli/v3" 26 "stream.place/streamplace/pkg/aqhttp" 27 "stream.place/streamplace/pkg/atproto" 28 "stream.place/streamplace/pkg/bus" ··· 54 // parse the CLI and fire up an streamplace node! 55 func start(build *config.BuildFlags, platformJobs []jobFunc) error { 56 iroh_streamplace.InitLogging() 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 57 58 + cli := config.CLI{Build: build} 59 + app := cli.NewCommand("streamplace") 60 + app.Usage = "decentralized live streaming platform" 61 + app.Version = build.Version 62 + app.Commands = []*urfavecli.Command{ 63 + makeSelfTestCommand(build), 64 + makeStreamCommand(build), 65 + makeLiveCommand(build), 66 + makeSignCommand(build), 67 + makeWhepCommand(build), 68 + makeWhipCommand(build), 69 + makeCombineCommand(build), 70 + makeSplitCommand(build), 71 + makeLivepeerCommand(build), 72 + makeMigrateCommand(build), 73 } 74 + // Add the verbosity flag 75 + // app.Flags = append(app.Flags, &urfavecli.StringFlag{ 76 + // Name: "v", 77 + // Usage: "log verbosity level", 78 + // Value: "3", 79 + // }) 80 + app.Before = func(ctx context.Context, cmd *urfavecli.Command) (context.Context, error) { 81 + // Run self-test before starting 82 + selfTest := cmd.Name == "self-test" 83 + err := media.RunSelfTest(ctx) 84 if err != nil { 85 + if selfTest { 86 + fmt.Println(err.Error()) 87 + os.Exit(1) 88 + } else { 89 + retryCount, _ := strconv.Atoi(os.Getenv("STREAMPLACE_SELFTEST_RETRY")) 90 + if retryCount >= 3 { 91 + log.Error(ctx, "gstreamer self-test failed 3 times, giving up", "error", err) 92 + return ctx, err 93 + } 94 + log.Log(ctx, "error in gstreamer self-test, attempting recovery", "error", err, "retry", retryCount+1) 95 + os.Setenv("STREAMPLACE_SELFTEST_RETRY", strconv.Itoa(retryCount+1)) 96 + err := syscall.Exec(os.Args[0], os.Args[1:], os.Environ()) 97 + if err != nil { 98 + log.Error(ctx, "error in gstreamer self-test, could not restart", "error", err) 99 + return ctx, err 100 + } 101 + panic("invalid code path: exec succeeded but we're still here???") 102 + } 103 } 104 + return ctx, nil 0 105 } 106 + app.Action = func(ctx context.Context, cmd *urfavecli.Command) error { 107 + return runMain(ctx, build, platformJobs, cmd, &cli) 0 108 } 109 110 + return app.Run(context.Background(), os.Args) 111 + } 0 0 0 0 0 0 0 0 112 113 + func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFunc, cmd *urfavecli.Command, cli *config.CLI) error { 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 114 _ = flag.Set("logtostderr", "true") 115 vFlag := flag.Lookup("v") 0 0 0 0 116 117 + err := cli.Validate(cmd) 0 0 118 if err != nil { 119 return err 120 } ··· 123 if err != nil { 124 return err 125 } 126 + verbosity := cmd.String("v") 127 + _ = vFlag.Value.Set(verbosity) 128 log.SetColorLogger(cli.Color) 0 129 ctx = log.WithDebugValue(ctx, cli.Debug) 130 131 log.Log(ctx, ··· 136 "runtime.GOOS", runtime.GOOS, 137 "runtime.GOARCH", runtime.GOARCH, 138 "runtime.Version", runtime.Version()) 139 + 140 + signer, err := createSigner(ctx, cli) 0 0 141 if err != nil { 142 return err 143 } 144 145 if len(os.Args) > 1 && os.Args[1] == "migrate" { 146 + return statedb.Migrate(cli) 147 } 148 149 spmetrics.Version.WithLabelValues(build.Version).Inc() ··· 193 if err != nil { 194 return err 195 } 196 + state, err := statedb.MakeDB(ctx, cli, noter, mod) 197 if err != nil { 198 return err 199 } 200 + handle, err := atproto.MakeLexiconRepo(ctx, cli, mod, state) 201 if err != nil { 202 return err 203 } ··· 217 218 b := bus.NewBus() 219 atsync := &atproto.ATProtoSynchronizer{ 220 + CLI: cli, 221 Model: mod, 222 StatefulDB: state, 223 Noter: noter, ··· 228 return fmt.Errorf("failed to migrate: %w", err) 229 } 230 231 + mm, err := media.MakeMediaManager(ctx, cli, signer, mod, b, atsync, ldb) 232 if err != nil { 233 return err 234 } 235 236 + ms, err := media.MakeMediaSigner(ctx, cli, cli.StreamerName, signer, mod) 237 if err != nil { 238 return err 239 } ··· 296 return err 297 } 298 } 299 + replicator, err = iroh_replicator.NewSwarm(ctx, cli, secret, topic, mm, b, mod) 300 if err != nil { 301 return err 302 } ··· 318 Public: cli.PublicOAuth, 319 HTTPClient: &aqhttp.Client, 320 }) 321 + d := director.NewDirector(mm, mod, cli, b, op, state, replicator, ldb) 322 + a, err := api.MakeStreamplaceAPI(cli, mod, state, noter, mm, ms, b, atsync, d, op, ldb) 323 if err != nil { 324 return err 325 } ··· 349 }) 350 if cli.RTMPServerAddon != "" { 351 group.Go(func() error { 352 + return rtmps.ServeRTMPSAddon(ctx, cli) 353 }) 354 } 355 group.Go(func() error { 356 + return a.ServeRTMPS(ctx, cli) 357 }) 358 } else { 359 group.Go(func() error { ··· 384 }) 385 386 group.Go(func() error { 387 + return storage.StartSegmentCleaner(ctx, ldb, cli) 388 }) 389 390 group.Go(func() error { ··· 392 }) 393 394 group.Go(func() error { 395 + return replicator.Start(ctx, cli) 396 }) 397 398 if cli.LivepeerGateway { ··· 406 return err 407 } 408 group.Go(func() error { 409 + err = GoLivepeer(ctx, config.LivepeerFlagSet) 410 if err != nil { 411 return err 412 } ··· 428 return err 429 } 430 did := atkey.DIDKey() 431 + testMediaSigner, err := media.MakeMediaSigner(ctx, cli, did, signer, mod) 432 if err != nil { 433 return err 434 } ··· 455 return err 456 } 457 did2 := atkey2.DIDKey() 458 + intermittentMediaSigner, err := media.MakeMediaSigner(ctx, cli, did2, signer, mod) 459 if err != nil { 460 return err 461 } ··· 492 493 for _, job := range platformJobs { 494 group.Go(func() error { 495 + return job(ctx, cli) 496 }) 497 } 498 499 if cli.WHIPTest != "" { 500 group.Go(func() error { 501 + // Parse WHIPTest string using the whip command's flag parser 502 + whipCmd := makeWhipCommand(build) 503 + args := strings.Split(cli.WHIPTest, " ") 504 + err := whipCmd.Run(ctx, append([]string{"streamplace", "whip"}, args...)) 505 log.Warn(ctx, "WHIP test complete, sleeping for 3 seconds and shutting down gstreamer") 506 time.Sleep(time.Second * 3) 507 // gst.Deinit() ··· 533 } 534 } 535 } 536 + 537 + func makeSelfTestCommand(build *config.BuildFlags) *urfavecli.Command { 538 + return &urfavecli.Command{ 539 + Name: "self-test", 540 + Usage: "run gstreamer self-test", 541 + Action: func(ctx context.Context, cmd *urfavecli.Command) error { 542 + err := media.RunSelfTest(ctx) 543 + if err != nil { 544 + fmt.Println(err.Error()) 545 + os.Exit(1) 546 + } 547 + runtime.GC() 548 + if err := pprof.Lookup("goroutine").WriteTo(os.Stderr, 2); err != nil { 549 + log.Error(ctx, "error creating pprof", "error", err) 550 + } 551 + fmt.Println("self-test successful!") 552 + return nil 553 + }, 554 + } 555 + } 556 + 557 + func makeStreamCommand(build *config.BuildFlags) *urfavecli.Command { 558 + return &urfavecli.Command{ 559 + Name: "stream", 560 + Usage: "stream command", 561 + ArgsUsage: "[user]", 562 + Action: func(ctx context.Context, cmd *urfavecli.Command) error { 563 + args := cmd.Args() 564 + if args.Len() != 1 { 565 + return fmt.Errorf("usage: streamplace stream [user]") 566 + } 567 + return Stream(args.First()) 568 + }, 569 + } 570 + } 571 + 572 + func makeLiveCommand(build *config.BuildFlags) *urfavecli.Command { 573 + cli := config.CLI{Build: build} 574 + liveCmd := cli.NewCommand("live") 575 + liveCmd.Usage = "start live stream" 576 + liveCmd.ArgsUsage = "[stream-key]" 577 + liveCmd.Action = func(ctx context.Context, cmd *urfavecli.Command) error { 578 + args := cmd.Args() 579 + if args.Len() != 1 { 580 + return fmt.Errorf("usage: streamplace live [flags] [stream-key]") 581 + } 582 + return Live(args.First(), cli.HTTPInternalAddr) 583 + } 584 + return liveCmd 585 + } 586 + 587 + func makeSignCommand(build *config.BuildFlags) *urfavecli.Command { 588 + return &urfavecli.Command{ 589 + Name: "sign", 590 + Usage: "sign command", 591 + Flags: []urfavecli.Flag{ 592 + &urfavecli.StringFlag{ 593 + Name: "cert", 594 + Usage: "path to the certificate file", 595 + }, 596 + &urfavecli.StringFlag{ 597 + Name: "key", 598 + Usage: "base58-encoded secp256k1 private key", 599 + }, 600 + &urfavecli.StringFlag{ 601 + Name: "streamer", 602 + Usage: "streamer name", 603 + }, 604 + &urfavecli.StringFlag{ 605 + Name: "ta-url", 606 + Usage: "timestamp authority server for signing", 607 + Value: "http://timestamp.digicert.com", 608 + }, 609 + &urfavecli.IntFlag{ 610 + Name: "start-time", 611 + Usage: "start time of the stream", 612 + }, 613 + &urfavecli.StringFlag{ 614 + Name: "manifest", 615 + Usage: "JSON manifest to use for signing", 616 + }, 617 + }, 618 + Action: func(ctx context.Context, cmd *urfavecli.Command) error { 619 + return Sign( 620 + ctx, 621 + cmd.String("cert"), 622 + cmd.String("key"), 623 + cmd.String("streamer"), 624 + cmd.String("ta-url"), 625 + int64(cmd.Int("start-time")), 626 + cmd.String("manifest"), 627 + ) 628 + }, 629 + } 630 + } 631 + 632 + func makeWhepCommand(build *config.BuildFlags) *urfavecli.Command { 633 + return &urfavecli.Command{ 634 + Name: "whep", 635 + Usage: "WHEP client", 636 + Flags: []urfavecli.Flag{ 637 + &urfavecli.IntFlag{ 638 + Name: "count", 639 + Usage: "number of concurrent streams (for load testing)", 640 + Value: 1, 641 + }, 642 + &urfavecli.DurationFlag{ 643 + Name: "duration", 644 + Usage: "stop after this long", 645 + }, 646 + &urfavecli.StringFlag{ 647 + Name: "endpoint", 648 + Usage: "endpoint to send the WHEP request to", 649 + }, 650 + }, 651 + Action: func(ctx context.Context, cmd *urfavecli.Command) error { 652 + return WHEP( 653 + ctx, 654 + cmd.Int("count"), 655 + cmd.Duration("duration"), 656 + cmd.String("endpoint"), 657 + ) 658 + }, 659 + } 660 + } 661 + 662 + func makeWhipCommand(build *config.BuildFlags) *urfavecli.Command { 663 + return &urfavecli.Command{ 664 + Name: "whip", 665 + Usage: "WHIP client", 666 + Flags: []urfavecli.Flag{ 667 + &urfavecli.StringFlag{ 668 + Name: "stream-key", 669 + Usage: "stream key", 670 + }, 671 + &urfavecli.IntFlag{ 672 + Name: "count", 673 + Usage: "number of concurrent streams (for load testing)", 674 + Value: 1, 675 + }, 676 + &urfavecli.IntFlag{ 677 + Name: "viewers", 678 + Usage: "number of viewers to simulate per stream", 679 + }, 680 + &urfavecli.DurationFlag{ 681 + Name: "duration", 682 + Usage: "duration of the stream", 683 + }, 684 + &urfavecli.StringFlag{ 685 + Name: "file", 686 + Usage: "file to stream (needs to be an MP4 containing H264 video and Opus audio)", 687 + Required: true, 688 + }, 689 + &urfavecli.StringFlag{ 690 + Name: "endpoint", 691 + Usage: "endpoint to send the WHIP request to", 692 + Value: "http://127.0.0.1:38080", 693 + }, 694 + &urfavecli.DurationFlag{ 695 + Name: "freeze-after", 696 + Usage: "freeze the stream after the given duration", 697 + }, 698 + }, 699 + Action: func(ctx context.Context, cmd *urfavecli.Command) error { 700 + return WHIP( 701 + ctx, 702 + cmd.String("stream-key"), 703 + cmd.Int("count"), 704 + cmd.Int("viewers"), 705 + cmd.Duration("duration"), 706 + cmd.String("file"), 707 + cmd.String("endpoint"), 708 + cmd.Duration("freeze-after"), 709 + ) 710 + }, 711 + } 712 + } 713 + 714 + func makeCombineCommand(build *config.BuildFlags) *urfavecli.Command { 715 + cli := config.CLI{Build: build} 716 + combineCmd := cli.NewCommand("combine") 717 + combineCmd.Usage = "combine segments" 718 + combineCmd.ArgsUsage = "[output] [input1] [input2...]" 719 + combineCmd.Flags = []urfavecli.Flag{ 720 + &urfavecli.StringFlag{ 721 + Name: "debug-dir", 722 + Usage: "directory to write debug output", 723 + }, 724 + } 725 + combineCmd.Action = func(ctx context.Context, cmd *urfavecli.Command) error { 726 + args := cmd.Args() 727 + if args.Len() < 2 { 728 + return fmt.Errorf("usage: streamplace combine [--debug-dir dir] [output] [input1] [input2...]") 729 + } 730 + ctx = log.WithDebugValue(ctx, cli.Debug) 731 + return Combine( 732 + ctx, 733 + &cli, 734 + cmd.String("debug-dir"), 735 + args.Get(0), 736 + args.Slice()[1:], 737 + ) 738 + } 739 + return combineCmd 740 + } 741 + 742 + func makeSplitCommand(build *config.BuildFlags) *urfavecli.Command { 743 + cli := config.CLI{Build: build} 744 + splitCmd := cli.NewCommand("split") 745 + splitCmd.Usage = "split video file" 746 + splitCmd.ArgsUsage = "[input file] [output directory]" 747 + splitCmd.Action = func(ctx context.Context, cmd *urfavecli.Command) error { 748 + args := cmd.Args() 749 + if args.Len() != 2 { 750 + return fmt.Errorf("usage: streamplace split [flags] [input file] [output directory]") 751 + } 752 + ctx = log.WithDebugValue(ctx, cli.Debug) 753 + gstinit.InitGST() 754 + return Split(ctx, args.Get(0), args.Get(1)) 755 + } 756 + return splitCmd 757 + } 758 + 759 + func makeLivepeerCommand(build *config.BuildFlags) *urfavecli.Command { 760 + return &urfavecli.Command{ 761 + Name: "livepeer", 762 + Usage: "run livepeer gateway", 763 + Action: func(ctx context.Context, cmd *urfavecli.Command) error { 764 + return GoLivepeer(ctx, config.LivepeerFlagSet) 765 + }, 766 + } 767 + } 768 + 769 + func makeMigrateCommand(build *config.BuildFlags) *urfavecli.Command { 770 + cli := config.CLI{Build: build} 771 + return &urfavecli.Command{ 772 + Name: "migrate", 773 + Usage: "run database migrations", 774 + Action: func(ctx context.Context, cmd *urfavecli.Command) error { 775 + return statedb.Migrate(&cli) 776 + }, 777 + } 778 + }

+5 -17

pkg/cmd/whep.go

··· 2 3 import ( 4 "context" 5 - "flag" 6 "fmt" 7 "io" 8 "net/http" ··· 15 "stream.place/streamplace/pkg/log" 16 ) 17 18 - func WHEP(args []string) error { 19 - fs := flag.NewFlagSet("whep", flag.ExitOnError) 20 - count := fs.Int("count", 1, "number of concurrent streams (for load testing)") 21 - duration := fs.Duration("duration", 0, "stop after this long") 22 - endpoint := fs.String("endpoint", "", "endpoint to send the WHEP request to") 23 - err := fs.Parse(args) 24 - 25 - if err != nil { 26 - return err 27 - } 28 - 29 - ctx := context.Background() 30 - if *duration > 0 { 31 var cancel context.CancelFunc 32 - ctx, cancel = context.WithTimeout(ctx, *duration) 33 defer cancel() 34 } 35 36 w := &WHEPClient{ 37 - Endpoint: *endpoint, 38 - Count: *count, 39 } 40 41 return w.WHEP(ctx)

··· 2 3 import ( 4 "context" 0 5 "fmt" 6 "io" 7 "net/http" ··· 14 "stream.place/streamplace/pkg/log" 15 ) 16 17 + func WHEP(ctx context.Context, count int, duration time.Duration, endpoint string) error { 18 + if duration > 0 { 0 0 0 0 0 0 0 0 0 0 0 19 var cancel context.CancelFunc 20 + ctx, cancel = context.WithTimeout(ctx, duration) 21 defer cancel() 22 } 23 24 w := &WHEPClient{ 25 + Endpoint: endpoint, 26 + Count: count, 27 } 28 29 return w.WHEP(ctx)

+10 -24

pkg/cmd/whip.go

··· 2 3 import ( 4 "context" 5 - "flag" 6 "fmt" 7 "io" 8 "net/http" ··· 20 "stream.place/streamplace/pkg/media" 21 ) 22 23 - func WHIP(args []string) error { 24 - fs := flag.NewFlagSet("whip", flag.ExitOnError) 25 - streamKey := fs.String("stream-key", "", "stream key") 26 - count := fs.Int("count", 1, "number of concurrent streams (for load testing)") 27 - viewers := fs.Int("viewers", 0, "number of viewers to simulate per stream") 28 - duration := fs.Duration("duration", 0, "duration of the stream") 29 - file := fs.String("file", "", "file to stream (needs to be an MP4 containing H264 video and Opus audio)") 30 - endpoint := fs.String("endpoint", "http://127.0.0.1:38080", "endpoint to send the WHIP request to") 31 - freezeAfter := fs.Duration("freeze-after", 0, "freeze the stream after the given duration") 32 - err := fs.Parse(args) 33 - if *file == "" { 34 return fmt.Errorf("file is required") 35 - } 36 - if err != nil { 37 - return err 38 } 39 gstinit.InitGST() 40 41 - ctx := context.Background() 42 - if *duration > 0 { 43 var cancel context.CancelFunc 44 - ctx, cancel = context.WithTimeout(ctx, *duration) 45 defer cancel() 46 } 47 48 w := &WHIPClient{ 49 - StreamKey: *streamKey, 50 - File: *file, 51 - Endpoint: *endpoint, 52 - Count: *count, 53 - FreezeAfter: *freezeAfter, 54 - Viewers: *viewers, 55 } 56 57 return w.WHIP(ctx)

··· 2 3 import ( 4 "context" 0 5 "fmt" 6 "io" 7 "net/http" ··· 19 "stream.place/streamplace/pkg/media" 20 ) 21 22 + func WHIP(ctx context.Context, streamKey string, count int, viewers int, duration time.Duration, file string, endpoint string, freezeAfter time.Duration) error { 23 + if file == "" { 0 0 0 0 0 0 0 0 0 24 return fmt.Errorf("file is required") 0 0 0 25 } 26 gstinit.InitGST() 27 28 + if duration > 0 { 0 29 var cancel context.CancelFunc 30 + ctx, cancel = context.WithTimeout(ctx, duration) 31 defer cancel() 32 } 33 34 w := &WHIPClient{ 35 + StreamKey: streamKey, 36 + File: file, 37 + Endpoint: endpoint, 38 + Count: count, 39 + FreezeAfter: freezeAfter, 40 + Viewers: viewers, 41 } 42 43 return w.WHIP(ctx)

+727 -231

pkg/config/config.go

··· 14 "os" 15 "path/filepath" 16 "runtime" 0 17 "strconv" 18 "strings" 19 "time" ··· 24 "github.com/livepeer/go-livepeer/cmd/livepeer/starter" 25 "github.com/lmittmann/tint" 26 slogGorm "github.com/orandin/slog-gorm" 27 - "github.com/peterbourgon/ff/v3" 28 "stream.place/streamplace/pkg/aqtime" 29 "stream.place/streamplace/pkg/constants" 30 "stream.place/streamplace/pkg/crypto/aqpub" ··· 160 ReplicatorIroh string = "iroh" 161 ) 162 163 - func (cli *CLI) NewFlagSet(name string) *flag.FlagSet { 164 - fs := flag.NewFlagSet("streamplace", flag.ExitOnError) 165 - fs.StringVar(&cli.DataDir, "data-dir", DefaultDataDir(), "directory for keeping all streamplace data") 166 - fs.StringVar(&cli.HTTPAddr, "http-addr", ":38080", "Public HTTP address") 167 - fs.StringVar(&cli.HTTPInternalAddr, "http-internal-addr", "127.0.0.1:39090", "Private, admin-only HTTP address") 168 - fs.StringVar(&cli.HTTPSAddr, "https-addr", ":38443", "Public HTTPS address") 169 - fs.BoolVar(&cli.Secure, "secure", false, "Run with HTTPS. Required for WebRTC output") 170 - cli.DataDirFlag(fs, &cli.TLSCertPath, "tls-cert", filepath.Join("tls", "tls.crt"), "Path to TLS certificate") 171 - cli.DataDirFlag(fs, &cli.TLSKeyPath, "tls-key", filepath.Join("tls", "tls.key"), "Path to TLS key") 172 - fs.StringVar(&cli.SigningKeyPath, "signing-key", "", "Path to signing key for pushing OTA updates to the app") 173 - fs.StringVar(&cli.DBURL, "db-url", "sqlite://$SP_DATA_DIR/state.sqlite", "URL of the database to use for storing private streamplace state") 174 - cli.dataDirFlags = append(cli.dataDirFlags, &cli.DBURL) 175 - fs.StringVar(&cli.AdminAccount, "admin-account", "", "ethereum account that administrates this streamplace node") 176 - fs.StringVar(&cli.FirebaseServiceAccount, "firebase-service-account", "", "Base64-encoded JSON string of a firebase service account key") 177 - fs.StringVar(&cli.FirebaseServiceAccountFile, "firebase-service-account-file", "", "Path to a JSON file containing a firebase service account key") 178 - fs.StringVar(&cli.GitLabURL, "gitlab-url", "https://git.stream.place/api/v4/projects/1", "gitlab url for generating download links") 179 - cli.DataDirFlag(fs, &cli.EthKeystorePath, "eth-keystore-path", "keystore", "path to ethereum keystore") 180 - fs.StringVar(&cli.EthAccountAddr, "eth-account-addr", "", "ethereum account address to use (if keystore contains more than one)") 181 - fs.StringVar(&cli.EthPassword, "eth-password", "", "password for encrypting keystore") 182 - fs.StringVar(&cli.TAURL, "ta-url", "http://timestamp.digicert.com", "timestamp authority server for signing") 183 - fs.StringVar(&cli.PKCS11ModulePath, "pkcs11-module-path", "", "path to a PKCS11 module for HSM signing, for example /usr/lib/x86_64-linux-gnu/opensc-pkcs11.so") 184 - fs.StringVar(&cli.PKCS11Pin, "pkcs11-pin", "", "PIN for logging into PKCS11 token. if not provided, will be prompted interactively") 185 - fs.StringVar(&cli.PKCS11TokenSlot, "pkcs11-token-slot", "", "slot number of PKCS11 token (only use one of slot, label, or serial)") 186 - fs.StringVar(&cli.PKCS11TokenLabel, "pkcs11-token-label", "", "label of PKCS11 token (only use one of slot, label, or serial)") 187 - fs.StringVar(&cli.PKCS11TokenSerial, "pkcs11-token-serial", "", "serial number of PKCS11 token (only use one of slot, label, or serial)") 188 - fs.StringVar(&cli.PKCS11KeypairLabel, "pkcs11-keypair-label", "", "label of signing keypair on PKCS11 token") 189 - fs.StringVar(&cli.PKCS11KeypairID, "pkcs11-keypair-id", "", "id of signing keypair on PKCS11 token") 190 - fs.StringVar(&cli.AppBundleID, "app-bundle-id", "", "bundle id of an app that we facilitate oauth login for") 191 - fs.StringVar(&cli.StreamerName, "streamer-name", "", "name of the person streaming from this streamplace node") 192 - fs.StringVar(&cli.FrontendProxy, "dev-frontend-proxy", "", "(FOR DEVELOPMENT ONLY) proxy frontend requests to this address instead of using the bundled frontend") 193 - fs.BoolVar(&cli.PublicOAuth, "dev-public-oauth", false, "(FOR DEVELOPMENT ONLY) enable public oauth login for http://127.0.0.1 development") 194 - fs.StringVar(&cli.LivepeerGatewayURL, "livepeer-gateway-url", "", "URL of the Livepeer Gateway to use for transcoding") 195 - fs.BoolVar(&cli.LivepeerGateway, "livepeer-gateway", false, "enable embedded Livepeer Gateway") 196 - fs.BoolVar(&cli.WideOpen, "wide-open", false, "allow ALL streams to be uploaded to this node (not recommended for production)") 197 - cli.StringSliceFlag(fs, &cli.AllowedStreams, "allowed-streams", []string{}, "if set, only allow these addresses or atproto DIDs to upload to this node") 198 - cli.StringSliceFlag(fs, &cli.Peers, "peers", []string{}, "other streamplace nodes to replicate to") 199 - cli.StringSliceFlag(fs, &cli.Redirects, "redirects", []string{}, "http 302s /path/one:/path/two,/path/three:/path/four") 200 - cli.DebugFlag(fs, &cli.Debug, "debug", "", "modified log verbosity for specific functions or files in form func=ToHLS:3,file=gstreamer.go:4") 201 - fs.BoolVar(&cli.TestStream, "test-stream", false, "run a built-in test stream on boot") 202 - fs.BoolVar(&cli.NoFirehose, "no-firehose", false, "disable the bluesky firehose") 203 - fs.BoolVar(&cli.PrintChat, "print-chat", false, "print chat messages to stdout") 204 - fs.StringVar(&cli.WHIPTest, "whip-test", "", "run a WHIP self-test with the given parameters") 205 - fs.StringVar(&cli.RelayHost, "relay-host", "wss://bsky.network", "websocket url for relay firehose") 206 - fs.StringVar(&cli.Color, "color", "", "'true' to enable colorized logging, 'false' to disable") 207 - fs.StringVar(&cli.BroadcasterHost, "broadcaster-host", "", "public host for the broadcaster group that this node is a part of (excluding https:// e.g. stream.place)") 208 - fs.StringVar(&cli.XXDeprecatedPublicHost, "public-host", "", "deprecated, use broadcaster-host or server-host instead as appropriate") 209 - fs.StringVar(&cli.ServerHost, "server-host", "", "public host for this particular physical streamplace node. defaults to broadcaster-host and only must be set for multi-node broadcasters") 210 - fs.BoolVar(&cli.Thumbnail, "thumbnail", true, "enable thumbnail generation") 211 - fs.BoolVar(&cli.SmearAudio, "smear-audio", false, "enable audio smearing to create 'perfect' segment timestamps") 212 213 - fs.StringVar(&cli.TracingEndpoint, "tracing-endpoint", "", "gRPC endpoint to send traces to") 214 - fs.IntVar(&cli.RateLimitPerSecond, "rate-limit-per-second", 0, "rate limit for requests per second per ip") 215 - fs.IntVar(&cli.RateLimitBurst, "rate-limit-burst", 0, "rate limit burst for requests per ip") 216 - fs.IntVar(&cli.RateLimitWebsocket, "rate-limit-websocket", 10, "number of concurrent websocket connections allowed per ip") 217 - fs.StringVar(&cli.RTMPServerAddon, "rtmp-server-addon", "", "address of external RTMP server to forward streams to") 218 - fs.StringVar(&cli.RTMPSAddonAddr, "rtmps-addon-addr", ":1936", "address to listen for RTMPS on the addon server") 219 - fs.StringVar(&cli.RTMPSAddr, "rtmps-addr", ":1935", "address to listen for RTMPS connections (when --secure=true)") 220 - fs.StringVar(&cli.RTMPAddr, "rtmp-addr", ":1935", "address to listen for RTMP connections (when --secure=false)") 221 - cli.JSONFlag(fs, &cli.DiscordWebhooks, "discord-webhooks", "[]", "JSON array of Discord webhooks to send notifications to") 222 - fs.BoolVar(&cli.NewWebRTCPlayback, "new-webrtc-playback", true, "enable new webrtc playback") 223 - fs.StringVar(&cli.AppleTeamID, "apple-team-id", "", "apple team id for deep linking") 224 - fs.StringVar(&cli.AndroidCertFingerprint, "android-cert-fingerprint", "", "android cert fingerprint for deep linking") 225 - cli.StringSliceFlag(fs, &cli.Labelers, "labelers", []string{}, "did of labelers that this instance should subscribe to") 226 - fs.StringVar(&cli.AtprotoDID, "atproto-did", "", "atproto did to respond to on /.well-known/atproto-did (default did:web:PUBLIC_HOST)") 227 - cli.JSONFlag(fs, &cli.ContentFilters, "content-filters", "{}", "JSON content filtering rules") 228 - cli.StringSliceFlag(fs, &cli.DefaultRecommendedStreamers, "default-recommended-streamers", []string{}, "comma-separated list of streamer DIDs to recommend by default when no other recommendations are available") 229 - fs.BoolVar(&cli.LivepeerHelp, "livepeer-help", false, "print help for livepeer flags and exit") 230 - fs.StringVar(&cli.PLCURL, "plc-url", "https://plc.directory", "url of the plc directory") 231 - fs.BoolVar(&cli.SQLLogging, "sql-logging", false, "enable sql logging") 232 - fs.StringVar(&cli.SentryDSN, "sentry-dsn", "", "sentry dsn for error reporting") 233 - fs.BoolVar(&cli.LivepeerDebug, "livepeer-debug", false, "log livepeer segments to $SP_DATA_DIR/livepeer-debug") 234 - fs.StringVar(&cli.SegmentDebugDir, "segment-debug-dir", "", "directory to log segment validation to") 235 - cli.StringSliceFlag(fs, &cli.Tickets, "tickets", []string{}, "tickets to join the swarm with") 236 - fs.StringVar(&cli.IrohTopic, "iroh-topic", "", "topic to use for the iroh swarm (must be 32 bytes in hex)") 237 - fs.BoolVar(&cli.DisableIrohRelay, "disable-iroh-relay", false, "disable the iroh relay") 238 - cli.KVSliceFlag(fs, &cli.DevAccountCreds, "dev-account-creds", "", "(FOR DEVELOPMENT ONLY) did=password pairs for logging into test accounts without oauth") 239 - fs.DurationVar(&cli.StreamSessionTimeout, "stream-session-timeout", 60*time.Second, "how long to wait before considering a stream inactive on this node?") 240 - cli.StringSliceFlag(fs, &cli.Replicators, "replicators", []string{ReplicatorWebsocket}, "list of replication protocols to use (http, iroh)") 241 - fs.StringVar(&cli.WebsocketURL, "websocket-url", "", "override the websocket (ws:// or wss://) url to use for replication (normally not necessary, used for testing)") 242 - fs.BoolVar(&cli.BehindHTTPSProxy, "behind-https-proxy", false, "set to true if this node is behind an https proxy and we should report https URLs even though the node isn't serving HTTPS") 243 - cli.StringSliceFlag(fs, &cli.AdminDIDs, "admin-dids", []string{}, "comma-separated list of DIDs that are authorized to modify branding and other admin operations") 244 - cli.StringSliceFlag(fs, &cli.Syndicate, "syndicate", []string{}, "list of DIDs that we should rebroadcast ('*' for everybody)") 245 - fs.BoolVar(&cli.PlayerTelemetry, "player-telemetry", true, "enable player telemetry") 246 - fs.StringVar(&cli.LocalDBURL, "local-db-url", "sqlite://$SP_DATA_DIR/localdb.sqlite", "URL of the local database to use for storing local data") 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 247 cli.dataDirFlags = append(cli.dataDirFlags, &cli.LocalDBURL) 0 0 0 248 249 - fs.Bool("external-signing", true, "DEPRECATED, does nothing.") 250 - fs.Bool("insecure", false, "DEPRECATED, does nothing.") 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 251 252 - lpFlags := flag.NewFlagSet("livepeer", flag.ContinueOnError) 253 - _ = starter.NewLivepeerConfig(lpFlags) 254 - lpFlags.VisitAll(func(f *flag.Flag) { 255 adapted := LivepeerFlags.CamelToSnake[f.Name] 256 - fs.Var(f.Value, fmt.Sprintf("livepeer.%s", adapted), f.Usage) 0 0 0 0 0 0 0 257 }) 258 259 - if runtime.GOOS == "linux" { 260 - fs.BoolVar(&cli.NoMist, "no-mist", true, "Disable MistServer") 261 - fs.IntVar(&cli.MistAdminPort, "mist-admin-port", 14242, "MistServer admin port (internal use only)") 262 - fs.IntVar(&cli.MistRTMPPort, "mist-rtmp-port", 11935, "MistServer RTMP port (internal use only)") 263 - fs.IntVar(&cli.MistHTTPPort, "mist-http-port", 18080, "MistServer HTTP port (internal use only)") 264 - } 265 - return fs 266 } 267 268 var StreamplaceSchemePrefix = "streamplace://" ··· 350 ) 351 } 352 353 - func (cli *CLI) Parse(fs *flag.FlagSet, args []string) error { 354 - err := ff.Parse( 355 - fs, args, 356 - ff.WithEnvVarPrefix("SP"), 357 - ) 358 - if err != nil { 359 - return err 360 - } 361 if cli.DataDir == "" { 362 return fmt.Errorf("could not determine default data dir (no $HOME) and none provided, please set --data-dir") 363 } ··· 366 } 367 if cli.LivepeerGateway { 368 log.MonkeypatchStderr() 369 - gatewayPath := cli.DataFilePath([]string{"livepeer", "gateway"}) 370 - err = fs.Set("livepeer.rtmp-addr", "127.0.0.1:0") 371 - if err != nil { 372 - return err 373 - } 374 - err = fs.Set("livepeer.data-dir", gatewayPath) 375 - if err != nil { 376 - return err 377 - } 378 - err = fs.Set("livepeer.gateway", "true") 379 - if err != nil { 380 - return err 381 - } 382 - httpAddrFlag := fs.Lookup("livepeer.http-addr") 383 - if httpAddrFlag == nil { 384 - return fmt.Errorf("livepeer.http-addr not found") 385 - } 386 - httpAddr := httpAddrFlag.Value.String() 387 - if httpAddr == "" { 388 - httpAddr = "127.0.0.1:8935" 389 - err = fs.Set("livepeer.http-addr", httpAddr) 390 - if err != nil { 391 - return err 392 - } 393 - } 394 - cli.LivepeerGatewayURL = fmt.Sprintf("http://%s", httpAddr) 395 } 396 for _, dest := range cli.dataDirFlags { 397 *dest = strings.Replace(*dest, SPDataDir, cli.DataDir, 1) ··· 420 return err 421 } 422 cli.FirebaseServiceAccount = string(bs) 0 0 0 0 423 } 424 return nil 425 } ··· 529 return nil 530 } 531 532 - func (cli *CLI) DataDirFlag(fs *flag.FlagSet, dest *string, name, defaultValue, usage string) { 533 - cli.dataDirFlags = append(cli.dataDirFlags, dest) 534 - *dest = filepath.Join(SPDataDir, defaultValue) 535 - usage = fmt.Sprintf(`%s (default: "%s")`, usage, *dest) 536 - fs.Func(name, usage, func(s string) error { 537 - *dest = s 538 - return nil 539 - }) 540 - } 541 - 542 func (cli *CLI) HasMist() bool { 543 return runtime.GOOS == "linux" 544 } 545 546 // type for comma-separated ethereum addresses 547 - func (cli *CLI) AddressSliceFlag(fs *flag.FlagSet, dest *[]aqpub.Pub, name, defaultValue, usage string) { 548 *dest = []aqpub.Pub{} 549 - usage = fmt.Sprintf(`%s (default: "%s")`, usage, *dest) 550 - fs.Func(name, usage, func(s string) error { 551 - if s == "" { 552 - return nil 553 - } 554 - strs := strings.Split(s, ",") 555 - for _, str := range strs { 556 - pub, err := aqpub.FromHexString(str) 557 - if err != nil { 558 - return err 559 - } 560 - *dest = append(*dest, pub) 561 - } 562 - return nil 563 - }) 564 - } 565 - 566 - func (cli *CLI) StringSliceFlag(fs *flag.FlagSet, dest *[]string, name string, defaultValue []string, usage string) { 567 - *dest = defaultValue 568 - usage = fmt.Sprintf(`%s (default: "%s")`, usage, *dest) 569 - fs.Func(name, usage, func(s string) error { 570 - if s == "" { 571 - return nil 572 - } 573 - strs := strings.Split(s, ",") 574 - *dest = append([]string{}, strs...) 575 - return nil 576 - }) 577 - } 578 - 579 - func (cli *CLI) KVSliceFlag(fs *flag.FlagSet, dest *map[string]string, name, defaultValue, usage string) { 580 - *dest = map[string]string{} 581 - usage = fmt.Sprintf(`%s (default: "%s")`, usage, *dest) 582 - fs.Func(name, usage, func(s string) error { 583 - if s == "" { 584 - return nil 585 - } 586 - pairs := strings.Split(s, ",") 587 - for _, pair := range pairs { 588 - parts := strings.Split(pair, "=") 589 - if len(parts) != 2 { 590 - return fmt.Errorf("invalid kv flag: %s", pair) 591 - } 592 - (*dest)[parts[0]] = parts[1] 593 - } 594 - return nil 595 - }) 596 - } 597 - 598 - func (cli *CLI) JSONFlag(fs *flag.FlagSet, dest any, name, defaultValue, usage string) { 599 usage = fmt.Sprintf(`%s (default: "%s")`, usage, defaultValue) 600 - fs.Func(name, usage, func(s string) error { 601 - if s == "" { 602 - return nil 603 - } 604 - return json.Unmarshal([]byte(s), dest) 605 - }) 606 - } 607 608 - // debug flag for turning func=ToHLS:3,file=gstreamer.go:4 into {"func": {"ToHLS": 3}, "file": {"gstreamer.go": 4}} 609 - func (cli *CLI) DebugFlag(fs *flag.FlagSet, dest *map[string]map[string]int, name, defaultValue, usage string) { 610 - *dest = map[string]map[string]int{} 611 - fs.Func(name, usage, func(s string) error { 612 - if s == "" { 613 - return nil 614 - } 615 - pairs := strings.Split(s, ",") 616 - for _, pair := range pairs { 617 - scoreSplit := strings.Split(pair, ":") 618 - if len(scoreSplit) != 2 { 619 - return fmt.Errorf("invalid debug flag: %s", pair) 620 } 621 - score, err := strconv.Atoi(scoreSplit[1]) 622 - if err != nil { 623 - return fmt.Errorf("invalid debug flag: %s", pair) 0 0 0 0 624 } 625 - selectorSplit := strings.Split(scoreSplit[0], "=") 626 - if len(selectorSplit) != 2 { 627 - return fmt.Errorf("invalid debug flag: %s", pair) 628 - } 629 - _, ok := (*dest)[selectorSplit[0]] 630 - if !ok { 631 - (*dest)[selectorSplit[0]] = map[string]int{} 632 - } 633 - (*dest)[selectorSplit[0]][selectorSplit[1]] = score 634 - } 635 - 636 - return nil 637 - }) 638 } 639 640 func (cli *CLI) StreamIsAllowed(did string) error { ··· 648 if openServer && !isDIDKey { 649 return nil 650 } 651 - for _, a := range cli.AllowedStreams { 652 - if a == did { 653 - return nil 654 - } 655 } 656 return fmt.Errorf("user is not allowed to stream") 657 }

··· 14 "os" 15 "path/filepath" 16 "runtime" 17 + "slices" 18 "strconv" 19 "strings" 20 "time" ··· 25 "github.com/livepeer/go-livepeer/cmd/livepeer/starter" 26 "github.com/lmittmann/tint" 27 slogGorm "github.com/orandin/slog-gorm" 28 + urfavecli "github.com/urfave/cli/v3" 29 "stream.place/streamplace/pkg/aqtime" 30 "stream.place/streamplace/pkg/constants" 31 "stream.place/streamplace/pkg/crypto/aqpub" ··· 161 ReplicatorIroh string = "iroh" 162 ) 163 164 + var LivepeerFlagSet *flag.FlagSet 165 + var LivepeerConfig starter.LivepeerConfig 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 166 167 + func (cli *CLI) NewCommand(name string) *urfavecli.Command { 168 + cmd := &urfavecli.Command{ 169 + Name: name, 170 + Usage: "streamplace server", 171 + Flags: []urfavecli.Flag{ 172 + &urfavecli.StringFlag{ 173 + Name: "data-dir", 174 + Usage: "directory for keeping all streamplace data", 175 + Value: DefaultDataDir(), 176 + Destination: &cli.DataDir, 177 + Sources: urfavecli.EnvVars("SP_DATA_DIR"), 178 + }, 179 + &urfavecli.StringFlag{ 180 + Name: "http-addr", 181 + Usage: "Public HTTP address", 182 + Value: ":38080", 183 + Destination: &cli.HTTPAddr, 184 + Sources: urfavecli.EnvVars("SP_HTTP_ADDR"), 185 + }, 186 + &urfavecli.StringFlag{ 187 + Name: "http-internal-addr", 188 + Usage: "Private, admin-only HTTP address", 189 + Value: "127.0.0.1:39090", 190 + Destination: &cli.HTTPInternalAddr, 191 + Sources: urfavecli.EnvVars("SP_HTTP_INTERNAL_ADDR"), 192 + }, 193 + &urfavecli.StringFlag{ 194 + Name: "https-addr", 195 + Usage: "Public HTTPS address", 196 + Value: ":38443", 197 + Destination: &cli.HTTPSAddr, 198 + Sources: urfavecli.EnvVars("SP_HTTPS_ADDR"), 199 + }, 200 + &urfavecli.BoolFlag{ 201 + Name: "secure", 202 + Usage: "Run with HTTPS. Required for WebRTC output", 203 + Value: false, 204 + Destination: &cli.Secure, 205 + Sources: urfavecli.EnvVars("SP_SECURE"), 206 + }, 207 + &urfavecli.StringFlag{ 208 + Name: "tls-cert", 209 + Usage: fmt.Sprintf(`Path to TLS certificate (default: "%s")`, filepath.Join(SPDataDir, "tls", "tls.crt")), 210 + Destination: &cli.TLSCertPath, 211 + Value: filepath.Join(SPDataDir, "tls", "tls.crt"), 212 + Sources: urfavecli.EnvVars("SP_TLS_CERT"), 213 + }, 214 + &urfavecli.StringFlag{ 215 + Name: "tls-key", 216 + Usage: fmt.Sprintf(`Path to TLS key (default: "%s")`, filepath.Join(SPDataDir, "tls", "tls.key")), 217 + Destination: &cli.TLSKeyPath, 218 + Value: filepath.Join(SPDataDir, "tls", "tls.key"), 219 + Sources: urfavecli.EnvVars("SP_TLS_KEY"), 220 + }, 221 + &urfavecli.StringFlag{ 222 + Name: "signing-key", 223 + Usage: "Path to signing key for pushing OTA updates to the app", 224 + Destination: &cli.SigningKeyPath, 225 + Sources: urfavecli.EnvVars("SP_SIGNING_KEY"), 226 + }, 227 + &urfavecli.StringFlag{ 228 + Name: "db-url", 229 + Usage: "URL of the database to use for storing private streamplace state", 230 + Value: "sqlite://$SP_DATA_DIR/state.sqlite", 231 + Destination: &cli.DBURL, 232 + Sources: urfavecli.EnvVars("SP_DB_URL"), 233 + }, 234 + &urfavecli.StringFlag{ 235 + Name: "admin-account", 236 + Usage: "ethereum account that administrates this streamplace node", 237 + Destination: &cli.AdminAccount, 238 + Sources: urfavecli.EnvVars("SP_ADMIN_ACCOUNT"), 239 + }, 240 + &urfavecli.StringFlag{ 241 + Name: "firebase-service-account", 242 + Usage: "Base64-encoded JSON string of a firebase service account key", 243 + Destination: &cli.FirebaseServiceAccount, 244 + Sources: urfavecli.EnvVars("SP_FIREBASE_SERVICE_ACCOUNT"), 245 + }, 246 + &urfavecli.StringFlag{ 247 + Name: "firebase-service-account-file", 248 + Usage: "Path to a JSON file containing a firebase service account key", 249 + Destination: &cli.FirebaseServiceAccountFile, 250 + Sources: urfavecli.EnvVars("SP_FIREBASE_SERVICE_ACCOUNT_FILE"), 251 + }, 252 + &urfavecli.StringFlag{ 253 + Name: "gitlab-url", 254 + Usage: "gitlab url for generating download links", 255 + Value: "https://git.stream.place/api/v4/projects/1", 256 + Destination: &cli.GitLabURL, 257 + Sources: urfavecli.EnvVars("SP_GITLAB_URL"), 258 + }, 259 + &urfavecli.StringFlag{ 260 + Name: "eth-keystore-path", 261 + Usage: fmt.Sprintf(`path to ethereum keystore (default: "%s")`, filepath.Join(SPDataDir, "keystore")), 262 + Destination: &cli.EthKeystorePath, 263 + Value: filepath.Join(SPDataDir, "keystore"), 264 + Sources: urfavecli.EnvVars("SP_ETH_KEYSTORE_PATH"), 265 + }, 266 + &urfavecli.StringFlag{ 267 + Name: "eth-account-addr", 268 + Usage: "ethereum account address to use (if keystore contains more than one)", 269 + Destination: &cli.EthAccountAddr, 270 + Sources: urfavecli.EnvVars("SP_ETH_ACCOUNT_ADDR"), 271 + }, 272 + &urfavecli.StringFlag{ 273 + Name: "eth-password", 274 + Usage: "password for encrypting keystore", 275 + Destination: &cli.EthPassword, 276 + Sources: urfavecli.EnvVars("SP_ETH_PASSWORD"), 277 + }, 278 + &urfavecli.StringFlag{ 279 + Name: "ta-url", 280 + Usage: "timestamp authority server for signing", 281 + Value: "http://timestamp.digicert.com", 282 + Destination: &cli.TAURL, 283 + Sources: urfavecli.EnvVars("SP_TA_URL"), 284 + }, 285 + &urfavecli.StringFlag{ 286 + Name: "pkcs11-module-path", 287 + Usage: "path to a PKCS11 module for HSM signing, for example /usr/lib/x86_64-linux-gnu/opensc-pkcs11.so", 288 + Destination: &cli.PKCS11ModulePath, 289 + Sources: urfavecli.EnvVars("SP_PKCS11_MODULE_PATH"), 290 + }, 291 + &urfavecli.StringFlag{ 292 + Name: "pkcs11-pin", 293 + Usage: "PIN for logging into PKCS11 token. if not provided, will be prompted interactively", 294 + Destination: &cli.PKCS11Pin, 295 + Sources: urfavecli.EnvVars("SP_PKCS11_PIN"), 296 + }, 297 + &urfavecli.StringFlag{ 298 + Name: "pkcs11-token-slot", 299 + Usage: "slot number of PKCS11 token (only use one of slot, label, or serial)", 300 + Destination: &cli.PKCS11TokenSlot, 301 + Sources: urfavecli.EnvVars("SP_PKCS11_TOKEN_SLOT"), 302 + }, 303 + &urfavecli.StringFlag{ 304 + Name: "pkcs11-token-label", 305 + Usage: "label of PKCS11 token (only use one of slot, label, or serial)", 306 + Destination: &cli.PKCS11TokenLabel, 307 + Sources: urfavecli.EnvVars("SP_PKCS11_TOKEN_LABEL"), 308 + }, 309 + &urfavecli.StringFlag{ 310 + Name: "pkcs11-token-serial", 311 + Usage: "serial number of PKCS11 token (only use one of slot, label, or serial)", 312 + Destination: &cli.PKCS11TokenSerial, 313 + Sources: urfavecli.EnvVars("SP_PKCS11_TOKEN_SERIAL"), 314 + }, 315 + &urfavecli.StringFlag{ 316 + Name: "pkcs11-keypair-label", 317 + Usage: "label of signing keypair on PKCS11 token", 318 + Destination: &cli.PKCS11KeypairLabel, 319 + Sources: urfavecli.EnvVars("SP_PKCS11_KEYPAIR_LABEL"), 320 + }, 321 + &urfavecli.StringFlag{ 322 + Name: "pkcs11-keypair-id", 323 + Usage: "id of signing keypair on PKCS11 token", 324 + Destination: &cli.PKCS11KeypairID, 325 + Sources: urfavecli.EnvVars("SP_PKCS11_KEYPAIR_ID"), 326 + }, 327 + &urfavecli.StringFlag{ 328 + Name: "app-bundle-id", 329 + Usage: "bundle id of an app that we facilitate oauth login for", 330 + Destination: &cli.AppBundleID, 331 + Sources: urfavecli.EnvVars("SP_APP_BUNDLE_ID"), 332 + }, 333 + &urfavecli.StringFlag{ 334 + Name: "streamer-name", 335 + Usage: "name of the person streaming from this streamplace node", 336 + Destination: &cli.StreamerName, 337 + Sources: urfavecli.EnvVars("SP_STREAMER_NAME"), 338 + }, 339 + &urfavecli.StringFlag{ 340 + Name: "dev-frontend-proxy", 341 + Usage: "(FOR DEVELOPMENT ONLY) proxy frontend requests to this address instead of using the bundled frontend", 342 + Destination: &cli.FrontendProxy, 343 + Sources: urfavecli.EnvVars("SP_DEV_FRONTEND_PROXY"), 344 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 345 + if s == "false" { 346 + cli.FrontendProxy = "" 347 + return nil 348 + } 349 + cli.FrontendProxy = s 350 + return nil 351 + }, 352 + }, 353 + &urfavecli.BoolFlag{ 354 + Name: "dev-public-oauth", 355 + Usage: "(FOR DEVELOPMENT ONLY) enable public oauth login for http://127.0.0.1 development", 356 + Value: false, 357 + Destination: &cli.PublicOAuth, 358 + Sources: urfavecli.EnvVars("SP_DEV_PUBLIC_OAUTH"), 359 + }, 360 + &urfavecli.StringFlag{ 361 + Name: "livepeer-gateway-url", 362 + Usage: "URL of the Livepeer Gateway to use for transcoding", 363 + Destination: &cli.LivepeerGatewayURL, 364 + Sources: urfavecli.EnvVars("SP_LIVEPEER_GATEWAY_URL"), 365 + }, 366 + &urfavecli.BoolFlag{ 367 + Name: "livepeer-gateway", 368 + Usage: "enable embedded Livepeer Gateway", 369 + Value: false, 370 + Destination: &cli.LivepeerGateway, 371 + Sources: urfavecli.EnvVars("SP_LIVEPEER_GATEWAY"), 372 + }, 373 + &urfavecli.BoolFlag{ 374 + Name: "wide-open", 375 + Usage: "allow ALL streams to be uploaded to this node (not recommended for production)", 376 + Value: false, 377 + Destination: &cli.WideOpen, 378 + Sources: urfavecli.EnvVars("SP_WIDE_OPEN"), 379 + }, 380 + &urfavecli.StringFlag{ 381 + Name: "allowed-streams", 382 + Usage: `if set, only allow these addresses or atproto DIDs to upload to this node (default: "")`, 383 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 384 + if s == "" { 385 + return nil 386 + } 387 + cli.AllowedStreams = strings.Split(s, ",") 388 + return nil 389 + }, 390 + Sources: urfavecli.EnvVars("SP_ALLOWED_STREAMS"), 391 + }, 392 + &urfavecli.StringFlag{ 393 + Name: "peers", 394 + Usage: `other streamplace nodes to replicate to (default: "")`, 395 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 396 + if s == "" { 397 + return nil 398 + } 399 + cli.Peers = strings.Split(s, ",") 400 + return nil 401 + }, 402 + Sources: urfavecli.EnvVars("SP_PEERS"), 403 + }, 404 + &urfavecli.StringFlag{ 405 + Name: "redirects", 406 + Usage: `http 302s /path/one:/path/two,/path/three:/path/four (default: "")`, 407 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 408 + if s == "" { 409 + return nil 410 + } 411 + cli.Redirects = strings.Split(s, ",") 412 + return nil 413 + }, 414 + Sources: urfavecli.EnvVars("SP_REDIRECTS"), 415 + }, 416 + &urfavecli.StringFlag{ 417 + Name: "debug", 418 + Usage: "modified log verbosity for specific functions or files in form func=ToHLS:3,file=gstreamer.go:4", 419 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 420 + if s == "" { 421 + return nil 422 + } 423 + cli.Debug = map[string]map[string]int{} 424 + pairs := strings.SplitSeq(s, ",") 425 + for pair := range pairs { 426 + scoreSplit := strings.Split(pair, ":") 427 + if len(scoreSplit) != 2 { 428 + return fmt.Errorf("invalid debug flag: %s", pair) 429 + } 430 + score, err := strconv.Atoi(scoreSplit[1]) 431 + if err != nil { 432 + return fmt.Errorf("invalid debug flag: %s", pair) 433 + } 434 + selectorSplit := strings.Split(scoreSplit[0], "=") 435 + if len(selectorSplit) != 2 { 436 + return fmt.Errorf("invalid debug flag: %s", pair) 437 + } 438 + _, ok := cli.Debug[selectorSplit[0]] 439 + if !ok { 440 + cli.Debug[selectorSplit[0]] = map[string]int{} 441 + } 442 + cli.Debug[selectorSplit[0]][selectorSplit[1]] = score 443 + } 444 + return nil 445 + }, 446 + Sources: urfavecli.EnvVars("SP_DEBUG"), 447 + }, 448 + &urfavecli.BoolFlag{ 449 + Name: "test-stream", 450 + Usage: "run a built-in test stream on boot", 451 + Value: false, 452 + Destination: &cli.TestStream, 453 + Sources: urfavecli.EnvVars("SP_TEST_STREAM"), 454 + }, 455 + &urfavecli.BoolFlag{ 456 + Name: "no-firehose", 457 + Usage: "disable the bluesky firehose", 458 + Value: false, 459 + Destination: &cli.NoFirehose, 460 + Sources: urfavecli.EnvVars("SP_NO_FIREHOSE"), 461 + }, 462 + &urfavecli.BoolFlag{ 463 + Name: "print-chat", 464 + Usage: "print chat messages to stdout", 465 + Value: false, 466 + Destination: &cli.PrintChat, 467 + Sources: urfavecli.EnvVars("SP_PRINT_CHAT"), 468 + }, 469 + &urfavecli.StringFlag{ 470 + Name: "whip-test", 471 + Usage: "run a WHIP self-test with the given parameters", 472 + Destination: &cli.WHIPTest, 473 + Sources: urfavecli.EnvVars("SP_WHIP_TEST"), 474 + }, 475 + &urfavecli.StringFlag{ 476 + Name: "relay-host", 477 + Usage: "websocket url for relay firehose", 478 + Value: "wss://bsky.network", 479 + Destination: &cli.RelayHost, 480 + Sources: urfavecli.EnvVars("SP_RELAY_HOST"), 481 + }, 482 + &urfavecli.StringFlag{ 483 + Name: "color", 484 + Usage: "'true' to enable colorized logging, 'false' to disable", 485 + Destination: &cli.Color, 486 + Sources: urfavecli.EnvVars("SP_COLOR"), 487 + }, 488 + &urfavecli.StringFlag{ 489 + Name: "broadcaster-host", 490 + Usage: "public host for the broadcaster group that this node is a part of (excluding https:// e.g. stream.place)", 491 + Destination: &cli.BroadcasterHost, 492 + Sources: urfavecli.EnvVars("SP_BROADCASTER_HOST"), 493 + }, 494 + &urfavecli.StringFlag{ 495 + Name: "public-host", 496 + Usage: "deprecated, use broadcaster-host or server-host instead as appropriate", 497 + Destination: &cli.XXDeprecatedPublicHost, 498 + Sources: urfavecli.EnvVars("SP_PUBLIC_HOST"), 499 + }, 500 + &urfavecli.StringFlag{ 501 + Name: "server-host", 502 + Usage: "public host for this particular physical streamplace node. defaults to broadcaster-host and only must be set for multi-node broadcasters", 503 + Destination: &cli.ServerHost, 504 + Sources: urfavecli.EnvVars("SP_SERVER_HOST"), 505 + }, 506 + &urfavecli.BoolFlag{ 507 + Name: "thumbnail", 508 + Usage: "enable thumbnail generation", 509 + Value: true, 510 + Destination: &cli.Thumbnail, 511 + Sources: urfavecli.EnvVars("SP_THUMBNAIL"), 512 + }, 513 + &urfavecli.BoolFlag{ 514 + Name: "smear-audio", 515 + Usage: "enable audio smearing to create 'perfect' segment timestamps", 516 + Value: false, 517 + Destination: &cli.SmearAudio, 518 + Sources: urfavecli.EnvVars("SP_SMEAR_AUDIO"), 519 + }, 520 + &urfavecli.StringFlag{ 521 + Name: "tracing-endpoint", 522 + Usage: "gRPC endpoint to send traces to", 523 + Destination: &cli.TracingEndpoint, 524 + Sources: urfavecli.EnvVars("SP_TRACING_ENDPOINT"), 525 + }, 526 + &urfavecli.IntFlag{ 527 + Name: "rate-limit-per-second", 528 + Usage: "rate limit for requests per second per ip", 529 + Value: 0, 530 + Destination: &cli.RateLimitPerSecond, 531 + Sources: urfavecli.EnvVars("SP_RATE_LIMIT_PER_SECOND"), 532 + }, 533 + &urfavecli.IntFlag{ 534 + Name: "rate-limit-burst", 535 + Usage: "rate limit burst for requests per ip", 536 + Value: 0, 537 + Destination: &cli.RateLimitBurst, 538 + Sources: urfavecli.EnvVars("SP_RATE_LIMIT_BURST"), 539 + }, 540 + &urfavecli.IntFlag{ 541 + Name: "rate-limit-websocket", 542 + Usage: "number of concurrent websocket connections allowed per ip", 543 + Value: 10, 544 + Destination: &cli.RateLimitWebsocket, 545 + Sources: urfavecli.EnvVars("SP_RATE_LIMIT_WEBSOCKET"), 546 + }, 547 + &urfavecli.StringFlag{ 548 + Name: "rtmp-server-addon", 549 + Usage: "address of external RTMP server to forward streams to", 550 + Destination: &cli.RTMPServerAddon, 551 + Sources: urfavecli.EnvVars("SP_RTMP_SERVER_ADDON"), 552 + }, 553 + &urfavecli.StringFlag{ 554 + Name: "rtmps-addon-addr", 555 + Usage: "address to listen for RTMPS on the addon server", 556 + Value: ":1936", 557 + Destination: &cli.RTMPSAddonAddr, 558 + Sources: urfavecli.EnvVars("SP_RTMPS_ADDON_ADDR"), 559 + }, 560 + &urfavecli.StringFlag{ 561 + Name: "rtmps-addr", 562 + Usage: "address to listen for RTMPS connections (when --secure=true)", 563 + Value: ":1935", 564 + Destination: &cli.RTMPSAddr, 565 + Sources: urfavecli.EnvVars("SP_RTMPS_ADDR"), 566 + }, 567 + &urfavecli.StringFlag{ 568 + Name: "rtmp-addr", 569 + Usage: "address to listen for RTMP connections (when --secure=false)", 570 + Value: ":1935", 571 + Destination: &cli.RTMPAddr, 572 + Sources: urfavecli.EnvVars("SP_RTMP_ADDR"), 573 + }, 574 + &urfavecli.StringFlag{ 575 + Name: "discord-webhooks", 576 + Usage: `JSON array of Discord webhooks to send notifications to (default: "[]")`, 577 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 578 + if s == "" { 579 + return nil 580 + } 581 + return json.Unmarshal([]byte(s), &cli.DiscordWebhooks) 582 + }, 583 + Sources: urfavecli.EnvVars("SP_DISCORD_WEBHOOKS"), 584 + }, 585 + &urfavecli.BoolFlag{ 586 + Name: "new-webrtc-playback", 587 + Usage: "enable new webrtc playback", 588 + Value: true, 589 + Destination: &cli.NewWebRTCPlayback, 590 + Sources: urfavecli.EnvVars("SP_NEW_WEBRTC_PLAYBACK"), 591 + }, 592 + &urfavecli.StringFlag{ 593 + Name: "apple-team-id", 594 + Usage: "apple team id for deep linking", 595 + Destination: &cli.AppleTeamID, 596 + Sources: urfavecli.EnvVars("SP_APPLE_TEAM_ID"), 597 + }, 598 + &urfavecli.StringFlag{ 599 + Name: "android-cert-fingerprint", 600 + Usage: "android cert fingerprint for deep linking", 601 + Destination: &cli.AndroidCertFingerprint, 602 + Sources: urfavecli.EnvVars("SP_ANDROID_CERT_FINGERPRINT"), 603 + }, 604 + &urfavecli.StringFlag{ 605 + Name: "labelers", 606 + Usage: `did of labelers that this instance should subscribe to (default: "")`, 607 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 608 + if s == "" { 609 + return nil 610 + } 611 + cli.Labelers = strings.Split(s, ",") 612 + return nil 613 + }, 614 + Sources: urfavecli.EnvVars("SP_LABELERS"), 615 + }, 616 + &urfavecli.StringFlag{ 617 + Name: "atproto-did", 618 + Usage: "atproto did to respond to on /.well-known/atproto-did (default did:web:PUBLIC_HOST)", 619 + Destination: &cli.AtprotoDID, 620 + Sources: urfavecli.EnvVars("SP_ATPROTO_DID"), 621 + }, 622 + &urfavecli.StringFlag{ 623 + Name: "content-filters", 624 + Usage: `JSON content filtering rules (default: "{}")`, 625 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 626 + if s == "" { 627 + return nil 628 + } 629 + return json.Unmarshal([]byte(s), &cli.ContentFilters) 630 + }, 631 + Sources: urfavecli.EnvVars("SP_CONTENT_FILTERS"), 632 + }, 633 + &urfavecli.StringFlag{ 634 + Name: "default-recommended-streamers", 635 + Usage: `comma-separated list of streamer DIDs to recommend by default when no other recommendations are available (default: "")`, 636 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 637 + if s == "" { 638 + return nil 639 + } 640 + cli.DefaultRecommendedStreamers = strings.Split(s, ",") 641 + return nil 642 + }, 643 + Sources: urfavecli.EnvVars("SP_DEFAULT_RECOMMENDED_STREAMERS"), 644 + }, 645 + &urfavecli.BoolFlag{ 646 + Name: "livepeer-help", 647 + Usage: "print help for livepeer flags and exit", 648 + Value: false, 649 + Destination: &cli.LivepeerHelp, 650 + Sources: urfavecli.EnvVars("SP_LIVEPEER_HELP"), 651 + }, 652 + &urfavecli.StringFlag{ 653 + Name: "plc-url", 654 + Usage: "url of the plc directory", 655 + Value: "https://plc.directory", 656 + Destination: &cli.PLCURL, 657 + Sources: urfavecli.EnvVars("SP_PLC_URL"), 658 + }, 659 + &urfavecli.BoolFlag{ 660 + Name: "sql-logging", 661 + Usage: "enable sql logging", 662 + Value: false, 663 + Destination: &cli.SQLLogging, 664 + Sources: urfavecli.EnvVars("SP_SQL_LOGGING"), 665 + }, 666 + &urfavecli.StringFlag{ 667 + Name: "sentry-dsn", 668 + Usage: "sentry dsn for error reporting", 669 + Destination: &cli.SentryDSN, 670 + Sources: urfavecli.EnvVars("SP_SENTRY_DSN"), 671 + }, 672 + &urfavecli.BoolFlag{ 673 + Name: "livepeer-debug", 674 + Usage: "log livepeer segments to $SP_DATA_DIR/livepeer-debug", 675 + Value: false, 676 + Destination: &cli.LivepeerDebug, 677 + Sources: urfavecli.EnvVars("SP_LIVEPEER_DEBUG"), 678 + }, 679 + &urfavecli.StringFlag{ 680 + Name: "segment-debug-dir", 681 + Usage: "directory to log segment validation to", 682 + Destination: &cli.SegmentDebugDir, 683 + Sources: urfavecli.EnvVars("SP_SEGMENT_DEBUG_DIR"), 684 + }, 685 + &urfavecli.StringFlag{ 686 + Name: "tickets", 687 + Usage: `tickets to join the swarm with (default: "")`, 688 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 689 + if s == "" { 690 + return nil 691 + } 692 + cli.Tickets = strings.Split(s, ",") 693 + return nil 694 + }, 695 + Sources: urfavecli.EnvVars("SP_TICKETS"), 696 + }, 697 + &urfavecli.StringFlag{ 698 + Name: "iroh-topic", 699 + Usage: "topic to use for the iroh swarm (must be 32 bytes in hex)", 700 + Destination: &cli.IrohTopic, 701 + Sources: urfavecli.EnvVars("SP_IROH_TOPIC"), 702 + }, 703 + &urfavecli.BoolFlag{ 704 + Name: "disable-iroh-relay", 705 + Usage: "disable the iroh relay", 706 + Value: false, 707 + Destination: &cli.DisableIrohRelay, 708 + Sources: urfavecli.EnvVars("SP_DISABLE_IROH_RELAY"), 709 + }, 710 + &urfavecli.StringFlag{ 711 + Name: "dev-account-creds", 712 + Usage: `(FOR DEVELOPMENT ONLY) did=password pairs for logging into test accounts without oauth (default: "")`, 713 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 714 + if s == "" { 715 + return nil 716 + } 717 + cli.DevAccountCreds = map[string]string{} 718 + pairs := strings.Split(s, ",") 719 + for _, pair := range pairs { 720 + parts := strings.Split(pair, "=") 721 + if len(parts) != 2 { 722 + return fmt.Errorf("invalid kv flag: %s", pair) 723 + } 724 + cli.DevAccountCreds[parts[0]] = parts[1] 725 + } 726 + return nil 727 + }, 728 + Sources: urfavecli.EnvVars("SP_DEV_ACCOUNT_CREDS"), 729 + }, 730 + &urfavecli.DurationFlag{ 731 + Name: "stream-session-timeout", 732 + Usage: "how long to wait before considering a stream inactive on this node?", 733 + Value: 60 * time.Second, 734 + Destination: &cli.StreamSessionTimeout, 735 + Sources: urfavecli.EnvVars("SP_STREAM_SESSION_TIMEOUT"), 736 + }, 737 + &urfavecli.StringFlag{ 738 + Name: "replicators", 739 + Usage: "comma-separated list of replication protocols to use (websocket, iroh)", 740 + Value: ReplicatorWebsocket, 741 + Sources: urfavecli.EnvVars("SP_REPLICATORS"), 742 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 743 + if s != "" { 744 + cli.Replicators = strings.Split(s, ",") 745 + } 746 + return nil 747 + }, 748 + }, 749 + &urfavecli.StringFlag{ 750 + Name: "websocket-url", 751 + Usage: "override the websocket (ws:// or wss://) url to use for replication (normally not necessary, used for testing)", 752 + Destination: &cli.WebsocketURL, 753 + Sources: urfavecli.EnvVars("SP_WEBSOCKET_URL"), 754 + }, 755 + &urfavecli.BoolFlag{ 756 + Name: "behind-https-proxy", 757 + Usage: "set to true if this node is behind an https proxy and we should report https URLs even though the node isn't serving HTTPS", 758 + Value: false, 759 + Destination: &cli.BehindHTTPSProxy, 760 + Sources: urfavecli.EnvVars("SP_BEHIND_HTTPS_PROXY"), 761 + }, 762 + &urfavecli.StringFlag{ 763 + Name: "admin-dids", 764 + Usage: `comma-separated list of DIDs that are authorized to modify branding and other admin operations (default: "")`, 765 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 766 + if s == "" { 767 + return nil 768 + } 769 + cli.AdminDIDs = strings.Split(s, ",") 770 + return nil 771 + }, 772 + Sources: urfavecli.EnvVars("SP_ADMIN_DIDS"), 773 + }, 774 + &urfavecli.StringFlag{ 775 + Name: "syndicate", 776 + Usage: `list of DIDs that we should rebroadcast ('*' for everybody) (default: "")`, 777 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 778 + if s == "" { 779 + return nil 780 + } 781 + cli.Syndicate = strings.Split(s, ",") 782 + return nil 783 + }, 784 + Sources: urfavecli.EnvVars("SP_SYNDICATE"), 785 + }, 786 + &urfavecli.BoolFlag{ 787 + Name: "player-telemetry", 788 + Usage: "enable player telemetry", 789 + Value: true, 790 + Destination: &cli.PlayerTelemetry, 791 + Sources: urfavecli.EnvVars("SP_PLAYER_TELEMETRY"), 792 + }, 793 + &urfavecli.StringFlag{ 794 + Name: "local-db-url", 795 + Usage: "URL of the local database to use for storing local data", 796 + Value: "sqlite://$SP_DATA_DIR/localdb.sqlite", 797 + Destination: &cli.LocalDBURL, 798 + Sources: urfavecli.EnvVars("SP_LOCAL_DB_URL"), 799 + }, 800 + &urfavecli.BoolFlag{ 801 + Name: "external-signing", 802 + Usage: "DEPRECATED, does nothing.", 803 + Value: true, 804 + }, 805 + &urfavecli.BoolFlag{ 806 + Name: "insecure", 807 + Usage: "DEPRECATED, does nothing.", 808 + Value: false, 809 + }, 810 + }, 811 + Before: func(ctx context.Context, cmd *urfavecli.Command) (context.Context, error) { 812 + return ctx, cli.Validate(cmd) 813 + }, 814 + } 815 + 816 + // Add data dir flags 817 + cli.dataDirFlags = append(cli.dataDirFlags, &cli.DBURL) 818 cli.dataDirFlags = append(cli.dataDirFlags, &cli.LocalDBURL) 819 + cli.dataDirFlags = append(cli.dataDirFlags, &cli.TLSCertPath) 820 + cli.dataDirFlags = append(cli.dataDirFlags, &cli.TLSKeyPath) 821 + cli.dataDirFlags = append(cli.dataDirFlags, &cli.EthKeystorePath) 822 823 + if runtime.GOOS == "linux" { 824 + cmd.Flags = append(cmd.Flags, &urfavecli.BoolFlag{ 825 + Name: "no-mist", 826 + Usage: "Disable MistServer", 827 + Value: true, 828 + Destination: &cli.NoMist, 829 + Sources: urfavecli.EnvVars("SP_NO_MIST"), 830 + }) 831 + cmd.Flags = append(cmd.Flags, &urfavecli.IntFlag{ 832 + Name: "mist-admin-port", 833 + Usage: "MistServer admin port (internal use only)", 834 + Value: 14242, 835 + Destination: &cli.MistAdminPort, 836 + Sources: urfavecli.EnvVars("SP_MIST_ADMIN_PORT"), 837 + }) 838 + cmd.Flags = append(cmd.Flags, &urfavecli.IntFlag{ 839 + Name: "mist-rtmp-port", 840 + Usage: "MistServer RTMP port (internal use only)", 841 + Value: 11935, 842 + Destination: &cli.MistRTMPPort, 843 + Sources: urfavecli.EnvVars("SP_MIST_RTMP_PORT"), 844 + }) 845 + cmd.Flags = append(cmd.Flags, &urfavecli.IntFlag{ 846 + Name: "mist-http-port", 847 + Usage: "MistServer HTTP port (internal use only)", 848 + Value: 18080, 849 + Destination: &cli.MistHTTPPort, 850 + Sources: urfavecli.EnvVars("SP_MIST_HTTP_PORT"), 851 + }) 852 + } 853 854 + LivepeerFlagSet = flag.NewFlagSet("livepeer", flag.ContinueOnError) 855 + LivepeerConfig = starter.NewLivepeerConfig(LivepeerFlagSet) 856 + LivepeerFlagSet.VisitAll(func(f *flag.Flag) { 857 adapted := LivepeerFlags.CamelToSnake[f.Name] 858 + cmd.Flags = append(cmd.Flags, &urfavecli.StringFlag{ 859 + Name: fmt.Sprintf("livepeer.%s", adapted), 860 + Usage: f.Usage, 861 + Sources: urfavecli.EnvVars(fmt.Sprintf("SP_LIVEPEER_%s", adapted)), 862 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 863 + return LivepeerFlagSet.Set(f.Name, s) 864 + }, 865 + }) 866 }) 867 868 + return cmd 0 0 0 0 0 0 869 } 870 871 var StreamplaceSchemePrefix = "streamplace://" ··· 953 ) 954 } 955 956 + func (cli *CLI) Validate(cmd *urfavecli.Command) error { 0 0 0 0 0 0 0 957 if cli.DataDir == "" { 958 return fmt.Errorf("could not determine default data dir (no $HOME) and none provided, please set --data-dir") 959 } ··· 962 } 963 if cli.LivepeerGateway { 964 log.MonkeypatchStderr() 965 + // Livepeer gateway configuration will be handled in the caller 966 + cli.LivepeerGatewayURL = "http://127.0.0.1:8935" 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 967 } 968 for _, dest := range cli.dataDirFlags { 969 *dest = strings.Replace(*dest, SPDataDir, cli.DataDir, 1) ··· 992 return err 993 } 994 cli.FirebaseServiceAccount = string(bs) 995 + } 996 + // Set default replicator if none specified 997 + if len(cli.Replicators) == 0 { 998 + cli.Replicators = []string{ReplicatorWebsocket} 999 } 1000 return nil 1001 } ··· 1105 return nil 1106 } 1107 0 0 0 0 0 0 0 0 0 0 1108 func (cli *CLI) HasMist() bool { 1109 return runtime.GOOS == "linux" 1110 } 1111 1112 // type for comma-separated ethereum addresses 1113 + func (cli *CLI) AddressSliceFlag(name, defaultValue, usage string, dest *[]aqpub.Pub) urfavecli.Flag { 1114 *dest = []aqpub.Pub{} 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 1115 usage = fmt.Sprintf(`%s (default: "%s")`, usage, defaultValue) 0 0 0 0 0 0 0 1116 1117 + return &urfavecli.StringFlag{ 1118 + Name: name, 1119 + Usage: usage, 1120 + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { 1121 + if s == "" { 1122 + return nil 0 0 0 0 0 0 1123 } 1124 + strs := strings.Split(s, ",") 1125 + for _, str := range strs { 1126 + pub, err := aqpub.FromHexString(str) 1127 + if err != nil { 1128 + return err 1129 + } 1130 + *dest = append(*dest, pub) 1131 } 1132 + return nil 1133 + }, 1134 + Sources: urfavecli.EnvVars(fmt.Sprintf("SP_%s", strings.ToUpper(strings.ReplaceAll(name, "-", "_")))), 1135 + } 0 0 0 0 0 0 0 0 0 1136 } 1137 1138 func (cli *CLI) StreamIsAllowed(did string) error { ··· 1146 if openServer && !isDIDKey { 1147 return nil 1148 } 1149 + if slices.Contains(cli.AllowedStreams, did) { 1150 + return nil 0 0 1151 } 1152 return fmt.Errorf("user is not allowed to stream") 1153 }

+5 -3

pkg/media/segment_roundtrip_test.go

··· 87 require.NoError(t, err) 88 89 signedSplitSegDir := makeTestSubdir(t, tempDir, "signed-split-segments") 90 - cli := &config.CLI{} 91 - fs := cli.NewFlagSet("rtcrec-test") 92 - err = cli.Parse(fs, []string{}) 0 0 93 require.NoError(t, err) 94 err = SplitSegments(context.Background(), cli, rws, func(fname string) ReadWriteSeekCloser { 95 fd, err := os.Create(filepath.Join(signedSplitSegDir, fname))

··· 87 require.NoError(t, err) 88 89 signedSplitSegDir := makeTestSubdir(t, tempDir, "signed-split-segments") 90 + cli := &config.CLI{ 91 + DataDir: tempDir, // Set data dir for test 92 + } 93 + cmd := cli.NewCommand("rtcrec-test") 94 + err = cli.Validate(cmd) 95 require.NoError(t, err) 96 err = SplitSegments(context.Background(), cli, rws, func(fname string) ReadWriteSeekCloser { 97 fd, err := os.Create(filepath.Join(signedSplitSegDir, fname))