From cfc8a99397a605eb2b34db16074bc380ccd47c98 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Mon, 12 Oct 2026 06:31:35 -0700 Subject: [PATCH] e2e: push a Music stem over fMP4 ingest and run the stems flows The harness's fMP4 push now sends what the OBS plugin sends with a "Music" stem: the fixture's audio as the full mix and as the stem, each in AAC and Opus, labeled through muxl's WithSegmentTrackMeta. CI runs the web stems spec with that push, and the Android Maestro suite runs on it so audio-stems.yaml executes; WHIP ingest stays covered by the main web run. --- .github/workflows/build.yaml | 13 +++-- hack/e2e-local.sh | 5 +- pkg/cmd/fmp4push.go | 92 +++++++++++++++++++++++++++++++----- 3 files changed, 94 insertions(+), 16 deletions(-) diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml index c32b0b69e..ad12190c5 100644 --- a/.github/workflows/build.yaml +++ b/.github/workflows/build.yaml @@ -170,10 +170,12 @@ jobs: run: hack/e2e-web-local.sh # The stream specs again, with the harness pushing the way the OBS plugin - # does (/api/ingest/client-configuration, then fragmented MP4), so that - # ingest path is covered from push to web playback. + # does (/api/ingest/client-configuration, then fragmented MP4 with a + # "Music" stem), so that ingest path and audio stems are covered from + # push to web playback. - name: Run web stream e2e tests over fMP4 ingest - run: hack/e2e-web-local.sh flows/stream.spec.ts + run: + hack/e2e-web-local.sh flows/stream.spec.ts flows/audio-stems.spec.ts env: SP_E2E_INGEST: fmp4 @@ -279,8 +281,11 @@ jobs: -no-snapshot-save -no-window -gpu swiftshader_indirect -noaudio -no-boot-anim -camera-back none disable-animations: true - # one line: the action runs each script line in its own shell + # one line: the action runs each script line in its own shell. + # fMP4 ingest gives the test stream a "Music" stem for the audio + # stems flow; WHIP ingest stays covered by the web suite. script: + SP_E2E_INGEST=fmp4 E2E_HARNESS_LOG=$PWD/.maestro/artifacts/harness.log bash hack/e2e-local.sh android diff --git a/hack/e2e-local.sh b/hack/e2e-local.sh index 67f6469fc..10f209a46 100755 --- a/hack/e2e-local.sh +++ b/hack/e2e-local.sh @@ -143,11 +143,14 @@ adb_ shell am force-stop com.android.chrome || true # read stored records from the PDS's plain-HTTP listener. device_offset="$(adb_ shell date +%z | tr -d '\r')" [[ "$device_offset" =~ ^[+-][0-9]{4}$ ]] || { echo "can't read $DEVICE's UTC offset: '$device_offset'"; exit 1; } +# SP_E2E_INGEST also picks how the harness streams; flows that need the fMP4 +# push (audio stems) check it. MAESTRO_ARGS=(-e APP_ID="$APP_ID" -e SERVER_URL="$SERVER_HTTPS_URL" -e SERVER_HTTP_URL="$SERVER_URL" -e ACCOUNT_HANDLE="$ACCOUNT_HANDLE" -e ACCOUNT_PASSWORD="$ACCOUNT_PASSWORD" -e ACCOUNT_DID="$ACCOUNT_DID" -e PDS_HTTP_URL="$PDS_HTTP_URL" - -e DEVICE_UTC_OFFSET="${device_offset:0:3}:${device_offset:3}") + -e DEVICE_UTC_OFFSET="${device_offset:0:3}:${device_offset:3}" + -e SP_E2E_INGEST="${SP_E2E_INGEST:-whip}") # --- run the flows --------------------------------------------------------- # takeScreenshot paths are relative to maestro's cwd, so run from the diff --git a/pkg/cmd/fmp4push.go b/pkg/cmd/fmp4push.go index 9169a44d3..ad9b09a9e 100644 --- a/pkg/cmd/fmp4push.go +++ b/pkg/cmd/fmp4push.go @@ -7,19 +7,33 @@ import ( "fmt" "io" "net/http" + "slices" + "strconv" "strings" "github.com/go-gst/go-gst/pkg/gst" app "github.com/go-gst/go-gst/pkg/gstapp" "stream.place/streamplace/pkg/gstinit" "stream.place/streamplace/pkg/media" + "stream.place/streamplace/pkg/muxl" ) +// pushFMP4Tracks labels the tracks pushFMP4's muxer creates (in pad order) the +// way the OBS plugin labels a full mix on mixer Track 1 plus a stem named +// "Music" on Track 2: an AAC and an Opus encode of each, sharing a group. +var pushFMP4Tracks = map[uint32]muxl.TrackMeta{ + 2: {AlternateGroup: 1, Label: "Track 1"}, + 3: {AlternateGroup: 1, Label: "Track 1"}, + 4: {AlternateGroup: 2, Label: "Music"}, + 5: {AlternateGroup: 2, Label: "Music"}, +} + // pushFMP4 streams file to the node at endpoint the way the Streamplace OBS -// plugin does: it asks /api/ingest/client-configuration for the FMP4 ingest -// endpoint, then POSTs the file there as realtime-paced fragmented MP4 with a -// chunked body, with AAC audio like OBS's default encoder. It returns when the -// file ends, the node hangs up, or ctx is done. +// plugin does with one stem: it asks /api/ingest/client-configuration for the +// FMP4 ingest endpoint, then POSTs realtime-paced fragmented MP4 with a +// chunked body. The file's audio is the full mix, and a copy of it is the +// "Music" stem, each as AAC and Opus. It returns when the file ends, the +// node hangs up, or ctx is done. func pushFMP4(ctx context.Context, endpoint, streamKey, file string) error { ingestURL, err := fmp4IngestURL(ctx, endpoint, streamKey) if err != nil { @@ -28,12 +42,18 @@ func pushFMP4(ctx context.Context, endpoint, streamKey, file string) error { gstinit.InitGST() // identity sync=true paces each track at realtime, so the push plays as a - // live stream instead of arriving in one burst. + // live stream instead of arriving in one burst. mp4mux numbers tracks in + // pad order, which pushFMP4Tracks relies on. launched, err := gst.ParseLaunch(strings.Join([]string{ "filesrc name=filesrc ! qtdemux name=demux", - "demux.video_0 ! queue ! identity sync=true ! h264parse ! mux.", - "demux.audio_0 ! queue ! identity sync=true ! opusdec ! audioconvert ! audioresample ! fdkaacenc ! aacparse ! mux.", "mp4mux name=mux fragment-duration=100 ! appsink name=sink sync=false", + "demux.video_0 ! queue ! identity sync=true ! h264parse ! mux.video_0", + "demux.audio_0 ! queue ! identity sync=true ! tee name=mix", + "mix. ! queue ! opusdec ! audioconvert ! audioresample ! tee name=pcm", + "pcm. ! queue ! fdkaacenc ! aacparse ! mux.audio_0", + "mix. ! queue ! opusparse ! mux.audio_1", + "pcm. ! queue ! fdkaacenc ! aacparse ! mux.audio_2", + "pcm. ! queue ! opusenc ! opusparse ! mux.audio_3", }, "\n")) if err != nil { return err @@ -55,10 +75,12 @@ func pushFMP4(ctx context.Context, endpoint, streamKey, file string) error { ctx, cancel := context.WithCancel(ctx) defer cancel() - body, bodyWriter := io.Pipe() + muxed, muxedWriter := io.Pipe() sink.(app.AppSink).SetCallbacks(&app.AppSinkCallbacks{ - NewSample: media.WriterNewSample(ctx, bodyWriter), + NewSample: media.WriterNewSample(ctx, muxedWriter), }) + body, bodyWriter := io.Pipe() + go func() { _ = bodyWriter.CloseWithError(labelTracks(ctx, muxed, bodyWriter)) }() pushErr := make(chan error, 1) go func() { @@ -76,6 +98,7 @@ func pushFMP4(ctx context.Context, endpoint, streamKey, file string) error { } // A node that hangs up mid-stream stops the pipeline too. _ = body.CloseWithError(err) + _ = muxed.CloseWithError(err) cancel() pushErr <- err }() @@ -84,16 +107,63 @@ func pushFMP4(ctx context.Context, endpoint, streamKey, file string) error { return fmt.Errorf("fmp4 push: set playing: state change failed") } pipelineErr := media.HandleBusMessages(ctx, pipeline) - _ = bodyWriter.CloseWithError(pipelineErr) + _ = muxedWriter.CloseWithError(pipelineErr) if err := <-pushErr; err != nil { return fmt.Errorf("push to %s: %w", strings.ReplaceAll(ingestURL, streamKey, "{stream_key}"), err) } return pipelineErr } +// labelTracks re-emits an fMP4 stream through muxl with pushFMP4Tracks applied, +// so its init carries the groups and titles the plugin would write: the +// canonical init, then each GoP's track segments. It stops when in closes: +// muxl's segmenter must not see a cancelled context (see media.signSegments). +func labelTracks(ctx context.Context, in io.Reader, out io.Writer) error { + ctx = context.WithoutCancel(ctx) + events := make(chan *muxl.MuxlEvent, 16) + errCh := make(chan error, 1) + go func() { + errCh <- muxl.RunMuxlSegmenterEvents(ctx, in, events, muxl.WithSegmentTrackMeta(pushFMP4Tracks)) + close(events) + }() + var writeErr error + for ev := range events { + if writeErr != nil { + continue // keep draining so the segmenter can finish + } + switch ev.Type { + case "init": + _, writeErr = out.Write(ev.Data) + case "segment": + ids := make([]int, 0, len(ev.Tracks)) + for k := range ev.Tracks { + id, _ := strconv.Atoi(k) + ids = append(ids, id) + } + slices.Sort(ids) + for _, id := range ids { + if _, writeErr = out.Write(ev.Tracks[strconv.Itoa(id)]); writeErr != nil { + break + } + } + } + } + if err := <-errCh; err != nil { + return err + } + return writeErr +} + // fmp4IngestURL asks the node for the plugin's ingest URL. func fmp4IngestURL(ctx context.Context, endpoint, streamKey string) (string, error) { - reqBody, err := json.Marshal(map[string]string{"authentication": streamKey}) + // Named mixer tracks are how a streamer asks for stems; see obs-plugin/README.md. + reqBody, err := json.Marshal(map[string]any{ + "authentication": streamKey, + "preferences": map[string]any{"audio_tracks": []map[string]any{ + {"track_id": 0, "name": ""}, + {"track_id": 1, "name": "Music"}, + }}, + }) if err != nil { return "", err } -- 2.54.0