Live video on the AT Protocol
stream.place
1package multitest
2
3import (
4 "bufio"
5 "context"
6 "fmt"
7 "net/http"
8 "os"
9 "os/exec"
10 "path/filepath"
11 "runtime"
12 "strings"
13 "testing"
14 "time"
15
16 comatproto "github.com/bluesky-social/indigo/api/atproto"
17 lexutil "github.com/bluesky-social/indigo/lex/util"
18 "github.com/bluesky-social/indigo/util"
19 scraper "github.com/starttoaster/prometheus-exporter-scraper"
20 "github.com/stretchr/testify/require"
21 "golang.org/x/sync/errgroup"
22 "stream.place/streamplace/pkg/cmd"
23 "stream.place/streamplace/pkg/crypto/spkey"
24 "stream.place/streamplace/pkg/devenv"
25 "stream.place/streamplace/pkg/gstinit"
26 "stream.place/streamplace/pkg/log"
27 "stream.place/streamplace/pkg/streamplace"
28 "stream.place/streamplace/test/remote"
29)
30
31func TestMultinodeSyndication(t *testing.T) {
32 if os.Getenv("GITHUB_ACTION") != "" {
33 t.Skip("Skipping multitest in GitHub Actions")
34 }
35 gstinit.InitGST()
36 dev := devenv.WithDevEnv(t)
37 acct1 := dev.CreateAccount(t)
38 acct2 := dev.CreateAccount(t)
39 ctx, cancel := context.WithCancel(context.Background())
40 defer cancel()
41 node1 := startStreamplaceNode(ctx, "node1", t, dev)
42 node2 := startStreamplaceNode(ctx, "node2", t, dev)
43 node3 := startStreamplaceNode(ctx, "node3", t, dev)
44 node1.StartStream(t, acct1)
45 node2.PlayStream(t, acct1)
46 node3.PlayStream(t, acct1)
47 <-time.After(10 * time.Second)
48 node2.Shutdown(t)
49 <-time.After(20 * time.Second)
50 node4 := startStreamplaceNode(ctx, "node4", t, dev)
51 node4.StartStream(t, acct2)
52 node4.PlayStream(t, acct1)
53 node1.PlayStream(t, acct2)
54 node3.PlayStream(t, acct2)
55 <-time.After(30 * time.Second)
56}
57
58func TestOriginSwap(t *testing.T) {
59 if os.Getenv("GITHUB_ACTION") != "" {
60 t.Skip("Skipping multitest in GitHub Actions")
61 }
62 gstinit.InitGST()
63 ctx, cancel := context.WithCancel(context.Background())
64 defer cancel()
65 dev := devenv.WithDevEnv(t)
66 acct1 := dev.CreateAccount(t)
67 acct2 := dev.CreateAccount(t)
68 node1 := startStreamplaceNode(ctx, "node1", t, dev)
69 node2 := startStreamplaceNode(ctx, "node2", t, dev)
70 node3 := startStreamplaceNode(ctx, "node3", t, dev)
71 // node4 := startStreamplaceNode(ctx, "node4", t, dev)
72 node1.StartStream(t, acct1)
73 node2.StartStream(t, acct2)
74 node1.PlayStream(t, acct1)
75 node2.PlayStream(t, acct1)
76 node3.PlayStream(t, acct1)
77 node1.PlayStream(t, acct2)
78 node2.PlayStream(t, acct2)
79 node3.PlayStream(t, acct2)
80 // node4.PlayStream(t, acct1)
81 <-time.After(30 * time.Second)
82 // node1.StopStream(t, acct1)
83 // node2.StartStream(t, acct1)
84 // <-time.After(20 * time.Second)
85 // // node2.StopStream(t, acct1)
86 // // node3.StartStream(t, acct1)
87 // // node4.Shutdown(t)
88 // // <-time.After(10 * time.Second)
89}
90
91var currentPort = 10000
92
93func nextPort() int {
94 currentPort++
95 return currentPort
96}
97
98type TestNode struct {
99 Env map[string]string
100 Dev *devenv.DevEnv
101 Cmd *exec.Cmd
102 Ctx context.Context // don't ever do this, it's just a test
103 Shutdown func(t *testing.T)
104 ActiveStreams map[string]context.CancelFunc
105 Name string
106}
107
108func startStreamplaceNode(ctx context.Context, name string, t *testing.T, dev *devenv.DevEnv) *TestNode {
109 nodeCtx, nodeCancel := context.WithCancel(ctx)
110 dataDir := t.TempDir()
111 devAccountCreds := []string{}
112 for _, acct := range dev.Accounts {
113 devAccountCreds = append(devAccountCreds, fmt.Sprintf("%s=%s", acct.DID, acct.Password))
114 }
115 apiPort := nextPort()
116 env := map[string]string{
117 "SP_HTTP_ADDR": fmt.Sprintf("127.0.0.1:%d", apiPort),
118 "SP_HTTP_INTERNAL_ADDR": fmt.Sprintf("127.0.0.1:%d", nextPort()),
119 "SP_RELAY_HOST": strings.ReplaceAll(dev.PDSURL, "http://", "ws://"),
120 "SP_PLC_URL": dev.PLCURL,
121 "SP_DATA_DIR": dataDir,
122 "SP_DEV_ACCOUNT_CREDS": strings.Join(devAccountCreds, ","),
123 "SP_STREAM_SESSION_TIMEOUT": "3s",
124 "SP_COLOR": "true",
125 "RUST_LOG": os.Getenv("RUST_LOG"),
126 "SP_BROADCASTER_HOST": fmt.Sprintf("%s.example.com", name),
127 "SP_WEBSOCKET_URL": fmt.Sprintf("ws://127.0.0.1:%d", apiPort),
128 }
129 _, file, _, _ := runtime.Caller(0)
130 buildDir := fmt.Sprintf("build-%s-%s", runtime.GOOS, runtime.GOARCH)
131 abs, err := filepath.Abs(filepath.Join(filepath.Dir(file), "..", "..", buildDir, "streamplace"))
132 require.NoErrorf(t, err, "[%s] failed to resolve absolute binary path", name)
133 // Run the streamplace binary at abs with the environment env
134 cmd := exec.Command(abs)
135 cmd.Env = []string{}
136 for k, v := range env {
137 cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", k, v))
138 }
139
140 stdoutPipe, err := cmd.StdoutPipe()
141 require.NoErrorf(t, err, "[%s] failed to get stdout pipe", name)
142 stderrPipe, err := cmd.StderrPipe()
143 require.NoErrorf(t, err, "[%s] failed to get stderr pipe", name)
144
145 stdoutDone := make(chan struct{})
146 stderrDone := make(chan struct{})
147
148 // Goroutine to read stdout and prefix lines
149 go func() {
150 defer close(stdoutDone)
151 scanner := bufio.NewScanner(stdoutPipe)
152 for scanner.Scan() {
153 fmt.Fprintf(os.Stdout, "[%s STDOUT] %s\n", name, scanner.Text())
154 }
155 if err := scanner.Err(); err != nil {
156 fmt.Fprintf(os.Stdout, "[%s STDOUT] Error reading stdout: %v\n", name, err)
157 }
158 }()
159 // Goroutine to read stderr and prefix lines
160 go func() {
161 defer close(stderrDone)
162 scanner := bufio.NewScanner(stderrPipe)
163 for scanner.Scan() {
164 fmt.Fprintf(os.Stdout, "[%s STDERR] %s\n", name, scanner.Text())
165 }
166 if err := scanner.Err(); err != nil {
167 fmt.Fprintf(os.Stdout, "[%s STDERR] Error reading stderr: %v\n", name, err)
168 }
169 }()
170
171 err = cmd.Start()
172 require.NoErrorf(t, err, "[%s] failed to start streamplace process", name)
173
174 // Wait for the streamplace node to be ready by polling the health endpoint
175 healthz := fmt.Sprintf("http://%s/api/healthz", env["SP_HTTP_ADDR"])
176 client := &http.Client{Timeout: 2 * time.Second}
177 for {
178 resp, err := client.Get(healthz)
179 if err == nil {
180 defer resp.Body.Close()
181 if resp.StatusCode == 200 {
182 break
183 }
184 }
185 time.Sleep(200 * time.Millisecond)
186 }
187 node := &TestNode{
188 Env: env,
189 Dev: dev,
190 Cmd: cmd,
191 Ctx: nodeCtx,
192 ActiveStreams: make(map[string]context.CancelFunc),
193 Name: name,
194 }
195 go func() {
196 <-nodeCtx.Done()
197 node.Shutdown(t)
198 }()
199 go func() {
200 for {
201 select {
202 case <-nodeCtx.Done():
203 return
204 case <-time.After(1 * time.Second):
205 scrp, err := scraper.NewWebScraper(fmt.Sprintf("http://%s/metrics", env["SP_HTTP_INTERNAL_ADDR"]))
206 require.NoErrorf(t, err, "[%s] failed to create scraper", name)
207 data, err := scrp.ScrapeWeb()
208 require.NoErrorf(t, err, "[%s] failed to scrape metrics", name)
209 found := false
210 for _, metric := range data.Gauges {
211 if metric.Key == "streamplace_send_segment_calls" {
212 // require.Lessf(t, metric.Value, float64(2), "[%s] send segment calls should be < 2, got %f", name, metric.Value)
213 log.Log(nodeCtx, fmt.Sprintf("[%s] open send_segment calls", name), "open", metric.Value)
214 found = true
215 break
216 }
217 }
218 if !found {
219 require.Fail(t, fmt.Sprintf("[%s] send segment calls metric not found", name))
220 }
221 }
222 }
223 }()
224 shuttingDown := false
225 nodeShutdown := func(t *testing.T) {
226 if shuttingDown {
227 return
228 }
229 shuttingDown = true
230 nodeCancel()
231 _ = cmd.Process.Kill()
232 _, _ = cmd.Process.Wait()
233 // Wait for stdout/stderr readers to finish
234 <-stdoutDone
235 <-stderrDone
236 }
237 node.Shutdown = nodeShutdown
238 t.Cleanup(func() {
239 node.Shutdown(t)
240 })
241 return node
242}
243
244func (node *TestNode) StartStream(t *testing.T, acct *devenv.DevEnvAccount) {
245 streamCtx, streamCancel := context.WithCancel(node.Ctx)
246 node.ActiveStreams[acct.DID] = streamCancel
247 priv, pub, err := spkey.GenerateStreamKeyForDID(acct.DID)
248 require.NoErrorf(t, err, "[%s] failed to generate stream key for DID %s", node.Name, acct.DID)
249 createdBy := "multitest"
250 streamKey := streamplace.Key{
251 SigningKey: pub.DIDKey(),
252 CreatedAt: time.Now().Format(util.ISO8601),
253 CreatedBy: &createdBy,
254 }
255 _, err = comatproto.RepoCreateRecord(context.TODO(), acct.XRPC, &comatproto.RepoCreateRecord_Input{
256 Collection: "place.stream.key",
257 Repo: acct.DID,
258 Record: &lexutil.LexiconTypeDecoder{Val: &streamKey},
259 })
260 require.NoErrorf(t, err, "[%s] failed to create Repo record for DID %s", node.Name, acct.DID)
261 log.Log(context.Background(), "created stream key", "did", acct.DID, "pub", pub.DIDKey())
262 time.Sleep(1 * time.Second)
263 whip := &cmd.WHIPClient{
264 StreamKey: priv,
265 File: remote.RemoteFixture("3188c071b354f2e548d7f2d332699758e8e3ab1600280e5b07cb67eedc64f274/BigBuckBunny_1sGOP_240p30_NoBframes.mp4"),
266 Endpoint: fmt.Sprintf("http://%s", node.Env["SP_HTTP_ADDR"]),
267 Count: 1,
268 }
269
270 g, ctx := errgroup.WithContext(streamCtx)
271 g.Go(func() error {
272 return whip.WHIP(ctx)
273 })
274}
275
276func (node *TestNode) StopStream(t *testing.T, acct *devenv.DevEnvAccount) {
277 cancel := node.ActiveStreams[acct.DID]
278 if cancel == nil {
279 require.FailNow(t, fmt.Sprintf("[%s] stream not active for did %s", node.Name, acct.DID))
280 }
281 cancel()
282 delete(node.ActiveStreams, acct.DID)
283}
284
285func (node *TestNode) PlayStream(t *testing.T, acct *devenv.DevEnvAccount) {
286 whep := &cmd.WHEPClient{
287 Endpoint: fmt.Sprintf("http://%s/api/playback/%s/webrtc", node.Env["SP_HTTP_ADDR"], acct.DID),
288 Count: 1,
289 }
290 g, ctx := errgroup.WithContext(node.Ctx)
291 g.Go(func() error {
292 return whep.WHEP(ctx)
293 })
294 start := time.Now()
295 // start at -1 to give them an extra go-round to boot
296 prevVideoTotal := -1
297 prevAudioTotal := -1
298 g.Go(func() error {
299 for {
300 select {
301 case <-ctx.Done():
302 return ctx.Err()
303 case <-time.After(5 * time.Second):
304 stats := whep.Stats[0]
305 videoStats := stats["video"]
306 audioStats := stats["audio"]
307 if videoStats.Total == prevVideoTotal || audioStats.Total == prevAudioTotal {
308 require.FailNow(t, fmt.Sprintf("[%s] stream playback stalled did=%s, video=%d, audio=%d, elapsed=%s", node.Name, acct.DID, videoStats.Total, audioStats.Total, time.Since(start)))
309 }
310 prevVideoTotal = videoStats.Total
311 prevAudioTotal = audioStats.Total
312 log.Log(ctx, fmt.Sprintf("[%s] stream playback", node.Name), "did", acct.DID, "video", videoStats.Total, "audio", audioStats.Total, "elapsed", time.Since(start))
313 }
314 }
315 })
316}