stream.place
Live video on the AT Protocol
1package media
2
3import (
4 "bytes"
5 "context"
6 "fmt"
7 "io"
8 "os"
9 "testing"
10
11 "github.com/go-gst/go-gst/gst"
12 "github.com/go-gst/go-gst/gst/app"
13 "github.com/stretchr/testify/require"
14 "golang.org/x/sync/errgroup"
15 "stream.place/streamplace/pkg/log"
16)
17
18func TestWriterNewSample(t *testing.T) {
19 withNoGSTLeaks(t, func() {
20 filePath := getFixture("5sec.mp4")
21 fileInfo, err := os.Stat(filePath)
22 require.NoError(t, err)
23 fileSize := fileInfo.Size()
24 t.Logf("Test file size: %d bytes", fileSize)
25 g, ctx := errgroup.WithContext(context.Background())
26 // ctx = log.WithDebugValue(ctx, map[string]map[string]int{"func": {"TestWriterNewSample": 9}})
27 for i := 0; i < streamplaceTestCount; i++ {
28 g.Go(func() error {
29 bs := bytes.Buffer{}
30 err := writerNewSampleInner(ctx, i, &bs)
31 if err != nil {
32 return err
33 }
34 if bs.Len() != int(fileSize) {
35 return fmt.Errorf("expected %d bytes, got %d", fileSize, bs.Len())
36 }
37 return nil
38 })
39 }
40 err = g.Wait()
41 require.NoError(t, err)
42 })
43}
44
45func writerNewSampleInner(ctx context.Context, i int, w io.Writer) error {
46 ctx = log.WithLogValues(ctx, "func", "TestWriterNewSample")
47 ctx, cancel := context.WithCancel(ctx)
48 defer cancel()
49
50 pipeline, err := gst.NewPipeline(fmt.Sprintf("TestWriterNewSample-%d", i))
51 if err != nil {
52 return err
53 }
54
55 fileSrc, err := gst.NewElementWithProperties("filesrc", map[string]any{
56 "location": getFixture("5sec.mp4"),
57 })
58 if err != nil {
59 return err
60 }
61 err = pipeline.Add(fileSrc)
62 if err != nil {
63 return err
64 }
65
66 var busErr error
67 go func() {
68 busErr = HandleBusMessages(ctx, pipeline)
69 cancel()
70 }()
71
72 appSink, err := gst.NewElementWithProperties("appsink", map[string]any{
73 "name": fmt.Sprintf("TestWriterNewSample-appsink-%d", i),
74 "sync": false,
75 })
76 if err != nil {
77 return err
78 }
79 err = pipeline.Add(appSink)
80 if err != nil {
81 return err
82 }
83
84 sink := app.SinkFromElement(appSink)
85 sink.SetCallbacks(&app.SinkCallbacks{
86 NewSampleFunc: WriterNewSample(ctx, w),
87 })
88
89 err = fileSrc.Link(appSink)
90 if err != nil {
91 return err
92 }
93
94 err = pipeline.SetState(gst.StatePlaying)
95 if err != nil {
96 return err
97 }
98
99 <-ctx.Done()
100
101 err = pipeline.SetState(gst.StateNull)
102 if err != nil {
103 return err
104 }
105
106 return busErr
107}