diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index c530a26bc..7b2996027 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -156,11 +156,10 @@ test: image: "$DOCKER_REPO:builder-$DOCKERFILE_HASH" timeout: 2 hours variables: - STREAMPLACE_IGNORE_LEAKS: "true" STREAMPLACE_TEST_COUNT: "1" script: - git fetch --unshallow || echo 'already unshallow' - - make ci-test -j$(nproc) && make ci-lexicons + - make dev-setup && make dev-test && make ci-lexicons build-docker-mistserver: stage: build @@ -284,7 +283,6 @@ build-android-release: tags: - tart-installed variables: - STREAMPLACE_IGNORE_LEAKS: "true" STREAMPLACE_TEST_COUNT: "1" NODE_OPTIONS: "--max_old_space_size=4096" timeout: 2 hours diff --git a/Makefile b/Makefile index e8b76c172..3d8436787 100644 --- a/Makefile +++ b/Makefile @@ -91,7 +91,7 @@ dev-test: PKG_CONFIG_PATH=$(SHARED_PKG_CONFIG_PATH) \ LD_LIBRARY_PATH=$(SHARED_LD_LIBRARY_PATH) \ DYLD_LIBRARY_PATH=$(SHARED_DYLD_LIBRARY_PATH) \ - go test -p 1 -timeout 120s ./... + go test -p 1 -timeout 300s ./... .PHONY: schema schema: diff --git a/meson.build b/meson.build index 18e6da51b..a9d5f7665 100644 --- a/meson.build +++ b/meson.build @@ -305,7 +305,7 @@ if not meson.is_cross_build() args: [ 'test', '-p', '1', - '-timeout', '120s', + '-timeout', '300s', '@0@/pkg/...'.format(meson.current_source_dir()), '@0@/cmd/...'.format(meson.current_source_dir()), ], diff --git a/pkg/media/audio_smear_test.go b/pkg/media/audio_smear_test.go index f5379203a..b7546bc1c 100644 --- a/pkg/media/audio_smear_test.go +++ b/pkg/media/audio_smear_test.go @@ -11,26 +11,21 @@ import ( "github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/pbutils" "github.com/stretchr/testify/require" - "go.uber.org/goleak" "golang.org/x/sync/errgroup" - "stream.place/streamplace/pkg/gstinit" ) func TestAudioSmear(t *testing.T) { - gstinit.InitGST() - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) - - g, _ := errgroup.WithContext(context.Background()) - for i := 0; i < streamplaceTestCount; i++ { - g.Go(func() error { - return testAudioSmearInner(t) - }) - } - err := g.Wait() - require.NoError(t, err) + withNoGSTLeaks(t, func() { + + g, _ := errgroup.WithContext(context.Background()) + for i := 0; i < streamplaceTestCount; i++ { + g.Go(func() error { + return testAudioSmearInner(t) + }) + } + err := g.Wait() + require.NoError(t, err) + }) } func testAudioSmearInner(t *testing.T) error { diff --git a/pkg/media/bus_handler_test.go b/pkg/media/bus_handler_test.go index d5b62f5cb..7d571db8f 100644 --- a/pkg/media/bus_handler_test.go +++ b/pkg/media/bus_handler_test.go @@ -8,30 +8,27 @@ import ( "github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "github.com/stretchr/testify/require" - "go.uber.org/goleak" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/log" ) func TestBusHandlerCleanup(t *testing.T) { - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) + withNoGSTLeaks(t, func() { - g, ctx := errgroup.WithContext(context.Background()) - ctx = log.WithDebugValue(ctx, map[string]map[string]int{"func": {"TestBusHandler": 9}}) - for i := range streamplaceTestCount { - g.Go(func() error { - err := testBusHandlerCleanupInner(ctx, i) - if err == nil { - return fmt.Errorf("expected error") - } - return nil - }) - } - err := g.Wait() - require.NoError(t, err) + g, ctx := errgroup.WithContext(context.Background()) + ctx = log.WithDebugValue(ctx, map[string]map[string]int{"func": {"TestBusHandler": 9}}) + for i := range streamplaceTestCount { + g.Go(func() error { + err := testBusHandlerCleanupInner(ctx, i) + if err == nil { + return fmt.Errorf("expected error") + } + return nil + }) + } + err := g.Wait() + require.NoError(t, err) + }) } func testBusHandlerCleanupInner(ctx context.Context, i int) error { diff --git a/pkg/media/concat2_test.go b/pkg/media/concat2_test.go index ee7fa9374..6cf643998 100644 --- a/pkg/media/concat2_test.go +++ b/pkg/media/concat2_test.go @@ -13,28 +13,23 @@ import ( "github.com/go-gst/go-gst/gst/app" "github.com/google/uuid" "github.com/stretchr/testify/require" - "go.uber.org/goleak" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/bus" - "stream.place/streamplace/pkg/gstinit" "stream.place/streamplace/pkg/log" ) func TestConcatBin(t *testing.T) { - gstinit.InitGST() - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) - - g, _ := errgroup.WithContext(context.Background()) - for range streamplaceTestCount { - g.Go(func() error { - return innerTestConcatBin(t) - }) - } - err := g.Wait() - require.NoError(t, err) + withNoGSTLeaks(t, func() { + + g, _ := errgroup.WithContext(context.Background()) + for range streamplaceTestCount { + g.Go(func() error { + return innerTestConcatBin(t) + }) + } + err := g.Wait() + require.NoError(t, err) + }) } // This function remains in scope for the duration of a single users' playback diff --git a/pkg/media/concat_demux_test.go b/pkg/media/concat_demux_test.go index 4a3c0eb3c..d8a1bc91c 100644 --- a/pkg/media/concat_demux_test.go +++ b/pkg/media/concat_demux_test.go @@ -11,28 +11,22 @@ import ( "github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "github.com/stretchr/testify/require" - "go.uber.org/goleak" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/bus" - "stream.place/streamplace/pkg/gstinit" "stream.place/streamplace/pkg/log" ) func TestConcatDemuxBin(t *testing.T) { - gstinit.InitGST() - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) - - g, _ := errgroup.WithContext(context.Background()) - for range streamplaceTestCount { - g.Go(func() error { - return innerTestConcatDemuxBin(t) - }) - } - err := g.Wait() - require.NoError(t, err) + withNoGSTLeaks(t, func() { + g, _ := errgroup.WithContext(context.Background()) + for range streamplaceTestCount { + g.Go(func() error { + return innerTestConcatDemuxBin(t) + }) + } + err := g.Wait() + require.NoError(t, err) + }) } // This function remains in scope for the duration of a single users' playback diff --git a/pkg/media/io_helpers_test.go b/pkg/media/io_helpers_test.go index 90628b040..027fda13c 100644 --- a/pkg/media/io_helpers_test.go +++ b/pkg/media/io_helpers_test.go @@ -6,57 +6,40 @@ import ( "fmt" "io" "os" - "strconv" "testing" "github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "github.com/stretchr/testify/require" - "go.uber.org/goleak" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/log" ) -var streamplaceTestCount = 50 - -func init() { - testRunsStr := os.Getenv("STREAMPLACE_TEST_COUNT") - if testRunsStr != "" { - var err error - streamplaceTestCount, err = strconv.Atoi(testRunsStr) - if err != nil { - panic(fmt.Sprintf("STREAMPLACE_TEST_COUNT is not a number: %s", testRunsStr)) - } - } -} - func TestWriterNewSample(t *testing.T) { - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) - filePath := getFixture("5sec.mp4") - fileInfo, err := os.Stat(filePath) - require.NoError(t, err) - fileSize := fileInfo.Size() - t.Logf("Test file size: %d bytes", fileSize) - g, ctx := errgroup.WithContext(context.Background()) - // ctx = log.WithDebugValue(ctx, map[string]map[string]int{"func": {"TestWriterNewSample": 9}}) - for i := 0; i < streamplaceTestCount; i++ { - g.Go(func() error { - bs := bytes.Buffer{} - err := writerNewSampleInner(ctx, i, &bs) - if err != nil { - return err - } - if bs.Len() != int(fileSize) { - return fmt.Errorf("expected %d bytes, got %d", fileSize, bs.Len()) - } - return nil - }) - } - err = g.Wait() - require.NoError(t, err) + withNoGSTLeaks(t, func() { + filePath := getFixture("5sec.mp4") + fileInfo, err := os.Stat(filePath) + require.NoError(t, err) + fileSize := fileInfo.Size() + t.Logf("Test file size: %d bytes", fileSize) + g, ctx := errgroup.WithContext(context.Background()) + // ctx = log.WithDebugValue(ctx, map[string]map[string]int{"func": {"TestWriterNewSample": 9}}) + for i := 0; i < streamplaceTestCount; i++ { + g.Go(func() error { + bs := bytes.Buffer{} + err := writerNewSampleInner(ctx, i, &bs) + if err != nil { + return err + } + if bs.Len() != int(fileSize) { + return fmt.Errorf("expected %d bytes, got %d", fileSize, bs.Len()) + } + return nil + }) + } + err = g.Wait() + require.NoError(t, err) + }) } func writerNewSampleInner(ctx context.Context, i int, w io.Writer) error { diff --git a/pkg/media/leak_test.go b/pkg/media/leak_test.go index 47dff1cb7..a3df0cd97 100644 --- a/pkg/media/leak_test.go +++ b/pkg/media/leak_test.go @@ -8,6 +8,7 @@ import ( "regexp" "runtime" "runtime/debug" + "strconv" "strings" "sync" "syscall" @@ -19,6 +20,21 @@ import ( "stream.place/streamplace/pkg/gstinit" ) +var streamplaceTestCount = 50 + +func init() { + testRunsStr := os.Getenv("STREAMPLACE_TEST_COUNT") + if testRunsStr != "" { + var err error + streamplaceTestCount, err = strconv.Atoi(testRunsStr) + if err != nil { + panic(fmt.Sprintf("STREAMPLACE_TEST_COUNT is not a number: %s", testRunsStr)) + } + } +} + +var LeakTestMutex sync.Mutex + const IgnoreLeaks = "STREAMPLACE_IGNORE_LEAKS" const GSTDebugNeeded = "leaks:9,GST_TRACER:9" const LeakLine = "GST_TRACER :0:: object-alive" @@ -152,6 +168,17 @@ func checkGStreamerLeaks(t *testing.T, expected int) { fmt.Println(l) } LeakReportMutex.Unlock() + require.Equal(t, expected, len(LeakReport), "Leaks found") } - require.Equal(t, expected, len(LeakReport), "Leaks found") +} + +func withNoGSTLeaks(t *testing.T, f func()) { + LeakTestMutex.Lock() + defer LeakTestMutex.Unlock() + gstinit.InitGST() + before := getLeakCount(t) + defer checkGStreamerLeaks(t, before) + // ignore := goleak.IgnoreCurrent() + // defer goleak.VerifyNone(t, ignore) + f() } diff --git a/pkg/media/media_data_parser_test.go b/pkg/media/media_data_parser_test.go index 9e4618253..c5333f8d1 100644 --- a/pkg/media/media_data_parser_test.go +++ b/pkg/media/media_data_parser_test.go @@ -2,40 +2,27 @@ package media import ( "context" - "fmt" "io" "os" "testing" "github.com/stretchr/testify/require" - "go.uber.org/goleak" - "stream.place/streamplace/pkg/gstinit" "stream.place/streamplace/pkg/log" ) func TestMediaDataParser(t *testing.T) { - gstinit.InitGST() - before := getLeakCount(t) - // defer checkGStreamerLeaks(t, before+1) - defer func() { - after := getLeakCount(t) - if after > before { - fmt.Printf("detected %d leaks", after-before) - } - }() - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) + withNoGSTLeaks(t, func() { + // Open input file + inputFile, err := os.Open(getFixture("sample-segment.mp4")) + require.NoError(t, err) + defer inputFile.Close() + bs, err := io.ReadAll(inputFile) + require.NoError(t, err) - // Open input file - inputFile, err := os.Open(getFixture("sample-segment.mp4")) - require.NoError(t, err) - defer inputFile.Close() - bs, err := io.ReadAll(inputFile) - require.NoError(t, err) - - ctx := log.WithDebugValue(context.Background(), map[string]map[string]int{"GStreamerFunc": {"ParseSegmentMediaData": 9}}) - mediaData, err := ParseSegmentMediaData(ctx, bs) - require.NoError(t, err) - require.NotNil(t, mediaData) - require.Greater(t, mediaData.Duration, int64(0), "Video duration should not be empty") + ctx := log.WithDebugValue(context.Background(), map[string]map[string]int{"GStreamerFunc": {"ParseSegmentMediaData": 9}}) + mediaData, err := ParseSegmentMediaData(ctx, bs) + require.NoError(t, err) + require.NotNil(t, mediaData) + require.Greater(t, mediaData.Duration, int64(0), "Video duration should not be empty") + }) } diff --git a/pkg/media/packetize_test.go b/pkg/media/packetize_test.go index cebd7d943..85530330b 100644 --- a/pkg/media/packetize_test.go +++ b/pkg/media/packetize_test.go @@ -8,28 +8,22 @@ import ( "time" "github.com/stretchr/testify/require" - "go.uber.org/goleak" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/bus" - "stream.place/streamplace/pkg/gstinit" ) func TestPacketize(t *testing.T) { - gstinit.InitGST() - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) - - g, _ := errgroup.WithContext(context.Background()) - for range streamplaceTestCount { - g.Go(func() error { - innerTestPacketize(t) - return nil - }) - } - err := g.Wait() - require.NoError(t, err) + withNoGSTLeaks(t, func() { + g, _ := errgroup.WithContext(context.Background()) + for range streamplaceTestCount { + g.Go(func() error { + innerTestPacketize(t) + return nil + }) + } + err := g.Wait() + require.NoError(t, err) + }) } func innerTestPacketize(t *testing.T) { diff --git a/pkg/media/rtcrec_test.go b/pkg/media/rtcrec_test.go index 75cd81429..a495a89d4 100644 --- a/pkg/media/rtcrec_test.go +++ b/pkg/media/rtcrec_test.go @@ -16,36 +16,38 @@ import ( ) func TestRTCRecording(t *testing.T) { - globalerror.GlobalErrors = []error{} - ctx := context.Background() - dir, err := os.MkdirTemp("", "rtcrec-test-*") - require.NoError(t, err) - defer os.RemoveAll(dir) - cli := &config.CLI{} - fs := cli.NewFlagSet("rtcrec-test") - err = cli.Parse(fs, []string{"--data-dir", dir, "-wide-open=true"}) - require.NoError(t, err) - mm, err := MakeMediaManager(context.Background(), cli, nil, &boring.BoringReplicator{}, nil, nil, nil) - require.NoError(t, err) - priv, pub, err := spkey.GenerateStreamKey() - require.NoError(t, err) - signer, err := spkey.KeyToSigner(priv) - require.NoError(t, err) - mediaSigner, err := MakeMediaSigner(ctx, cli, pub.DIDKey(), signer) - require.NoError(t, err) - // ctx := context.Background() - // mm, ms := getStaticTestMediaManager(t) - fd, err := os.Open(getFixture("intermittent-tracks.cbor")) - require.NoError(t, err) - defer fd.Close() - pc, err := rtcrec.NewReplayPeerConnection(ctx, fd) - require.NoError(t, err) - done := make(chan struct{}) - answer, err := mm.WebRTCIngest(ctx, &webrtc.SessionDescription{SDP: "placeholder"}, mediaSigner, pc, done) - require.NoError(t, err) - fmt.Println(answer.SDP) - <-done - for _, err := range globalerror.GlobalErrors { - fmt.Printf("got error, non-fatal for now: %v\n", err) - } + withNoGSTLeaks(t, func() { + globalerror.GlobalErrors = []error{} + ctx := context.Background() + dir, err := os.MkdirTemp("", "rtcrec-test-*") + require.NoError(t, err) + defer os.RemoveAll(dir) + cli := &config.CLI{} + fs := cli.NewFlagSet("rtcrec-test") + err = cli.Parse(fs, []string{"--data-dir", dir, "-wide-open=true"}) + require.NoError(t, err) + mm, err := MakeMediaManager(context.Background(), cli, nil, &boring.BoringReplicator{}, nil, nil, nil) + require.NoError(t, err) + priv, pub, err := spkey.GenerateStreamKey() + require.NoError(t, err) + signer, err := spkey.KeyToSigner(priv) + require.NoError(t, err) + mediaSigner, err := MakeMediaSigner(ctx, cli, pub.DIDKey(), signer) + require.NoError(t, err) + // ctx := context.Background() + // mm, ms := getStaticTestMediaManager(t) + fd, err := os.Open(getFixture("intermittent-tracks.cbor")) + require.NoError(t, err) + defer fd.Close() + pc, err := rtcrec.NewReplayPeerConnection(ctx, fd) + require.NoError(t, err) + done := make(chan struct{}) + answer, err := mm.WebRTCIngest(ctx, &webrtc.SessionDescription{SDP: "placeholder"}, mediaSigner, pc, done) + require.NoError(t, err) + fmt.Println(answer.SDP) + <-done + for _, err := range globalerror.GlobalErrors { + fmt.Printf("got error, non-fatal for now: %v\n", err) + } + }) } diff --git a/pkg/media/segment_conv_test.go b/pkg/media/segment_conv_test.go index 01880696e..74a9c6af4 100644 --- a/pkg/media/segment_conv_test.go +++ b/pkg/media/segment_conv_test.go @@ -9,68 +9,62 @@ import ( "time" "github.com/stretchr/testify/require" - "go.uber.org/goleak" "golang.org/x/sync/errgroup" ) func TestMP4ToMPEGTS(t *testing.T) { - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) + withNoGSTLeaks(t, func() { + for _, file := range []string{"sample-segment.mp4", "short-video.mp4"} { + // Open input file + inputFile, err := os.Open(getFixture(file)) + require.NoError(t, err) + defer inputFile.Close() + + // Create a buffer for output + buf := bytes.Buffer{} + + bs, err := io.ReadAll(inputFile) + require.NoError(t, err) + // Create temporary output file + + g, _ := errgroup.WithContext(context.Background()) + for range streamplaceTestCount { + g.Go(func() error { + _, err := MP4ToMPEGTS(context.Background(), bytes.NewReader(bs), &buf) + return err + }) + } + err = g.Wait() + require.NoError(t, err) + // Convert MPEG-TS to MP4 + + // Convert MP4 to MPEG-TS + dur, err := MP4ToMPEGTS(context.Background(), bytes.NewReader(bs), &buf) + require.NoError(t, err) + require.Greater(t, dur, int64(0), "Duration should be greater than 0") + + // Verify buffer has content + require.Greater(t, buf.Len(), 0, "Output buffer should not be empty") + } + }) +} - for _, file := range []string{"sample-segment.mp4", "short-video.mp4"} { - // Open input file - inputFile, err := os.Open(getFixture(file)) +func TestMP4ToMPEGTSInvalid(t *testing.T) { + withNoGSTLeaks(t, func() { + + inputFile, err := os.Open(getFixture("sample-segment.mpegts")) require.NoError(t, err) defer inputFile.Close() - // Create a buffer for output buf := bytes.Buffer{} bs, err := io.ReadAll(inputFile) require.NoError(t, err) - // Create temporary output file - - g, _ := errgroup.WithContext(context.Background()) - for range streamplaceTestCount { - g.Go(func() error { - _, err := MP4ToMPEGTS(context.Background(), bytes.NewReader(bs), &buf) - return err - }) - } - err = g.Wait() - require.NoError(t, err) - // Convert MPEG-TS to MP4 // Convert MP4 to MPEG-TS - dur, err := MP4ToMPEGTS(context.Background(), bytes.NewReader(bs), &buf) - require.NoError(t, err) - require.Greater(t, dur, int64(0), "Duration should be greater than 0") - - // Verify buffer has content - require.Greater(t, buf.Len(), 0, "Output buffer should not be empty") - } -} - -func TestMP4ToMPEGTSInvalid(t *testing.T) { - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) - - inputFile, err := os.Open(getFixture("sample-segment.mpegts")) - require.NoError(t, err) - defer inputFile.Close() - - buf := bytes.Buffer{} - - bs, err := io.ReadAll(inputFile) - require.NoError(t, err) - - // Convert MP4 to MPEG-TS - _, err = MP4ToMPEGTS(context.Background(), bytes.NewReader(bs), &buf) - require.ErrorContains(t, err, "This file is invalid and cannot be played") + _, err = MP4ToMPEGTS(context.Background(), bytes.NewReader(bs), &buf) + require.ErrorContains(t, err, "This file is invalid and cannot be played") + }) } // func TestNoRealtime(t *testing.T) { @@ -104,65 +98,61 @@ func TestMP4ToMPEGTSInvalid(t *testing.T) { // } func TestMPEGTSToMP4(t *testing.T) { - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) - - // Open input file - inputFile, err := os.Open(getFixture("sample-segment.mpegts")) - require.NoError(t, err) - defer inputFile.Close() - bs, err := io.ReadAll(inputFile) - require.NoError(t, err) - // Create temporary output file - buf := bytes.Buffer{} - - g, _ := errgroup.WithContext(context.Background()) - for i := 0; i < streamplaceTestCount; i++ { - g.Go(func() error { - return MPEGTSToMP4(context.Background(), bytes.NewReader(bs), &buf) - }) - } - err = g.Wait() - // Convert MPEG-TS to MP4 - - require.NoError(t, err) - require.Greater(t, buf.Len(), 0, "Output file should not be empty") + withNoGSTLeaks(t, func() { + + // Open input file + inputFile, err := os.Open(getFixture("sample-segment.mpegts")) + require.NoError(t, err) + defer inputFile.Close() + bs, err := io.ReadAll(inputFile) + require.NoError(t, err) + // Create temporary output file + buf := bytes.Buffer{} + + g, _ := errgroup.WithContext(context.Background()) + for i := 0; i < streamplaceTestCount; i++ { + g.Go(func() error { + return MPEGTSToMP4(context.Background(), bytes.NewReader(bs), &buf) + }) + } + err = g.Wait() + // Convert MPEG-TS to MP4 + + require.NoError(t, err) + require.Greater(t, buf.Len(), 0, "Output file should not be empty") + }) } func TestMP4ToMPEGTSVideoMP4Audio(t *testing.T) { - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) - - // Open input file - inputFile, err := os.Open(getFixture("5sec.mp4")) - require.NoError(t, err) - defer inputFile.Close() - - // Create buffers for output - videoBuf := bytes.Buffer{} - audioBuf := bytes.Buffer{} - - // Split MP4 into MPEG-TS video and MP4 audio - start := time.Now() - err = MP4ToMPEGTSVideoMP4Audio(context.Background(), inputFile, &videoBuf, &audioBuf) - require.NoError(t, err) - elapsed := time.Since(start) - require.Less(t, elapsed, 4*time.Second, "MP4 to MPEG-TS/MP4 conversion should take less than 4 seconds") - - // Verify outputs - require.Greater(t, videoBuf.Len(), 0, "Video buffer should not be empty") - require.Greater(t, audioBuf.Len(), 0, "Audio buffer should not be empty") - - // Join video and audio back together - buf := bytes.Buffer{} - start = time.Now() - err = MPEGTSVideoMP4AudioToMP4(context.Background(), &videoBuf, &audioBuf, &buf) - require.NoError(t, err) - require.Greater(t, buf.Len(), 0, "Output buffer should not be empty") - elapsed = time.Since(start) - require.Less(t, elapsed, 4*time.Second, "MPEG-TS/MP4 to MP4 conversion should take less than 4 seconds") + withNoGSTLeaks(t, func() { + + // Open input file + inputFile, err := os.Open(getFixture("5sec.mp4")) + require.NoError(t, err) + defer inputFile.Close() + + // Create buffers for output + videoBuf := bytes.Buffer{} + audioBuf := bytes.Buffer{} + + // Split MP4 into MPEG-TS video and MP4 audio + start := time.Now() + err = MP4ToMPEGTSVideoMP4Audio(context.Background(), inputFile, &videoBuf, &audioBuf) + require.NoError(t, err) + elapsed := time.Since(start) + require.Less(t, elapsed, 4*time.Second, "MP4 to MPEG-TS/MP4 conversion should take less than 4 seconds") + + // Verify outputs + require.Greater(t, videoBuf.Len(), 0, "Video buffer should not be empty") + require.Greater(t, audioBuf.Len(), 0, "Audio buffer should not be empty") + + // Join video and audio back together + buf := bytes.Buffer{} + start = time.Now() + err = MPEGTSVideoMP4AudioToMP4(context.Background(), &videoBuf, &audioBuf, &buf) + require.NoError(t, err) + require.Greater(t, buf.Len(), 0, "Output buffer should not be empty") + elapsed = time.Since(start) + require.Less(t, elapsed, 4*time.Second, "MPEG-TS/MP4 to MP4 conversion should take less than 4 seconds") + }) } diff --git a/pkg/media/thumbnail_test.go b/pkg/media/thumbnail_test.go index 41fe1ccea..9792d83a3 100644 --- a/pkg/media/thumbnail_test.go +++ b/pkg/media/thumbnail_test.go @@ -9,75 +9,69 @@ import ( "testing" "github.com/stretchr/testify/require" - "go.uber.org/goleak" "golang.org/x/sync/errgroup" - "stream.place/streamplace/pkg/gstinit" "stream.place/streamplace/test/remote" ) func TestThumbnail(t *testing.T) { - gstinit.InitGST() - before := getLeakCount(t) - defer checkGStreamerLeaks(t, before) - ignore := goleak.IgnoreCurrent() - defer goleak.VerifyNone(t, ignore) + withNoGSTLeaks(t, func() { + // Open input file + inputFile, err := os.Open(getFixture("sample-segment.mp4")) + require.NoError(t, err) + defer inputFile.Close() + bs, err := io.ReadAll(inputFile) + require.NoError(t, err) - // Open input file - inputFile, err := os.Open(getFixture("sample-segment.mp4")) - require.NoError(t, err) - defer inputFile.Close() - bs, err := io.ReadAll(inputFile) - require.NoError(t, err) + ctx := context.Background() + g, ctx := errgroup.WithContext(ctx) - ctx := context.Background() - g, ctx := errgroup.WithContext(ctx) + for i := 0; i < streamplaceTestCount; i++ { + g.Go(func() error { + thumbnail := bytes.Buffer{} + // thumbnailCtx = log.WithDebugValue(ctx, map[string]map[string]int{"function": {"Thumbnail": 9}}) + err := Thumbnail(ctx, bytes.NewReader(bs), &thumbnail, "png") + if err != nil { + return err + } + if thumbnail.Len() == 0 { + return fmt.Errorf("thumbnail buffer is empty") + } + require.Equal(t, thumbnail.Len(), 1418910) + return nil + }) + g.Go(func() error { + thumbnail := bytes.Buffer{} + // thumbnailCtx = log.WithDebugValue(ctx, map[string]map[string]int{"function": {"Thumbnail": 9}}) + err := Thumbnail(ctx, bytes.NewReader(bs), &thumbnail, "jpeg") + if err != nil { + return err + } + if thumbnail.Len() == 0 { + return fmt.Errorf("thumbnail buffer is empty") + } + require.Equal(t, thumbnail.Len(), 140969) + return nil + }) + } - for i := 0; i < streamplaceTestCount; i++ { - g.Go(func() error { - thumbnail := bytes.Buffer{} - // thumbnailCtx = log.WithDebugValue(ctx, map[string]map[string]int{"function": {"Thumbnail": 9}}) - err := Thumbnail(ctx, bytes.NewReader(bs), &thumbnail, "png") - if err != nil { - return err - } - if thumbnail.Len() == 0 { - return fmt.Errorf("thumbnail buffer is empty") - } - require.Equal(t, thumbnail.Len(), 1418910) - return nil - }) - g.Go(func() error { - thumbnail := bytes.Buffer{} - // thumbnailCtx = log.WithDebugValue(ctx, map[string]map[string]int{"function": {"Thumbnail": 9}}) - err := Thumbnail(ctx, bytes.NewReader(bs), &thumbnail, "jpeg") - if err != nil { - return err - } - if thumbnail.Len() == 0 { - return fmt.Errorf("thumbnail buffer is empty") - } - require.Equal(t, thumbnail.Len(), 140969) - return nil - }) - } - - err = g.Wait() - require.NoError(t, err) + err = g.Wait() + require.NoError(t, err) + }) } // This segment once caused a segfault in gst-libav. // It doesn't gotta work but it does gotta not crash. func TestKryptoniteThumbnail(t *testing.T) { - gstinit.InitGST() - - inputFile, err := os.Open(remote.RemoteFixture("46c876d5e6c4124275b8856431833adaad31cb5246caca8ded9dc4d37de400a4/kryptonite-screenshot.mp4")) - require.NoError(t, err) - defer inputFile.Close() - bs, err := io.ReadAll(inputFile) - require.NoError(t, err) + withNoGSTLeaks(t, func() { + inputFile, err := os.Open(remote.RemoteFixture("46c876d5e6c4124275b8856431833adaad31cb5246caca8ded9dc4d37de400a4/kryptonite-screenshot.mp4")) + require.NoError(t, err) + defer inputFile.Close() + bs, err := io.ReadAll(inputFile) + require.NoError(t, err) - thumbnail := bytes.Buffer{} - err = Thumbnail(context.Background(), bytes.NewReader(bs), &thumbnail, "png") - require.NoError(t, err) - require.Equal(t, 561486, thumbnail.Len()) + thumbnail := bytes.Buffer{} + err = Thumbnail(context.Background(), bytes.NewReader(bs), &thumbnail, "png") + require.NoError(t, err) + require.Equal(t, 561486, thumbnail.Len()) + }) } diff --git a/pkg/media/webrtc_ingest.go b/pkg/media/webrtc_ingest.go index 974508c12..17c808e39 100644 --- a/pkg/media/webrtc_ingest.go +++ b/pkg/media/webrtc_ingest.go @@ -47,15 +47,6 @@ func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionD return nil, fmt.Errorf("failed to get queue element from pipeline: %w", err) } - signerElem, err := mm.SegmentAndSignElem(ctx, signer) - if err != nil { - return nil, fmt.Errorf("failed create signer element: %w", err) - } - err = pipeline.Add(signerElem) - if err != nil { - return nil, fmt.Errorf("failed to add signer element to pipeline: %w", err) - } - // err = queue.Link(signerElem) // if err != nil { // return nil, fmt.Errorf("failed to link queue to signer element: %w", err) @@ -70,18 +61,6 @@ func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionD videoSrcPad := videoSrcPads[0] audioSrcPad := videoSrcPads[1] - signerElemPads, err := signerElem.GetPads() - if err != nil { - return nil, fmt.Errorf("failed to get signerElemPads from signer element: %w", err) - } - if len(signerElemPads) != 2 { - return nil, fmt.Errorf("failed to get signerElemPads from signer element") - } - signerElemVideoPad := signerElemPads[0] - signerElemAudioPad := signerElemPads[1] - videoSrcPad.Link(signerElemVideoPad) - audioSrcPad.Link(signerElemAudioPad) - videoSrcElem, err := pipeline.GetElementByName("videosrc") if err != nil { return nil, fmt.Errorf("failed to get videoSrcElem element from pipeline: %w", err) @@ -113,10 +92,41 @@ func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionD // Create channel that is blocked until ICE Gathering is complete gatherComplete := rtcrec.GatheringCompletePromise(peerConnection) - // Setup complete! Now we boot up streaming in the background while returning the SDP offer to the user. + ctx, cancel := context.WithCancel(ctx) + signerElem, err := mm.SegmentAndSignElem(ctx, signer) + if err != nil { + cancel() + return nil, fmt.Errorf("failed create signer element: %w", err) + } + err = pipeline.Add(signerElem) + if err != nil { + cancel() + return nil, fmt.Errorf("failed to add signer element to pipeline: %w", err) + } + signerElemPads, err := signerElem.GetPads() + if err != nil { + cancel() + return nil, fmt.Errorf("failed to get signerElemPads from signer element: %w", err) + } + if len(signerElemPads) != 2 { + cancel() + return nil, fmt.Errorf("failed to get signerElemPads from signer element") + } + signerElemVideoPad := signerElemPads[0] + signerElemAudioPad := signerElemPads[1] + linked := videoSrcPad.Link(signerElemVideoPad) + if linked != gst.PadLinkOK { + cancel() + return nil, fmt.Errorf("failed to link videoSrcPad to signerElemVideoPad") + } + linked = audioSrcPad.Link(signerElemAudioPad) + if linked != gst.PadLinkOK { + cancel() + return nil, fmt.Errorf("failed to link audioSrcPad to signerElemAudioPad") + } + // Setup complete! Now we boot up streaming in the background while returning the SDP offer to the user. go func() { - ctx, cancel := context.WithCancel(ctx) defer cancel() defer func() { close(done) }() diff --git a/pkg/rtcrec/cbor.go b/pkg/rtcrec/cbor.go index 007505d1b..1dff9ab61 100644 --- a/pkg/rtcrec/cbor.go +++ b/pkg/rtcrec/cbor.go @@ -4,6 +4,7 @@ import ( "fmt" "io" "sort" + "sync" "time" "github.com/fxamacker/cbor/v2" @@ -39,9 +40,10 @@ func (d *WebRTCEventDecoder) Next() (*WebRTCEvent, error) { } type WebRTCEventGroup struct { - Events map[string][]*WebRTCEvent - Tracks map[webrtc.SSRC]map[string][]*WebRTCEvent - FirstTime time.Time + Events map[string][]*WebRTCEvent + Tracks map[webrtc.SSRC]map[string][]*WebRTCEvent + FirstTime time.Time + EventMutex sync.Mutex } const ( @@ -200,8 +202,10 @@ func (g *WebRTCEventGroup) Next(eventType string) *WebRTCEvent { if len(g.Events[eventType]) == 0 { return nil } + g.EventMutex.Lock() ev := g.Events[eventType][0] g.Events[eventType] = g.Events[eventType][1:] + g.EventMutex.Unlock() return ev }