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 }