stream.place
Live video on the AT Protocol
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 env := map[string]string{
116 "SP_HTTP_ADDR": fmt.Sprintf("127.0.0.1:%d", nextPort()),
117 "SP_HTTP_INTERNAL_ADDR": fmt.Sprintf("127.0.0.1:%d", nextPort()),
118 "SP_RELAY_HOST": strings.ReplaceAll(dev.PDSURL, "http://", "ws://"),
119 "SP_PLC_URL": dev.PLCURL,
120 "SP_DATA_DIR": dataDir,
121 "SP_DEV_ACCOUNT_CREDS": strings.Join(devAccountCreds, ","),
122 "SP_STREAM_SESSION_TIMEOUT": "3s",
123 "SP_COLOR": "true",
124 "RUST_LOG": os.Getenv("RUST_LOG"),
125 }
126 _, file, _, _ := runtime.Caller(0)
127 buildDir := fmt.Sprintf("build-%s-%s", runtime.GOOS, runtime.GOARCH)
128 abs, err := filepath.Abs(filepath.Join(filepath.Dir(file), "..", "..", buildDir, "streamplace"))
129 require.NoErrorf(t, err, "[%s] failed to resolve absolute binary path", name)
130 // Run the streamplace binary at abs with the environment env
131 cmd := exec.Command(abs)
132 cmd.Env = []string{}
133 for k, v := range env {
134 cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", k, v))
135 }
136
137 stdoutPipe, err := cmd.StdoutPipe()
138 require.NoErrorf(t, err, "[%s] failed to get stdout pipe", name)
139 stderrPipe, err := cmd.StderrPipe()
140 require.NoErrorf(t, err, "[%s] failed to get stderr pipe", name)
141
142 stdoutDone := make(chan struct{})
143 stderrDone := make(chan struct{})
144
145 // Goroutine to read stdout and prefix lines
146 go func() {
147 defer close(stdoutDone)
148 scanner := bufio.NewScanner(stdoutPipe)
149 for scanner.Scan() {
150 fmt.Fprintf(os.Stdout, "[%s STDOUT] %s\n", name, scanner.Text())
151 }
152 if err := scanner.Err(); err != nil {
153 fmt.Fprintf(os.Stdout, "[%s STDOUT] Error reading stdout: %v\n", name, err)
154 }
155 }()
156 // Goroutine to read stderr and prefix lines
157 go func() {
158 defer close(stderrDone)
159 scanner := bufio.NewScanner(stderrPipe)
160 for scanner.Scan() {
161 fmt.Fprintf(os.Stdout, "[%s STDERR] %s\n", name, scanner.Text())
162 }
163 if err := scanner.Err(); err != nil {
164 fmt.Fprintf(os.Stdout, "[%s STDERR] Error reading stderr: %v\n", name, err)
165 }
166 }()
167
168 err = cmd.Start()
169 require.NoErrorf(t, err, "[%s] failed to start streamplace process", name)
170
171 // Wait for the streamplace node to be ready by polling the health endpoint
172 healthz := fmt.Sprintf("http://%s/api/healthz", env["SP_HTTP_ADDR"])
173 client := &http.Client{Timeout: 2 * time.Second}
174 for {
175 resp, err := client.Get(healthz)
176 if err == nil {
177 defer resp.Body.Close()
178 if resp.StatusCode == 200 {
179 break
180 }
181 }
182 time.Sleep(200 * time.Millisecond)
183 }
184 node := &TestNode{
185 Env: env,
186 Dev: dev,
187 Cmd: cmd,
188 Ctx: nodeCtx,
189 ActiveStreams: make(map[string]context.CancelFunc),
190 Name: name,
191 }
192 go func() {
193 <-nodeCtx.Done()
194 node.Shutdown(t)
195 }()
196 go func() {
197 for {
198 select {
199 case <-nodeCtx.Done():
200 return
201 case <-time.After(1 * time.Second):
202 scrp, err := scraper.NewWebScraper(fmt.Sprintf("http://%s/metrics", env["SP_HTTP_INTERNAL_ADDR"]))
203 require.NoErrorf(t, err, "[%s] failed to create scraper", name)
204 data, err := scrp.ScrapeWeb()
205 require.NoErrorf(t, err, "[%s] failed to scrape metrics", name)
206 found := false
207 for _, metric := range data.Gauges {
208 if metric.Key == "streamplace_send_segment_calls" {
209 // require.Lessf(t, metric.Value, float64(2), "[%s] send segment calls should be < 2, got %f", name, metric.Value)
210 log.Log(nodeCtx, fmt.Sprintf("[%s] open send_segment calls", name), "open", metric.Value)
211 found = true
212 break
213 }
214 }
215 if !found {
216 require.Fail(t, fmt.Sprintf("[%s] send segment calls metric not found", name))
217 }
218 }
219 }
220 }()
221 shuttingDown := false
222 nodeShutdown := func(t *testing.T) {
223 if shuttingDown {
224 return
225 }
226 shuttingDown = true
227 nodeCancel()
228 _ = cmd.Process.Kill()
229 _, _ = cmd.Process.Wait()
230 // Wait for stdout/stderr readers to finish
231 <-stdoutDone
232 <-stderrDone
233 }
234 node.Shutdown = nodeShutdown
235 t.Cleanup(func() {
236 node.Shutdown(t)
237 })
238 return node
239}
240
241func (node *TestNode) StartStream(t *testing.T, acct *devenv.DevEnvAccount) {
242 streamCtx, streamCancel := context.WithCancel(node.Ctx)
243 node.ActiveStreams[acct.DID] = streamCancel
244 priv, pub, err := spkey.GenerateStreamKeyForDID(acct.DID)
245 require.NoErrorf(t, err, "[%s] failed to generate stream key for DID %s", node.Name, acct.DID)
246 createdBy := "multitest"
247 streamKey := streamplace.Key{
248 SigningKey: pub.DIDKey(),
249 CreatedAt: time.Now().Format(util.ISO8601),
250 CreatedBy: &createdBy,
251 }
252 _, err = comatproto.RepoCreateRecord(context.TODO(), acct.XRPC, &comatproto.RepoCreateRecord_Input{
253 Collection: "place.stream.key",
254 Repo: acct.DID,
255 Record: &lexutil.LexiconTypeDecoder{Val: &streamKey},
256 })
257 require.NoErrorf(t, err, "[%s] failed to create Repo record for DID %s", node.Name, acct.DID)
258 log.Log(context.Background(), "created stream key", "did", acct.DID, "pub", pub.DIDKey())
259 time.Sleep(1 * time.Second)
260 whip := &cmd.WHIPClient{
261 StreamKey: priv,
262 File: remote.RemoteFixture("3188c071b354f2e548d7f2d332699758e8e3ab1600280e5b07cb67eedc64f274/BigBuckBunny_1sGOP_240p30_NoBframes.mp4"),
263 Endpoint: fmt.Sprintf("http://%s", node.Env["SP_HTTP_ADDR"]),
264 Count: 1,
265 }
266
267 g, ctx := errgroup.WithContext(streamCtx)
268 g.Go(func() error {
269 return whip.WHIP(ctx)
270 })
271}
272
273func (node *TestNode) StopStream(t *testing.T, acct *devenv.DevEnvAccount) {
274 cancel := node.ActiveStreams[acct.DID]
275 if cancel == nil {
276 require.FailNow(t, fmt.Sprintf("[%s] stream not active for did %s", node.Name, acct.DID))
277 }
278 cancel()
279 delete(node.ActiveStreams, acct.DID)
280}
281
282func (node *TestNode) PlayStream(t *testing.T, acct *devenv.DevEnvAccount) {
283 whep := &cmd.WHEPClient{
284 Endpoint: fmt.Sprintf("http://%s/api/playback/%s/webrtc", node.Env["SP_HTTP_ADDR"], acct.DID),
285 Count: 1,
286 }
287 g, ctx := errgroup.WithContext(node.Ctx)
288 g.Go(func() error {
289 return whep.WHEP(ctx)
290 })
291 start := time.Now()
292 // start at -1 to give them an extra go-round to boot
293 prevVideoTotal := -1
294 prevAudioTotal := -1
295 g.Go(func() error {
296 for {
297 select {
298 case <-ctx.Done():
299 return ctx.Err()
300 case <-time.After(5 * time.Second):
301 stats := whep.Stats[0]
302 videoStats := stats["video"]
303 audioStats := stats["audio"]
304 if videoStats.Total == prevVideoTotal || audioStats.Total == prevAudioTotal {
305 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)))
306 }
307 prevVideoTotal = videoStats.Total
308 prevAudioTotal = audioStats.Total
309 log.Log(ctx, fmt.Sprintf("[%s] stream playback", node.Name), "did", acct.DID, "video", videoStats.Total, "audio", audioStats.Total, "elapsed", time.Since(start))
310 }
311 }
312 })
313}