From e7f8e29916be033d1fa1e4fc4b3bd54af2702a55 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Mon, 12 Oct 2026 06:30:36 -0700 Subject: [PATCH] Carry audio stems from the OBS plugin to WebRTC viewers When the plugin's config request names OBS mixer tracks 2-6, the node asks for an AAC and an Opus encoder for the full mix (Track 1) and for up to two named tracks. The fmp4-direct ingest signs with muxl's SourceTrackMeta, so every segment's catalog carries each track's alternateGroup and label. Audio selection is group-aware: the lowest group is the main audio, and HLS, RTMP push, rendition playback, codec completion and the WebRTC mix all pick from it as before. Segments without groups keep today's meaning. Segment records list every audio track with its real codec and label. WebRTC playback adds one Opus track per stem group after the main audio, in order, so they land on the viewer's offered audio transceivers; transcoded renditions carry the stems too. Needs streamplace/muxl#8 (track labels); go.mod pins its branch commit until a release. --- go.mod | 2 +- go.sum | 4 +- pkg/api/fmp4_ingest.go | 68 ++++- pkg/api/fmp4_ingest_test.go | 44 +++ pkg/bus/segchanman.go | 24 ++ pkg/bus/segchanman_test.go | 15 + pkg/livehls/livehls.go | 77 +++-- pkg/livehls/stems_test.go | 54 ++++ pkg/localdb/segment.go | 43 ++- pkg/localdb/segment_audio_test.go | 44 +++ pkg/media/audio_stems_test.go | 372 +++++++++++++++++++++++ pkg/media/concat_demux.go | 128 ++++---- pkg/media/concat_demux_test.go | 2 +- pkg/media/dynamic_manifest_test.go | 2 +- pkg/media/ingest_worker.go | 4 +- pkg/media/media_data_parser.go | 27 ++ pkg/media/media_signer.go | 8 +- pkg/media/mp4_ingest.go | 8 +- pkg/media/muxl_segment.go | 5 +- pkg/media/packetize.go | 172 ++++++----- pkg/media/rendition_playback.go | 66 ++-- pkg/media/rtmp_ingest.go | 2 +- pkg/media/thumbnail.go | 2 +- pkg/media/transcode.go | 155 ++++++++-- pkg/media/transcode_stream.go | 33 +- pkg/media/transcode_stream_test.go | 2 +- pkg/media/transcode_video_test.go | 4 +- pkg/media/validate.go | 20 +- pkg/media/validate_bare_test.go | 4 +- pkg/media/webrtc_playback2.go | 131 ++++++-- pkg/muxl/muxl.go | 40 ++- pkg/replication/websocketrep/moq_test.go | 2 +- pkg/vod/clip_test.go | 2 +- 33 files changed, 1263 insertions(+), 303 deletions(-) create mode 100644 pkg/api/fmp4_ingest_test.go create mode 100644 pkg/livehls/stems_test.go create mode 100644 pkg/localdb/segment_audio_test.go create mode 100644 pkg/media/audio_stems_test.go diff --git a/go.mod b/go.mod index 353e6da3f..2c1e28071 100644 --- a/go.mod +++ b/go.mod @@ -79,7 +79,7 @@ require ( github.com/streamplace/atmoq/go v0.0.4-0.20260701223355-13757de4ae08 github.com/streamplace/atproto-oauth-golang v0.0.0-20260413212710-98956064d06c github.com/streamplace/glex v0.0.0-20260820164827-814f46540f22 - github.com/streamplace/muxl/go v0.3.6 + github.com/streamplace/muxl/go v0.3.7-0.20261012131447-1c84542b3758 github.com/streamplace/oatproxy v0.0.0-20260710202406-60d97b9d780b github.com/stretchr/testify v1.11.1 github.com/tdewolff/canvas v0.0.0-20250728095813-50d4cb1eee71 diff --git a/go.sum b/go.sum index 6c4320d65..d56afafe8 100644 --- a/go.sum +++ b/go.sum @@ -1399,8 +1399,8 @@ github.com/streamplace/moq/go/ffi v0.0.0-20261017225858-0014d50204dd h1:RrOws2E5 github.com/streamplace/moq/go/ffi v0.0.0-20261017225858-0014d50204dd/go.mod h1:a+uEvcnNtY3e9T4/3UqjHBnyYHNLY8rZp+/m7tmL5n8= github.com/streamplace/moq/go/wrapper v0.0.0-20261017225858-0014d50204dd h1:eoAa4ApPlxQ6DUntQF+KP6gXWUxr0fH+f/5VGBLAe5w= github.com/streamplace/moq/go/wrapper v0.0.0-20261017225858-0014d50204dd/go.mod h1:byuivIHyq+guVzIq5jSpxZ1h6KXQ2jG4FuCSoFZDiOA= -github.com/streamplace/muxl/go v0.3.6 h1:lyWnY15nft/0GK3LBA4OkD2aHrLV9RRqI+YcDRPFk3g= -github.com/streamplace/muxl/go v0.3.6/go.mod h1:aCyYTW3o6c1Kush9UJ/Yv6EYMUbj8l8GTD7cHKcSxw8= +github.com/streamplace/muxl/go v0.3.7-0.20261012131447-1c84542b3758 h1:qyakbXaQagdSTeKxMr8onok3ahMSmJGURJUUJGuKAs0= +github.com/streamplace/muxl/go v0.3.7-0.20261012131447-1c84542b3758/go.mod h1:aCyYTW3o6c1Kush9UJ/Yv6EYMUbj8l8GTD7cHKcSxw8= github.com/streamplace/oatproxy v0.0.0-20260710202406-60d97b9d780b h1:eWbwCtBbMyrDTHLYIold07OR2hmvzXsbAUxi57ElMLk= github.com/streamplace/oatproxy v0.0.0-20260710202406-60d97b9d780b/go.mod h1:wpY+T/wE00jrUhgh2dKXbbE91D36u86KGlENK/hWFkE= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= diff --git a/pkg/api/fmp4_ingest.go b/pkg/api/fmp4_ingest.go index a9bd6def1..ed061c4c2 100644 --- a/pkg/api/fmp4_ingest.go +++ b/pkg/api/fmp4_ingest.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "net/http" + "slices" "github.com/google/uuid" "github.com/julienschmidt/httprouter" @@ -21,12 +22,35 @@ import ( // both. Only the fields the node reads or sets are modeled here. type clientConfigurationRequest struct { Authentication string `json:"authentication"` + Preferences struct { + // AudioTracks names the OBS mixer tracks: track_id 0..5, name empty + // when the streamer left the track unnamed. + AudioTracks []audioTrackPreference `json:"audio_tracks"` + } `json:"preferences"` +} + +type audioTrackPreference struct { + TrackID int `json:"track_id"` + Name string `json:"name"` } type clientConfiguration struct { Meta clientConfigurationMeta `json:"meta"` Status clientConfigurationStatus `json:"status"` IngestEndpoints []ingestEndpoint `json:"ingest_endpoints,omitempty"` + // AudioConfigurations lists the audio encoders the plugin creates, keyed + // by the output they feed ("live"); absent, the plugin keeps the encoder + // from OBS's own output settings. + AudioConfigurations map[string][]audioConfiguration `json:"audio_configurations,omitempty"` +} + +type audioConfiguration struct { + Codec string `json:"codec"` + TrackID int `json:"track_id"` + Channels int `json:"channels"` + Settings struct { + Bitrate int `json:"bitrate"` + } `json:"settings"` } type clientConfigurationMeta struct { @@ -45,11 +69,46 @@ type ingestEndpoint struct { URLTemplate string `json:"url_template"` } +// maxStems is how many named mixer tracks become stems beside the full mix: +// OBS allows six audio encoders per output, and each track takes two. +const maxStems = 2 + +// stemAudioConfigurations lists the encoders the plugin should create when +// the streamer named mixer tracks beyond the first: an AAC and an Opus +// encode of track 0 (the full mix) and of each named track in track order, +// up to maxStems of them. With no named track it returns nil, and the plugin +// keeps the encoder from OBS's output settings: one audio track, as today. +func stemAudioConfigurations(tracks []audioTrackPreference) []audioConfiguration { + var stems []int + for _, t := range tracks { + if t.TrackID >= 1 && t.TrackID <= 5 && t.Name != "" { + stems = append(stems, t.TrackID) + } + } + if len(stems) == 0 { + return nil + } + slices.Sort(stems) + ids := append([]int{0}, stems[:min(len(stems), maxStems)]...) + var out []audioConfiguration + for _, id := range ids { + for _, c := range []struct { + codec string + bitrate int + }{{"aac", 160}, {"opus", 128}} { + cfg := audioConfiguration{Codec: c.codec, TrackID: id, Channels: 2} + cfg.Settings.Bitrate = c.bitrate + out = append(out, cfg) + } + } + return out +} + // HandleClientConfiguration answers the plugin's configuration request. A bad // stream key is reported in status, as GoLiveApi clients expect, so OBS shows -// the message. The response names no encoders, so the plugin keeps the -// encoders from OBS's output settings: one video and one audio track, which is -// what the node's segment pipeline handles today. +// the message. Unless the streamer named mixer tracks (see +// stemAudioConfigurations) the response names no encoders, so the plugin keeps +// the encoders from OBS's output settings: one video and one audio track. func (a *StreamplaceAPI) HandleClientConfiguration(ctx context.Context) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { var req clientConfigurationRequest @@ -76,6 +135,9 @@ func (a *StreamplaceAPI) HandleClientConfiguration(ctx context.Context) http.Han Protocol: "FMP4", URLTemplate: fmt.Sprintf("%s://%s/api/ingest/fmp4/{stream_key}", scheme, r.Host), }} + if live := stemAudioConfigurations(req.Preferences.AudioTracks); live != nil { + config.AudioConfigurations = map[string][]audioConfiguration{"live": live} + } } w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(config); err != nil { diff --git a/pkg/api/fmp4_ingest_test.go b/pkg/api/fmp4_ingest_test.go new file mode 100644 index 000000000..6b3818139 --- /dev/null +++ b/pkg/api/fmp4_ingest_test.go @@ -0,0 +1,44 @@ +package api + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +// The plugin's configuration request names the OBS mixer tracks; named tracks +// beyond the first become stems, each encoded as AAC and Opus beside the full +// mix, in track order and at most two of them. With nothing named the +// response carries no audio configuration at all, so the plugin keeps the +// single encoder from OBS's own settings. +func TestStemAudioConfigurations(t *testing.T) { + decode := func(body string) clientConfigurationRequest { + var req clientConfigurationRequest + require.NoError(t, json.Unmarshal([]byte(body), &req)) + return req + } + + unnamed := decode(`{"authentication":"k","preferences":{"audio_tracks":[ + {"track_id":0,"name":""},{"track_id":1,"name":""},{"track_id":2,"name":""}]}}`) + require.Nil(t, stemAudioConfigurations(unnamed.Preferences.AudioTracks), "unnamed tracks keep OBS's own encoder") + require.Nil(t, stemAudioConfigurations(nil), "a request without preferences keeps OBS's own encoder") + + onlyFirstNamed := decode(`{"preferences":{"audio_tracks":[{"track_id":0,"name":"Mix"}]}}`) + require.Nil(t, stemAudioConfigurations(onlyFirstNamed.Preferences.AudioTracks), "naming only the full mix makes no stems") + + // Three named stems arrive out of order: the two lowest mixer tracks win, + // and the full mix (track 0) always comes first. + named := decode(`{"preferences":{"audio_tracks":[ + {"track_id":0,"name":""},{"track_id":4,"name":"Game"},{"track_id":1,"name":"Music"}, + {"track_id":2,"name":""},{"track_id":3,"name":"Mic"},{"track_id":5,"name":"Alerts"}]}}`) + got, err := json.Marshal(stemAudioConfigurations(named.Preferences.AudioTracks)) + require.NoError(t, err) + require.JSONEq(t, `[ + {"codec":"aac","track_id":0,"channels":2,"settings":{"bitrate":160}}, + {"codec":"opus","track_id":0,"channels":2,"settings":{"bitrate":128}}, + {"codec":"aac","track_id":1,"channels":2,"settings":{"bitrate":160}}, + {"codec":"opus","track_id":1,"channels":2,"settings":{"bitrate":128}}, + {"codec":"aac","track_id":3,"channels":2,"settings":{"bitrate":160}}, + {"codec":"opus","track_id":3,"channels":2,"settings":{"bitrate":128}}]`, string(got)) +} diff --git a/pkg/bus/segchanman.go b/pkg/bus/segchanman.go index c88daefd5..570636345 100644 --- a/pkg/bus/segchanman.go +++ b/pkg/bus/segchanman.go @@ -31,12 +31,24 @@ type PacketizedSample struct { Duration time.Duration } +// PacketizedSegment is a segment demuxed into WebRTC samples: the video and +// the main audio (the full mix), plus one Opus track per stem the stream +// carries (see media.Packetize), all on the segment's one timeline. type PacketizedSegment struct { Video []PacketizedSample Audio []PacketizedSample + Stems []PacketizedStem // ascending by group Duration time.Duration } +// PacketizedStem is one stem's Opus samples, identified by its audio +// alternateGroup (the segment record's alternateGroup, above the main +// audio's). +type PacketizedStem struct { + Group uint32 + Audio []PacketizedSample +} + var chanSize = 1024 type SegChan struct { @@ -150,6 +162,18 @@ func (b *Bus) PublishSegment(ctx context.Context, user string, rendition string, } } +// LatestSegment returns the most recently published segment for a user and +// rendition, or nil when none is cached. +func (b *Bus) LatestSegment(user string, rendition string) *Seg { + b.segBufMutex.RLock() + defer b.segBufMutex.RUnlock() + curBuf := b.segBuf[segChanKey(user, rendition)] + if len(curBuf) == 0 { + return nil + } + return curBuf[len(curBuf)-1] +} + func (b *Bus) EndSession(ctx context.Context, user string, rendition string) { b.segChansMutex.Lock() defer b.segChansMutex.Unlock() diff --git a/pkg/bus/segchanman_test.go b/pkg/bus/segchanman_test.go index 24693a528..114b0a15a 100644 --- a/pkg/bus/segchanman_test.go +++ b/pkg/bus/segchanman_test.go @@ -29,6 +29,21 @@ func TestPublishSegmentKeepsOrder(t *testing.T) { } } +// LatestSegment is the newest published segment, nil before the first one +// and after the session ends. +func TestLatestSegment(t *testing.T) { + b := NewBus() + ctx := context.Background() + require.Nil(t, b.LatestSegment("u", "source")) + for i := 0; i < bufSize+2; i++ { + b.PublishSegment(ctx, "u", "source", &Seg{Filepath: fmt.Sprint(i)}) + } + require.Equal(t, fmt.Sprint(bufSize+1), b.LatestSegment("u", "source").Filepath) + require.Nil(t, b.LatestSegment("u", "720p"), "another rendition has its own cache") + b.EndSession(ctx, "u", "source") + require.Nil(t, b.LatestSegment("u", "source")) +} + // A subscriber that never reads does not hold the publisher. func TestPublishSegmentDropsForStalledSubscriber(t *testing.T) { b := NewBus() diff --git a/pkg/livehls/livehls.go b/pkg/livehls/livehls.go index 690755bf9..734473030 100644 --- a/pkg/livehls/livehls.go +++ b/pkg/livehls/livehls.go @@ -73,6 +73,11 @@ type Track struct { Height uint32 Channels uint32 SampleRate uint32 + // Group is an audio track's alternateGroup: tracks sharing one carry the + // same sound in different codecs, and the lowest group is the full mix + // (0 when the stream doesn't distinguish its audio). Higher groups are + // stems for WebRTC viewers; HLS plays the full mix only. + Group uint32 Segments []Segment @@ -387,18 +392,44 @@ func (w *Writer) SegmentData(trackID string, seq uint64) []byte { func (w *Writer) PrimaryAudioTrackID() string { w.mu.Lock() defer w.mu.Unlock() - primaryAudio := "" + return w.primaryAudio() +} + +// mainAudioTracks lists the full mix's audio tracks (the lowest Group), in +// order. Caller holds w.mu. +func (w *Writer) mainAudioTracks() []string { + var main []string + var group uint32 for _, tid := range w.order { - if t := w.tracks[tid]; t != nil && t.Type == "audio" { - if primaryAudio == "" { - primaryAudio = tid - } - if strings.HasPrefix(t.Codec, "mp4a") { - return tid - } + t := w.tracks[tid] + if t == nil || t.Type != "audio" { + continue + } + switch { + case len(main) == 0 || t.Group < group: + main, group = []string{tid}, t.Group + case t.Group == group: + main = append(main, tid) } } - return primaryAudio + return main +} + +// primaryAudio picks the default audio rendition from the full mix: AAC +// (broadest HLS support — Safari has no Opus), else the first track. A +// segment may carry both after codec completion. "" without audio. Caller +// holds w.mu. +func (w *Writer) primaryAudio() string { + main := w.mainAudioTracks() + for _, tid := range main { + if strings.HasPrefix(w.tracks[tid].Codec, "mp4a") { + return tid + } + } + if len(main) > 0 { + return main[0] + } + return "" } // MediaPlaylist renders the live HLS media playlist for trackID. initURL is @@ -448,8 +479,10 @@ func (w *Writer) MediaPlaylist(trackID, initURL string, segURI func(seq uint64) } // MasterPlaylist renders the HLS master playlist. trackURL maps a track id to -// its media-playlist URL. Audio tracks become EXT-X-MEDIA renditions in the -// "audio" group; video tracks become EXT-X-STREAM-INF variants. +// its media-playlist URL. The full mix's audio tracks become EXT-X-MEDIA +// renditions in the "audio" group (the primary one default, any other codec a +// selectable alternate); stems are left out; video tracks become +// EXT-X-STREAM-INF variants. func (w *Writer) MasterPlaylist(trackURL func(trackID string) string) string { w.mu.Lock() defer w.mu.Unlock() @@ -458,28 +491,11 @@ func (w *Writer) MasterPlaylist(trackURL func(trackID string) string) string { var b strings.Builder b.WriteString("#EXTM3U\n#EXT-X-VERSION:7\n#EXT-X-INDEPENDENT-SEGMENTS\n") - // Pick the primary audio rendition: prefer AAC (broadest HLS support — - // Safari has no Opus) so it's the default; any other codec (Opus) stays a - // selectable alternate. A segment may carry both after codec completion. - primaryAudio := "" - for _, tid := range w.order { - if t := w.tracks[tid]; t.Type == "audio" { - if primaryAudio == "" { - primaryAudio = tid - } - if strings.HasPrefix(t.Codec, "mp4a") { - primaryAudio = tid - break - } - } - } + primaryAudio := w.primaryAudio() var audioCodec string haveAudio := primaryAudio != "" - for _, tid := range w.order { + for _, tid := range w.mainAudioTracks() { t := w.tracks[tid] - if t.Type != "audio" { - continue - } def := "NO" if tid == primaryAudio { def = "YES" @@ -541,6 +557,7 @@ func (w *Writer) applyCatalog(cat *muxl.MuxlCatalog) { t.Timescale = c.Timescale() t.Channels = c.NumberOfChannels t.SampleRate = c.SampleRate + t.Group = c.AlternateGroup } } } diff --git a/pkg/livehls/stems_test.go b/pkg/livehls/stems_test.go new file mode 100644 index 000000000..212b24e2a --- /dev/null +++ b/pkg/livehls/stems_test.go @@ -0,0 +1,54 @@ +package livehls + +import ( + "bytes" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/muxl" +) + +// A stream with stems groups its audio: the full mix (the lowest group, here +// AAC + Opus) is what HLS plays, the stem (a higher group) stays out of the +// master playlist's audio group but is still served as a track. +func TestMasterPlaylistPlaysTheFullMixOnly(t *testing.T) { + audio := func(id uint32, codec string, group uint32, label string) muxl.MuxlAudioConfig { + return muxl.MuxlAudioConfig{Codec: codec, Container: muxl.MuxlContainer{Kind: "cmaf", Timescale: 48000, TrackID: id}, + SampleRate: 48000, NumberOfChannels: 2, AlternateGroup: group, Label: label} + } + w := NewWriter() + require.NoError(t, w.Observe(&muxl.MuxlEvent{ + Type: "init", + TrackInits: map[string][]byte{"1": []byte("V"), "2": []byte("A"), "3": []byte("B"), "4": []byte("C"), "5": []byte("D")}, + Catalog: &muxl.MuxlCatalog{ + Video: &muxl.MuxlCatalogVideo{Renditions: map[string]muxl.MuxlVideoConfig{ + "v": {Codec: "avc1.640028", Container: muxl.MuxlContainer{Kind: "cmaf", Timescale: 90000, TrackID: 1}, CodedWidth: 1280, CodedHeight: 720}, + }}, + Audio: &muxl.MuxlCatalogAudio{Renditions: map[string]muxl.MuxlAudioConfig{ + // The stem's Opus track has the lowest id: order must not decide. + "stem-opus": audio(2, "opus", 2, "Music"), + "mix-opus": audio(3, "opus", 1, "Mix"), + "mix-aac": audio(4, "mp4a.40.2", 1, "Mix"), + "stem-aac": audio(5, "mp4a.40.2", 2, "Music"), + }}, + }, + })) + seg := bytes.Repeat([]byte{1}, 10) + require.NoError(t, w.Observe(&muxl.MuxlEvent{ + Type: "segment", + Tracks: map[string][]byte{"1": seg, "2": seg, "3": seg, "4": seg, "5": seg}, + Durations: map[string]uint64{"1": 90000, "2": 48000, "3": 48000, "4": 48000, "5": 48000}, + SampleCounts: map[string]uint32{"1": 30, "2": 50, "3": 50, "4": 50, "5": 50}, + })) + + require.Equal(t, "4", w.PrimaryAudioTrackID(), "the mix's AAC is the default audio") + master := w.MasterPlaylist(func(tid string) string { return "track" + tid + ".m3u8" }) + require.Equal(t, 2, strings.Count(master, "#EXT-X-MEDIA:TYPE=AUDIO"), "the mix's two codecs, no stems:\n%s", master) + require.Contains(t, master, `DEFAULT=YES,AUTOSELECT=YES,CHANNELS="2",URI="track4.m3u8"`) + require.Contains(t, master, `DEFAULT=NO,AUTOSELECT=YES,CHANNELS="2",URI="track3.m3u8"`) + require.NotContains(t, master, "track2.m3u8") + require.NotContains(t, master, "track5.m3u8") + require.Contains(t, master, `CODECS="avc1.640028,mp4a.40.2"`) + require.NotEmpty(t, w.MediaPlaylist("2", "init", func(uint64) string { return "s.m4s" }), "the stem is still a served track") +} diff --git a/pkg/localdb/segment.go b/pkg/localdb/segment.go index 7859a1640..9564ad486 100644 --- a/pkg/localdb/segment.go +++ b/pkg/localdb/segment.go @@ -22,9 +22,16 @@ type SegmentMediadataVideo struct { BFrames bool `json:"bframes"` } +// SegmentMediadataAudio is one audio track. Codec is the lexicon's name +// ("aac", "opus"); AlternateGroup and Label are the track's group and name +// when the stream distinguishes its audio tracks (the OBS plugin's stems), +// zero and empty otherwise. type SegmentMediadataAudio struct { - Rate int `json:"rate"` - Channels int `json:"channels"` + Codec string `json:"codec,omitempty"` + Rate int `json:"rate"` + Channels int `json:"channels"` + AlternateGroup uint32 `json:"alternateGroup,omitempty"` + Label string `json:"label,omitempty"` } type SegmentMediaData struct { @@ -192,6 +199,30 @@ func (s *Segment) ToStreamplaceSegment() (*placestream.Segment, error) { } } + audio := make([]placestream.Segment_Audio, 0, len(s.MediaData.Audio)) + for _, a := range s.MediaData.Audio { + codec := a.Codec + if codec == "" { + // A row written before the codec was recorded: every such + // segment's audio was Opus. + codec = "opus" + } + track := placestream.Segment_Audio{ + Codec: codec, + Rate: int64(a.Rate), + Channels: int64(a.Channels), + } + if a.AlternateGroup != 0 { + group := int64(a.AlternateGroup) + track.AlternateGroup = &group + } + if a.Label != "" { + label := a.Label + track.Label = &label + } + audio = append(audio, track) + } + return &placestream.Segment{ LexiconTypeID: "place.stream.segment", Creator: s.RepoDID, @@ -215,13 +246,7 @@ func (s *Segment) ToStreamplaceSegment() (*placestream.Segment, error) { Bframes: &s.MediaData.Video[0].BFrames, }, }, - Audio: []placestream.Segment_Audio{ - { - Codec: "opus", - Rate: int64(s.MediaData.Audio[0].Rate), - Channels: int64(s.MediaData.Audio[0].Channels), - }, - }, + Audio: audio, }, nil } diff --git a/pkg/localdb/segment_audio_test.go b/pkg/localdb/segment_audio_test.go new file mode 100644 index 000000000..fd5a97a27 --- /dev/null +++ b/pkg/localdb/segment_audio_test.go @@ -0,0 +1,44 @@ +package localdb + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// The segment record lists every audio track with its real codec, and the +// group and label of a stream that distinguishes its tracks; a row written +// before the codec was recorded still reads as the Opus it was. +func TestToStreamplaceSegmentListsEveryAudioTrack(t *testing.T) { + seg := &Segment{ + ID: "seg", RepoDID: "did:plc:test", StartTime: time.Now(), + MediaData: &SegmentMediaData{ + Video: []*SegmentMediadataVideo{{Width: 1280, Height: 720, FPSNum: 30, FPSDen: 1}}, + Audio: []*SegmentMediadataAudio{ + {Codec: "aac", Rate: 48000, Channels: 2, AlternateGroup: 1, Label: "Mix"}, + {Codec: "opus", Rate: 48000, Channels: 2, AlternateGroup: 1, Label: "Mix"}, + {Codec: "opus", Rate: 48000, Channels: 2, AlternateGroup: 2, Label: "Music"}, + }, + }, + } + record, err := seg.ToStreamplaceSegment() + require.NoError(t, err) + require.Len(t, record.Audio, 3) + require.Equal(t, "aac", record.Audio[0].Codec) + require.Equal(t, "opus", record.Audio[2].Codec) + require.Equal(t, int64(48000), record.Audio[2].Rate) + require.Equal(t, int64(2), record.Audio[2].Channels) + require.NotNil(t, record.Audio[2].AlternateGroup) + require.Equal(t, int64(2), *record.Audio[2].AlternateGroup) + require.NotNil(t, record.Audio[2].Label) + require.Equal(t, "Music", *record.Audio[2].Label) + + seg.MediaData.Audio = []*SegmentMediadataAudio{{Rate: 48000, Channels: 2}} + record, err = seg.ToStreamplaceSegment() + require.NoError(t, err) + require.Len(t, record.Audio, 1) + require.Equal(t, "opus", record.Audio[0].Codec, "a legacy row's audio was Opus") + require.Nil(t, record.Audio[0].AlternateGroup, "an ungrouped track carries no group") + require.Nil(t, record.Audio[0].Label) +} diff --git a/pkg/media/audio_stems_test.go b/pkg/media/audio_stems_test.go new file mode 100644 index 000000000..c5d4b3c46 --- /dev/null +++ b/pkg/media/audio_stems_test.go @@ -0,0 +1,372 @@ +package media + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "strings" + "testing" + "time" + + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/bus" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/crypto/signers" + "stream.place/streamplace/pkg/ingestframe" + "stream.place/streamplace/pkg/livehls" + "stream.place/streamplace/pkg/muxl" +) + +// audioCfg is a catalog audio rendition at track id with codec and group. +func audioCfg(id uint32, codec string, group uint32, label string) muxl.MuxlAudioConfig { + return muxl.MuxlAudioConfig{ + Codec: codec, + Container: muxl.MuxlContainer{Kind: "cmaf", Timescale: 48000, TrackID: id}, + SampleRate: 48000, + NumberOfChannels: 2, + AlternateGroup: group, + Label: label, + } +} + +func catalogOf(audio ...muxl.MuxlAudioConfig) *muxl.MuxlCatalog { + cat := &muxl.MuxlCatalog{ + Video: &muxl.MuxlCatalogVideo{Renditions: map[string]muxl.MuxlVideoConfig{ + "v": {Codec: "avc1.640028", Container: muxl.MuxlContainer{Kind: "cmaf", Timescale: 90000, TrackID: 1}}, + }}, + Audio: &muxl.MuxlCatalogAudio{Renditions: map[string]muxl.MuxlAudioConfig{}}, + } + for i, a := range audio { + cat.Audio.Renditions[fmt.Sprint("a", i)] = a + } + return cat +} + +// The main audio is the lowest alternateGroup; an ungrouped segment is one +// group. Codec completion looks at the main group only: a stem missing a +// codec is distributed as sent, and a main group with both codecs needs +// nothing whatever the stems carry. +func TestMainAudioGroupSelectionAndCompletion(t *testing.T) { + ungroupedAAC := catalogOf(audioCfg(2, "mp4a.40.2", 0, "")) + ungroupedBoth := catalogOf(audioCfg(3, "opus", 0, ""), audioCfg(2, "mp4a.40.2", 0, "")) + grouped := catalogOf( + audioCfg(4, "mp4a.40.2", 2, "Music"), audioCfg(5, "opus", 2, "Music"), + audioCfg(2, "mp4a.40.2", 1, "Mix"), audioCfg(3, "opus", 1, "Mix"), + audioCfg(7, "opus", 3, "Mic"), audioCfg(8, "mp4a.40.2", 4, "Alerts"), audioCfg(9, "opus", 5, "Game"), + ) + mainAACOnly := catalogOf(audioCfg(2, "mp4a.40.2", 1, "Mix"), audioCfg(4, "mp4a.40.2", 2, "Music"), audioCfg(5, "opus", 2, "Music")) + stemAACOnly := catalogOf(audioCfg(2, "mp4a.40.2", 1, "Mix"), audioCfg(3, "opus", 1, "Mix"), audioCfg(4, "mp4a.40.2", 2, "Music")) + + ids := func(tracks []muxl.MuxlAudioConfig) []uint32 { + var out []uint32 + for _, a := range tracks { + out = append(out, a.TrackID()) + } + return out + } + require.Equal(t, []uint32{2, 3}, ids(mainAudioTracks(audioTracks(ungroupedBoth))), "ungrouped: every track is main, in track order") + require.Equal(t, []uint32{2, 3}, ids(mainAudioTracks(audioTracks(grouped))), "grouped: the lowest group is main") + require.Equal(t, []uint32{5, 7}, ids(stemOpusTracks(audioTracks(grouped))), "stems: one Opus track per higher group, ascending, at most two") + require.Empty(t, stemOpusTracks(audioTracks(ungroupedBoth)), "ungrouped: no stems") + require.Empty(t, stemOpusTracks(audioTracks(stemAACOnly)), "a stem without Opus can't reach WebRTC") + + for _, tc := range []struct { + name string + cat *muxl.MuxlCatalog + target string + need bool + }{ + {"ungrouped AAC only", ungroupedAAC, "opus", true}, + {"ungrouped both", ungroupedBoth, "", false}, + {"grouped, every group complete", grouped, "", false}, + {"grouped, main lacks Opus", mainAACOnly, "opus", true}, + {"grouped, only a stem lacks Opus", stemAACOnly, "", false}, + {"no video", &muxl.MuxlCatalog{Audio: ungroupedAAC.Audio}, "", false}, + {"no catalog", nil, "", false}, + } { + target, need := completionTarget(tc.cat) + require.Equal(t, tc.need, need, tc.name) + require.Equal(t, tc.target, target, tc.name) + } +} + +// groupedTestMeta labels the tracks groupedTestFMP4's muxer creates (video 1, +// then the audio pads in order) the way the OBS plugin labels a full mix on +// mixer Track 1 and a "Music" stem on Track 2, each as AAC and Opus. +var groupedTestMeta = map[uint32]muxl.TrackMeta{ + 2: {AlternateGroup: 1, Label: "Mix"}, + 3: {AlternateGroup: 1, Label: "Mix"}, + 4: {AlternateGroup: 2, Label: "Music"}, + 5: {AlternateGroup: 2, Label: "Music"}, +} + +// groupedTestFMP4 synthesizes two seconds of the plugin's stem shape: H264 +// video with a 1s GoP plus two AAC and two Opus tracks, as a fragmented MP4. +// The tracks carry no groups or names yet (mp4mux writes none). +func groupedTestFMP4(t *testing.T, ctx context.Context) []byte { + t.Helper() + return runSynthPipeline(t, ctx, strings.Join([]string{ + "videotestsrc num-buffers=60 ! video/x-raw,width=320,height=240,framerate=30/1 ! x264enc key-int-max=30 tune=zerolatency speed-preset=ultrafast ! h264parse ! mp4mux name=mux fragment-duration=500 ! appsink name=sink", + "audiotestsrc num-buffers=94 samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! fdkaacenc ! aacparse ! mux.", + "audiotestsrc num-buffers=100 samplesperbuffer=960 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! opusenc ! opusparse ! mux.", + "audiotestsrc wave=triangle num-buffers=94 samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! fdkaacenc ! aacparse ! mux.", + "audiotestsrc wave=triangle num-buffers=100 samplesperbuffer=960 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! opusenc ! opusparse ! mux.", + }, "\n")) +} + +// groupedTestSegment returns groupedTestFMP4's first GoP as an unsigned bare +// canonical .m4s whose catalogs carry groupedTestMeta. +func groupedTestSegment(t *testing.T, ctx context.Context) []byte { + t.Helper() + events, err := collectMuxlEvents(func(ch chan *muxl.MuxlEvent) error { + return muxl.RunMuxlSegmenterEvents(ctx, bytes.NewReader(groupedTestFMP4(t, ctx)), ch, muxl.WithSegmentTrackMeta(groupedTestMeta)) + }) + require.NoError(t, err) + for _, ev := range events { + if ev.Type == "segment" { + return concatTracksByID(ev.Tracks) + } + } + t.Fatal("segmenter emitted no segment") + return nil +} + +// requireGroupedCatalog checks that a segment's catalog carries +// groupedTestMeta: four audio tracks with the plugin's groups and names. +func requireGroupedCatalog(t *testing.T, ctx context.Context, seg []byte) { + t.Helper() + events, err := unwrapMuxlEvents(ctx, seg) + require.NoError(t, err) + cat, _ := catalogAndTracks(events) + tracks := audioTracks(cat) + require.Len(t, tracks, 4) + for _, a := range tracks { + want := groupedTestMeta[a.TrackID()] + require.Equal(t, want.AlternateGroup, a.AlternateGroup, "track %d group", a.TrackID()) + require.Equal(t, want.Label, a.Label, "track %d label", a.TrackID()) + } + require.True(t, isAACCodec(tracks[0].Codec) && isOpusCodec(tracks[1].Codec), "mix is AAC then Opus: %v", tracks) + require.True(t, isAACCodec(tracks[2].Codec) && isOpusCodec(tracks[3].Codec), "stem is AAC then Opus: %v", tracks) +} + +// The OBS plugin's push (fmp4-direct) is signed with its track metadata: the +// groups and names it wrote into the fMP4 reach every signed segment's +// catalog, on the detached-worker path. The segment record then lists every +// audio track with its codec, group and label. +func TestFMP4DirectSignsTrackMetaAndRecordsStems(t *testing.T) { + ctx := context.Background() + // The plugin's fMP4 carries tkhd.alternate_group and trak/udta/name; muxl + // writes exactly those from the meta. + labeled, err := muxl.RunMuxlCanonicalize(ctx, groupedTestFMP4(t, ctx), nil, muxl.WithTrackMeta(groupedTestMeta)) + require.NoError(t, err) + + ms := newBareSegmentSigner(t) + keyPEM, err := signers.MarshalES256KPrivateKeyPEM(ms.Signer) + require.NoError(t, err) + manifest, err := ms.buildManifest(ctx, time.Now().UnixMilli()) + require.NoError(t, err) + cfg := IngestWorkerConfig{ + StreamerDID: ms.Streamer(), + KeyPEM: keyPEM, + CertPEM: ms.Cert, + Manifest: manifest, + Transport: IngestTransportFMP4Direct, + } + var buf bytes.Buffer + require.NoError(t, RunMP4IngestWorker(ctx, cfg, bytes.NewReader(labeled), ingestframe.NewWriter(&buf), func() []byte { return cfg.Manifest })) + + r := ingestframe.NewReader(&buf) + var segs [][]byte + for { + _, payload, err := r.ReadFrame() + if errors.Is(err, io.EOF) { + break + } + require.NoError(t, err) + segs = append(segs, payload) + } + require.GreaterOrEqual(t, len(segs), 2, "a 2s stream with 1s GoPs signs into several segments") + for _, seg := range segs { + requireGroupedCatalog(t, ctx, seg) + } + + // Validation derives the record's audio list from the catalog, not from + // the one track the gst parse reads. + valid, err := ValidateMP4Media(ctx, segs[0]) + require.NoError(t, err) + require.NotNil(t, valid.Catalog) + audio := valid.MediaData.Audio + require.Len(t, audio, 4) + for i, want := range []struct { + codec string + group uint32 + label string + }{{"aac", 1, "Mix"}, {"opus", 1, "Mix"}, {"aac", 2, "Music"}, {"opus", 2, "Music"}} { + require.Equal(t, want.codec, audio[i].Codec, "track %d codec", i) + require.Equal(t, want.group, audio[i].AlternateGroup, "track %d group", i) + require.Equal(t, want.label, audio[i].Label, "track %d label", i) + require.Equal(t, 48000, audio[i].Rate, "track %d rate", i) + require.Equal(t, 2, audio[i].Channels, "track %d channels", i) + } + _, need := completionTarget(valid.Catalog) + require.False(t, need, "the mix has both codecs, so nothing is completed") +} + +// A segment with stems packetizes into the mix plus one Opus track per stem +// group, on the mix's clock; the codec filters keep the mix's track of the +// requested codec. An ungrouped segment packetizes exactly as before: no +// stems. +func TestPacketizeStems(t *testing.T) { + withNoGSTLeaks(t, func() { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + grouped := groupedTestSegment(t, ctx) + + aac, err := filterSegmentToCodec(ctx, grouped, false) + require.NoError(t, err) + require.Equal(t, []string{"mp4a.40.2"}, audioCodecsOf(t, ctx, aac), "RTMP/HLS get the mix's AAC only") + _, kept, err := filterSegmentAudio(ctx, grouped, true, true) + require.NoError(t, err) + require.Equal(t, []playbackAudio{{TrackID: 3, Group: 1, Main: true}, {TrackID: 5, Group: 2}}, kept) + + packet, err := Packetize(ctx, &config.CLI{}, &bus.Seg{Muxl: grouped}) + require.NoError(t, err) + require.Len(t, packet.Video, 30) + require.NotEmpty(t, packet.Audio, "the mix's Opus") + require.Len(t, packet.Stems, 1) + require.Equal(t, uint32(2), packet.Stems[0].Group) + require.InDelta(t, len(packet.Audio), len(packet.Stems[0].Audio), 1, "the stem spans the segment like the mix") + var stemDur time.Duration + for _, s := range packet.Stems[0].Audio { + stemDur += s.Duration + } + require.Equal(t, packet.Duration, stemDur, "the stem is stretched to the segment's duration") + + ungrouped := sourceSegmentForTest(t) + packet, err = Packetize(ctx, &config.CLI{}, &bus.Seg{Muxl: ungrouped}) + require.NoError(t, err) + require.NotEmpty(t, packet.Audio) + require.Empty(t, packet.Stems) + }) +} + +// A WebRTC viewer's answer carries the mix and one "audio-g" track per +// stem of the stream's latest segment, in that order, as many as the offer +// has audio transceivers for; without stems (or without a segment yet) the +// answer is the one audio track it always was. +func TestWebRTCPlayback2StemTracks(t *testing.T) { + mm, _ := getStaticTestMediaManager(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + offer := func(audioTransceivers int) *webrtc.SessionDescription { + pc, err := webrtc.NewPeerConnection(webrtc.Configuration{}) + require.NoError(t, err) + defer pc.Close() + _, err = pc.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionRecvonly}) + require.NoError(t, err) + for range audioTransceivers { + _, err = pc.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio, webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionRecvonly}) + require.NoError(t, err) + } + offer, err := pc.CreateOffer(nil) + require.NoError(t, err) + return &offer + } + audioTrackIDs := func(answer *webrtc.SessionDescription) []string { + parsed, err := answer.Unmarshal() + require.NoError(t, err) + var ids []string + // The answer has an m-line per offered audio transceiver; those + // without a track of ours carry no msid. + for _, md := range parsed.MediaDescriptions { + if msid, ok := md.Attribute("msid"); ok && md.MediaName.Media == "audio" { + ids = append(ids, strings.Fields(msid)[1]) + } + } + return ids + } + + answer, err := mm.WebRTCPlayback2(ctx, "streamer", "source", offer(3), "") + require.NoError(t, err) + require.Equal(t, []string{"audio"}, audioTrackIDs(answer), "no segment yet: the mix only") + + mm.bus.PublishSegment(ctx, "streamer", "source", &bus.Seg{Published: true, PacketizedData: &bus.PacketizedSegment{ + Duration: time.Second, + Stems: []bus.PacketizedStem{{Group: 2}, {Group: 3}}, + }}) + answer, err = mm.WebRTCPlayback2(ctx, "streamer", "source", offer(3), "") + require.NoError(t, err) + require.Equal(t, []string{"audio", "audio-g2", "audio-g3"}, audioTrackIDs(answer)) + + answer, err = mm.WebRTCPlayback2(ctx, "streamer", "source", offer(2), "") + require.NoError(t, err) + require.Equal(t, []string{"audio", "audio-g2"}, audioTrackIDs(answer), "an offer with fewer transceivers gets what fits") + + answer, err = mm.WebRTCPlayback2(ctx, "streamer", "source", offer(1), "") + require.NoError(t, err) + require.Equal(t, []string{"audio"}, audioTrackIDs(answer), "a client that can't take stems gets the mix") +} + +// groupedSignedSegment is groupedTestFMP4's first GoP signed the way the +// in-process fmp4-direct ingest signs the plugin's push: the groups and +// names in the fMP4 reach the signed catalogs. +func groupedSignedSegment(t *testing.T, ctx context.Context) []byte { + t.Helper() + labeled, err := muxl.RunMuxlCanonicalize(ctx, groupedTestFMP4(t, ctx), nil, muxl.WithTrackMeta(groupedTestMeta)) + require.NoError(t, err) + ms := newBareSegmentSigner(t) + eventCh := make(chan *muxl.MuxlEvent, 16) + errCh := make(chan error, 1) + go func() { + err := ms.SignSegmentStream(ctx, bytes.NewReader(labeled), eventCh, true) + close(eventCh) + errCh <- err + }() + var first []byte + for ev := range eventCh { + if ev.Type == "signed-segment" && first == nil { + first = concatTracksByID(ev.Tracks) + } + } + require.NoError(t, <-errCh) + requireGroupedCatalog(t, ctx, first) + return first +} + +// A syndicating node pairs a rendition with the source segment's Opus +// tracks: the mix and the stems, so a viewer who picked a rendition keeps +// the stems. +func TestPublishRenditionsForPlaybackCarriesStems(t *testing.T) { + ctx := context.Background() + src := groupedSignedSegment(t, ctx) + rendition := runSynthPipeline(t, ctx, + "videotestsrc num-buffers=30 ! video/x-raw,width=160,height=120,framerate=30/1 ! x264enc key-int-max=30 tune=zerolatency speed-preset=ultrafast ! h264parse ! mp4mux fragment-duration=1000 ! appsink name=sink") + cert, keyPEM := nodeSignerForTest(t) + b := bus.NewBus() + mm := &MediaManager{cli: &config.CLI{BroadcasterHost: "node.test"}, liveWindows: map[string]*livehls.Writer{}, bus: b} + addendum, err := mm.mintVideoRenditions(ctx, src, []RenditionInput{{Name: "120p", MP4: rendition}}, cert, keyPEM) + require.NoError(t, err) + require.NotEmpty(t, addendum) + + const did = "did:test:streamer" + sub := b.SubscribeSegment(ctx, did, "120p") + defer b.UnsubscribeSegment(ctx, did, "120p", sub) + mm.rememberSourceAudio(ctx, did, src) + mm.PublishRenditionsForPlayback(ctx, did, addendum, true) + select { + case seg := <-sub.C: + packet := seg.PacketizedData + require.NotNil(t, packet) + require.NotEmpty(t, packet.Video, "video samples") + require.NotEmpty(t, packet.Audio, "the mix's Opus") + require.Len(t, packet.Stems, 1) + require.Equal(t, uint32(2), packet.Stems[0].Group) + require.InDelta(t, len(packet.Audio), len(packet.Stems[0].Audio), 1) + case <-time.After(20 * time.Second): + t.Fatal("no rendition segment published") + } +} diff --git a/pkg/media/concat_demux.go b/pkg/media/concat_demux.go index 2c3495f8d..844b175f9 100644 --- a/pkg/media/concat_demux.go +++ b/pkg/media/concat_demux.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "strconv" "strings" "sync/atomic" @@ -20,10 +21,14 @@ func doNothing(self gst.Element, pad gst.Pad) {} // doNothing for signals without a pad argument func doNothingElement(self gst.Element) {} -// Function for demuxing a single segment. Needs to be handled very carefully. -// In particular: users of this MUST cancel the passed context when they're -// done with the bin. -func ConcatDemuxBin(ctx context.Context, seg *bus.Seg, doH264Parse bool) (gst.Bin, error) { +// ConcatDemuxBin demuxes a single segment (seg.Data, a presentation MP4) into +// a "video_0" pad and one "audio_" Opus pad per audio stream, k in order of +// the file's audio tracks, for audioTracks of them (at least one; a stream the +// segment lacks ends with EOS, one it has beyond audioTracks is dropped). +// Needs to be handled very carefully. In particular: users of this MUST +// cancel the passed context when they're done with the bin. +func ConcatDemuxBin(ctx context.Context, seg *bus.Seg, doH264Parse bool, audioTracks int) (gst.Bin, error) { + audioTracks = max(audioTracks, 1) ctx = log.WithLogValues(ctx, "func", "ConcatDemuxBin") bin := gst.NewBin("seg-demux-bin").(gst.Bin) @@ -69,48 +74,52 @@ func ConcatDemuxBin(ctx context.Context, seg *bus.Seg, doH264Parse bool) (gst.Bi return nil, fmt.Errorf("failed to add multiqueue to bin") } - opusparse := gst.ElementFactoryMakeWithProperties("opusparse", map[string]interface{}{ - "name": "concat-demux-opusparse", - "disable-passthrough": true, - }) - if opusparse == nil { - return nil, fmt.Errorf("failed to create opusparse element") - } - if !bin.Add(opusparse) { - return nil, fmt.Errorf("failed to add opusparse to bin") - } - opusparseSinkPad := opusparse.GetStaticPad("sink") - if opusparseSinkPad == nil { - return nil, fmt.Errorf("failed to get opusparse sink pad") - } - opusparseSrcPad := opusparse.GetStaticPad("src") - if opusparseSrcPad == nil { - return nil, fmt.Errorf("failed to get opusparse source pad") - } - mqVideoSink := mq.RequestPadSimple("sink_%u") if mqVideoSink == nil { return nil, fmt.Errorf("video sink pad not found") } - - mqAudioSink := mq.RequestPadSimple("sink_%u") - if mqAudioSink == nil { - return nil, fmt.Errorf("audio sink pad not found") - } - - mqVideoSrc := mq.GetStaticPad("src_0") + mqVideoSrc := mq.GetStaticPad(strings.Replace(mqVideoSink.GetName(), "sink", "src", 1)) if mqVideoSrc == nil { return nil, fmt.Errorf("video source pad not found") } - mqAudioSrc := mq.GetStaticPad("src_1") - if mqAudioSrc == nil { - return nil, fmt.Errorf("audio source pad not found") - } - - linked := mqAudioSrc.Link(opusparseSinkPad) - if linked != gst.PadLinkOK { - return nil, fmt.Errorf("failed to link opusparse sink pad to mq audio sink pad") + // One multiqueue lane + opusparse per audio stream, ghosted as audio_. + mqAudioSinks := make([]gst.Pad, audioTracks) + audioGhosts := make([]gst.Pad, audioTracks) + for k := range mqAudioSinks { + opusparse := gst.ElementFactoryMakeWithProperties("opusparse", map[string]interface{}{ + "name": fmt.Sprintf("concat-demux-opusparse-%d", k), + "disable-passthrough": true, + }) + if opusparse == nil { + return nil, fmt.Errorf("failed to create opusparse element") + } + if !bin.Add(opusparse) { + return nil, fmt.Errorf("failed to add opusparse to bin") + } + opusparseSinkPad := opusparse.GetStaticPad("sink") + if opusparseSinkPad == nil { + return nil, fmt.Errorf("failed to get opusparse sink pad") + } + opusparseSrcPad := opusparse.GetStaticPad("src") + if opusparseSrcPad == nil { + return nil, fmt.Errorf("failed to get opusparse source pad") + } + mqAudioSinks[k] = mq.RequestPadSimple("sink_%u") + if mqAudioSinks[k] == nil { + return nil, fmt.Errorf("audio sink pad not found") + } + mqAudioSrc := mq.GetStaticPad(strings.Replace(mqAudioSinks[k].GetName(), "sink", "src", 1)) + if mqAudioSrc == nil { + return nil, fmt.Errorf("audio source pad not found") + } + if linked := mqAudioSrc.Link(opusparseSinkPad); linked != gst.PadLinkOK { + return nil, fmt.Errorf("failed to link opusparse sink pad to mq audio sink pad") + } + audioGhosts[k] = gst.NewGhostPad(fmt.Sprintf("audio_%d", k), opusparseSrcPad) + if audioGhosts[k] == nil { + return nil, fmt.Errorf("failed to create audio ghost pad") + } } var videoGhost gst.Pad @@ -150,14 +159,9 @@ func ConcatDemuxBin(ctx context.Context, seg *bus.Seg, doH264Parse bool) (gst.Bi } } - audioGhost := gst.NewGhostPad("audio_0", opusparseSrcPad) - if audioGhost == nil { - return nil, fmt.Errorf("failed to create audio ghost pad") - } - - needed := 2 + needed := 1 + audioTracks videoLinked := false - audioLinked := false + audioLinked := make([]bool, audioTracks) // The signal handlers below are swapped for no-ops once they've done // their job (or the ctx ends, whichever comes first) so their closures @@ -170,20 +174,27 @@ func ConcatDemuxBin(ctx context.Context, seg *bus.Seg, doH264Parse bool) (gst.Bi var padAdded atomic.Value padAdded.Store(func(self gst.Element, pad gst.Pad) { var downstreamPad gst.Pad - if strings.HasPrefix(pad.GetName(), "video_") { + name := pad.GetName() + if strings.HasPrefix(name, "video_") { downstreamPad = mqVideoSink videoLinked = true - } else if strings.HasPrefix(pad.GetName(), "audio_") { - downstreamPad = mqAudioSink - audioLinked = true + } else if k, ok := strings.CutPrefix(name, "audio_"); ok { + // qtdemux numbers its audio pads in the file's track order. + idx, err := strconv.Atoi(k) + if err != nil || idx >= audioTracks { + log.Warn(ctx, "dropping audio stream beyond the requested tracks", "name", name) + return + } + downstreamPad = mqAudioSinks[idx] + audioLinked[idx] = true } else { - log.Error(ctx, "unknown pad", "name", pad.GetName(), "direction", pad.GetDirection()) + log.Error(ctx, "unknown pad", "name", name, "direction", pad.GetDirection()) // cancel() return } ret := pad.Link(downstreamPad) if ret != gst.PadLinkOK { - log.Error(ctx, "failed to link demux to downstream pad", "name", pad.GetName(), "direction", pad.GetDirection(), "error", ret) + log.Error(ctx, "failed to link demux to downstream pad", "name", name, "direction", pad.GetDirection(), "error", ret) // cancel() return } @@ -212,9 +223,11 @@ func ConcatDemuxBin(ctx context.Context, seg *bus.Seg, doH264Parse bool) (gst.Bi log.Warn(ctx, "segment has no video track; completing video branch with EOS") mqVideoSink.SendEvent(gst.NewEventEOS()) } - if !audioLinked { - log.Warn(ctx, "segment has no audio track; completing audio branch with EOS") - mqAudioSink.SendEvent(gst.NewEventEOS()) + for k, linked := range audioLinked { + if !linked { + log.Warn(ctx, "segment has no audio track; completing audio branch with EOS", "audio", k) + mqAudioSinks[k].SendEvent(gst.NewEventEOS()) + } } noMorePads.Store(doNothingElement) }) @@ -238,9 +251,10 @@ func ConcatDemuxBin(ctx context.Context, seg *bus.Seg, doH264Parse bool) (gst.Bi return nil, fmt.Errorf("failed to add video ghost pad to bin") } - ok = bin.AddPad(audioGhost) - if !ok { - return nil, fmt.Errorf("failed to add audio ghost pad to bin") + for _, ghost := range audioGhosts { + if ok := bin.AddPad(ghost); !ok { + return nil, fmt.Errorf("failed to add audio ghost pad to bin") + } } src := appSrc.(app.AppSrc) diff --git a/pkg/media/concat_demux_test.go b/pkg/media/concat_demux_test.go index d2d29f6cd..27c6e75a6 100644 --- a/pkg/media/concat_demux_test.go +++ b/pkg/media/concat_demux_test.go @@ -61,7 +61,7 @@ func innerTestConcatDemuxBin(t *testing.T) error { Filepath: filename, } - concatBin, err := ConcatDemuxBin(ctx, testSeg, true) + concatBin, err := ConcatDemuxBin(ctx, testSeg, true, 1) if err != nil { return fmt.Errorf("failed to create concat bin: %w", err) } diff --git a/pkg/media/dynamic_manifest_test.go b/pkg/media/dynamic_manifest_test.go index 291abf2e0..8cd38389a 100644 --- a/pkg/media/dynamic_manifest_test.go +++ b/pkg/media/dynamic_manifest_test.go @@ -79,7 +79,7 @@ func TestSignSegmentStreamRefreshesManifestPerSegment(t *testing.T) { eventCh := make(chan *muxl.MuxlEvent, 16) errCh := make(chan error, 1) go func() { - err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh) + err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh, false) close(eventCh) errCh <- err }() diff --git a/pkg/media/ingest_worker.go b/pkg/media/ingest_worker.go index 694b8720c..89027c266 100644 --- a/pkg/media/ingest_worker.go +++ b/pkg/media/ingest_worker.go @@ -157,7 +157,8 @@ func WorkerInput(cfg IngestWorkerConfig, raw io.Reader) io.Reader { // needed. The manifest is read FRESH per GoP from the holder, so a manifest main // pushes mid-stream (e.g. pre-live → live) takes effect on the next GoP — the // same fresh-per-GoP shape as the in-process signer. Shared by the MP4 and WHIP -// workers. +// workers. An fMP4 signed as-is (the OBS plugin's push) keeps the audio groups +// and track names the plugin wrote; the demuxed transports leave them alone. func workerSignStream(cfg IngestWorkerConfig, getManifest func() []byte) SignSegmentStreamFunc { return func(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent) error { fetchManifest := func() ([]byte, error) { return getManifest(), nil } @@ -166,6 +167,7 @@ func workerSignStream(cfg IngestWorkerConfig, getManifest func() []byte) SignSeg KeyPEM: cfg.KeyPEM, TrackManifestFn: fetchManifest, WrapperManifestFn: fetchManifest, + SourceTrackMeta: cfg.Transport == IngestTransportFMP4Direct, }, nil, nil, eventCh) } } diff --git a/pkg/media/media_data_parser.go b/pkg/media/media_data_parser.go index eaa50067b..5a2782435 100644 --- a/pkg/media/media_data_parser.go +++ b/pkg/media/media_data_parser.go @@ -16,6 +16,7 @@ import ( "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/localdb" "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/muxl" ) func padProbeEmpty(_ gst.Pad, _ *gst.PadProbeInfo) gst.PadProbeReturn { @@ -279,6 +280,32 @@ func ParseSegmentMediaData(ctx context.Context, mp4bs []byte) (*localdb.SegmentM return meta, nil } +// audioMediaData describes a catalog's audio tracks for the segment record: +// every track, in track-id order, with the lexicon's codec name ("aac", +// "opus") and its alternateGroup and label when the stream carries them. The +// gst parse in ParseSegmentMediaData reads only the first audio track, so +// this is what a segment with stems lists them from. +func audioMediaData(cat *muxl.MuxlCatalog) []*localdb.SegmentMediadataAudio { + var out []*localdb.SegmentMediadataAudio + for _, a := range audioTracks(cat) { + codec := a.Codec + switch { + case isAACCodec(codec): + codec = "aac" + case isOpusCodec(codec): + codec = "opus" + } + out = append(out, &localdb.SegmentMediadataAudio{ + Codec: codec, + Rate: int(a.SampleRate), + Channels: int(a.NumberOfChannels), + AlternateGroup: a.AlternateGroup, + Label: a.Label, + }) + } + return out +} + func ParseSegmentMediaDataSinkNewSampleFunc(ctx context.Context, foundThisTrack *bool, duration *time.Duration) func(sink app.AppSink) gst.FlowReturn { var firstPTS gst.ClockTime haveFirstPTS := false diff --git a/pkg/media/media_signer.go b/pkg/media/media_signer.go index 8468d0888..57db4f2a3 100644 --- a/pkg/media/media_signer.go +++ b/pkg/media/media_signer.go @@ -33,7 +33,10 @@ type MediaSigner interface { // signer, emitting one signed-segment event per GoP on eventCh. The // segment bytes routed in each event are bare canonical .m4s // ([c2pa-uuid][muxl-uuid][moof][mdat] per track) — no flat wrapper. - SignSegmentStream(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent) error + // sourceTrackMeta carries the input's audio alternate groups and track + // names into the signed catalogs (the OBS plugin's stems); other inputs + // leave it off, since muxers disagree on what those fields mean. + SignSegmentStream(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent, sourceTrackMeta bool) error Pub() aqpub.Pub Streamer() string DID() string @@ -199,7 +202,7 @@ func (ms *MediaSignerLocal) buildManifest(ctx context.Context, start int64) ([]b // it signs. Signing backend: an *ecdsa.PrivateKey is marshaled to PEM and // signed in-wasm, otherwise the host-callback path keeps the key out of the // sandbox. -func (ms *MediaSignerLocal) SignSegmentStream(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent) error { +func (ms *MediaSignerLocal) SignSegmentStream(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent, sourceTrackMeta bool) error { ctx, span := signerTracer.Start(ctx, "SignSegmentStream", trace.WithAttributes( attribute.String("streamer", ms.StreamerName), )) @@ -215,6 +218,7 @@ func (ms *MediaSignerLocal) SignSegmentStream(ctx context.Context, input io.Read CertPEM: ms.Cert, TrackManifestFn: fetchManifest, WrapperManifestFn: fetchManifest, + SourceTrackMeta: sourceTrackMeta, } if _, ok := ms.Signer.(*ecdsa.PrivateKey); ok { keyPEM, err := signers.MarshalES256KPrivateKeyPEM(ms.Signer) diff --git a/pkg/media/mp4_ingest.go b/pkg/media/mp4_ingest.go index 19b40d7ad..c603a6dec 100644 --- a/pkg/media/mp4_ingest.go +++ b/pkg/media/mp4_ingest.go @@ -15,6 +15,7 @@ import ( "stream.place/streamplace/pkg/aqtime" "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/muxl" ) // ingest a fragmented-MP4 stream: H264+AAC through the demux pipeline (the @@ -41,7 +42,12 @@ func (mm *MediaManager) MP4Ingest(ctx context.Context, input io.Reader, ms Media cancel(fmt.Errorf("ending stream: %s", reason)) }) ctx, onSegment := mm.validatingSegmentSink(ctx, ms) - return signFMP4Direct(ctx, mm.cli, ms.SignSegmentStream, input, onSegment) + // The plugin labels its stems (tkhd.alternate_group, trak/udta/name); + // signing as-is carries them into the segments' catalogs. + signStream := func(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent) error { + return ms.SignSegmentStream(ctx, input, eventCh, true) + } + return signFMP4Direct(ctx, mm.cli, signStream, input, onSegment) } signer, err := mm.SegmentAndSignElem(ctx, ms) diff --git a/pkg/media/muxl_segment.go b/pkg/media/muxl_segment.go index d39b85ee7..595818de0 100644 --- a/pkg/media/muxl_segment.go +++ b/pkg/media/muxl_segment.go @@ -29,7 +29,10 @@ type SignSegmentStreamFunc func(ctx context.Context, input io.Reader, eventCh ch // produced here. Presentation headers are synthesized downstream (ValidateMP4 // / playback) only when needed. func MuxlSignSegmentElem(ctx context.Context, cli *config.CLI, ms MediaSigner, onSegment func(ctx context.Context, segment []byte) error) (gst.Element, error) { - elem, _, err := muxlSignSegmentElem(ctx, cli, ms.SignSegmentStream, onSegment) + signStream := func(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent) error { + return ms.SignSegmentStream(ctx, input, eventCh, false) + } + elem, _, err := muxlSignSegmentElem(ctx, cli, signStream, onSegment) return elem, err } diff --git a/pkg/media/packetize.go b/pkg/media/packetize.go index 79231abc3..3ab949257 100644 --- a/pkg/media/packetize.go +++ b/pkg/media/packetize.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "sort" "strings" "time" @@ -43,7 +44,9 @@ func hasVideoSlice(data []byte) bool { return false } -// take in a segment and return a bunch of packets suitable for webrtc +// Packetize demuxes a segment into WebRTC samples: the video, the main audio +// (the full mix) and, for a segment with stems, one Opus track per stem group +// (see filterSegmentAudio), all on the segment's one timeline. func Packetize(ctx context.Context, cli *config.CLI, seg *bus.Seg) (*bus.PacketizedSegment, error) { uu, err := uuid.NewV7() @@ -55,11 +58,14 @@ func Packetize(ctx context.Context, cli *config.CLI, seg *bus.Seg) (*bus.Packeti ctx = log.WithLogValues(ctx, "func", "Packetize", "uuid", uu.String()) - // WebRTC playback needs Opus. From a dual-codec segment select video+Opus - // and present it as a flat MP4, so the demux below yields exactly one - // (Opus) audio pad for opusparse — no extra AAC pad to strand. + // WebRTC playback needs Opus. From a multi-track segment select video, the + // main group's Opus track and each stem's, and present them as a flat MP4, + // so the demux below yields exactly one Opus pad per kept track for + // opusparse — no extra AAC pad to strand. A presentation without muxl + // bytes (a rendition, a fixture) has one audio stream: the mix. + audio := []playbackAudio{{Main: true}} if len(seg.Muxl) > 0 { - opusM4s, err := filterSegmentToCodec(ctx, seg.Muxl, true) + opusM4s, kept, err := filterSegmentAudio(ctx, seg.Muxl, true, true) if err != nil { return nil, fmt.Errorf("select opus audio: %w", err) } @@ -68,13 +74,18 @@ func Packetize(ctx context.Context, cli *config.CLI, seg *bus.Seg) (*bus.Packeti return nil, fmt.Errorf("wrap opus segment: %w", err) } seg = &bus.Seg{Filepath: seg.Filepath, Data: flat.Bytes(), Muxl: opusM4s} + if len(kept) > 0 { + audio = kept + } } cli.DumpDebugSegment(ctx, fmt.Sprintf("packetize-input-%s.mp4", uu.String()), bytes.NewReader(seg.Data)) pipelineSlice := []string{ fmt.Sprintf("%s name=videoparse ! h264parse ! video/x-h264,stream-format=byte-stream ! appsink sync=false name=videoappsink", constants.Queue2Big), - fmt.Sprintf("%s name=audioparse ! opusparse ! appsink sync=false name=audioappsink", constants.Queue2Big), + } + for k := range audio { + pipelineSlice = append(pipelineSlice, fmt.Sprintf("%s name=audioparse%d ! opusparse ! appsink sync=false name=audioappsink%d", constants.Queue2Big, k, k)) } pipelineElem, err := gst.ParseLaunch(strings.Join(pipelineSlice, "\n")) @@ -86,7 +97,7 @@ func Packetize(ctx context.Context, cli *config.CLI, seg *bus.Seg) (*bus.Packeti return nil, fmt.Errorf("failed to create GStreamer pipeline: not a pipeline") } - demuxBin, err := ConcatDemuxBin(ctx, seg, true) + demuxBin, err := ConcatDemuxBin(ctx, seg, true, len(audio)) if err != nil { return nil, fmt.Errorf("failed to create concat bin: %w", err) } @@ -95,34 +106,28 @@ func Packetize(ctx context.Context, cli *config.CLI, seg *bus.Seg) (*bus.Packeti return nil, fmt.Errorf("failed to add demux bin to bin") } - demuxBinPadVideoSrc := demuxBin.GetStaticPad("video_0") - if demuxBinPadVideoSrc == nil { - return nil, fmt.Errorf("failed to get demux bin video src pad") - } - - demuxBinPadAudioSrc := demuxBin.GetStaticPad("audio_0") - if demuxBinPadAudioSrc == nil { - return nil, fmt.Errorf("failed to get demux bin audio src pad") - } - - videoParse := pipeline.GetByName("videoparse") - if videoParse == nil { - return nil, fmt.Errorf("failed to get video parse element") - } - - audioParse := pipeline.GetByName("audioparse") - if audioParse == nil { - return nil, fmt.Errorf("failed to get audio parse element") + // link connects the demux bin's src pad to the named parser's sink. + link := func(pad, parser string) error { + src := demuxBin.GetStaticPad(pad) + if src == nil { + return fmt.Errorf("failed to get demux bin %s src pad", pad) + } + parse := pipeline.GetByName(parser) + if parse == nil { + return fmt.Errorf("failed to get %s element", parser) + } + if linked := src.Link(parse.GetStaticPad("sink")); linked != gst.PadLinkOK { + return fmt.Errorf("failed to link demux bin %s src pad to %s: %v", pad, parser, linked) + } + return nil } - - linked := demuxBinPadVideoSrc.Link(videoParse.GetStaticPad("sink")) - if linked != gst.PadLinkOK { - return nil, fmt.Errorf("failed to link demux bin video src pad to video parse element: %v", linked) + if err := link("video_0", "videoparse"); err != nil { + return nil, err } - - linked = demuxBinPadAudioSrc.Link(audioParse.GetStaticPad("sink")) - if linked != gst.PadLinkOK { - return nil, fmt.Errorf("failed to link demux bin audio src pad to audio parse element: %v", linked) + for k := range audio { + if err := link(fmt.Sprintf("audio_%d", k), fmt.Sprintf("audioparse%d", k)); err != nil { + return nil, err + } } videoSink := pipeline.GetByName("videoappsink") @@ -130,13 +135,8 @@ func Packetize(ctx context.Context, cli *config.CLI, seg *bus.Seg) (*bus.Packeti return nil, fmt.Errorf("failed to get video appsink element") } - audioSink := pipeline.GetByName("audioappsink") - if audioSink == nil { - return nil, fmt.Errorf("failed to get audio appsink element") - } - videoOutput := []rawSample{} - audioOutput := []rawSample{} + audioOutput := make([][]rawSample, len(audio)) // eosCh := make(chan struct{}) // Slice-less video data (parameter sets / SEI metadata with no picture) @@ -194,42 +194,50 @@ func Packetize(ctx context.Context, cli *config.CLI, seg *bus.Seg) (*bus.Packeti segDur := time.Duration(0) - audioSink.(app.AppSink).SetCallbacks(&app.AppSinkCallbacks{ - NewSample: func(sink app.AppSink) gst.FlowReturn { - sample := sink.PullSample() - if sample == nil { - log.Warn(ctx, "audioappsink NewSampleFunc EOS") - return gst.FlowEOS - } - - buffer := sample.GetBuffer() - if buffer == nil { - return gst.FlowError - } - - samples, err := copyBufferData(buffer) - if err != nil { - log.Error(ctx, "failed to read audio buffer", "error", err) - return gst.FlowError - } - // log.Warn(ctx, "audioappsink NewSampleFunc", "sample", len(samples)) - - audioOutput = append(audioOutput, newRawSample(samples, buffer)) - - dur, ok := clockTimeDuration(buffer.Duration()) - if ok { - segDur += dur - } else { - log.Error(ctx, "no audio duration", "samples", len(samples)) - return gst.FlowError - } + for k, a := range audio { + audioSink := pipeline.GetByName(fmt.Sprintf("audioappsink%d", k)) + if audioSink == nil { + return nil, fmt.Errorf("failed to get audio appsink element") + } + main := a.Main + audioSink.(app.AppSink).SetCallbacks(&app.AppSinkCallbacks{ + NewSample: func(sink app.AppSink) gst.FlowReturn { + sample := sink.PullSample() + if sample == nil { + log.Warn(ctx, "audioappsink NewSampleFunc EOS", "audio", k) + return gst.FlowEOS + } + + buffer := sample.GetBuffer() + if buffer == nil { + return gst.FlowError + } + + samples, err := copyBufferData(buffer) + if err != nil { + log.Error(ctx, "failed to read audio buffer", "error", err) + return gst.FlowError + } + audioOutput[k] = append(audioOutput[k], newRawSample(samples, buffer)) + if !main { + return gst.FlowOK // the stems ride on the main audio's clock + } + + dur, ok := clockTimeDuration(buffer.Duration()) + if ok { + segDur += dur + } else { + log.Error(ctx, "no audio duration", "samples", len(samples)) + return gst.FlowError + } - return gst.FlowOK - }, - EOS: func(sink app.AppSink) { - log.Debug(ctx, "audioappsink EOSFunc") - }, - }) + return gst.FlowOK + }, + EOS: func(sink app.AppSink) { + log.Debug(ctx, "audioappsink EOSFunc", "audio", k) + }, + }) + } busErr := make(chan error) go func() { @@ -268,11 +276,17 @@ func Packetize(ctx context.Context, cli *config.CLI, seg *bus.Seg) (*bus.Packeti duration += s.Duration } } - return &bus.PacketizedSegment{ - Video: video, - Audio: finalizeSampleDurations(audioOutput, segDur), - Duration: duration, - }, nil + packet := &bus.PacketizedSegment{Video: video, Duration: duration} + for k, a := range audio { + samples := finalizeSampleDurations(audioOutput[k], segDur) + if a.Main { + packet.Audio = samples + } else { + packet.Stems = append(packet.Stems, bus.PacketizedStem{Group: a.Group, Audio: samples}) + } + } + sort.Slice(packet.Stems, func(i, j int) bool { return packet.Stems[i].Group < packet.Stems[j].Group }) + return packet, nil } // rawSample is a demuxed sample plus the source timing needed to compute its diff --git a/pkg/media/rendition_playback.go b/pkg/media/rendition_playback.go index 648cb8a80..1a89ec13d 100644 --- a/pkg/media/rendition_playback.go +++ b/pkg/media/rendition_playback.go @@ -1,14 +1,12 @@ package media import ( - "bytes" "context" "sort" "strconv" "stream.place/streamplace/pkg/bus" "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/muxl" ) // sourceAudioKeep is how many recent source segments' Opus tracks a node @@ -17,12 +15,14 @@ import ( const sourceAudioKeep = 8 type sourceAudio struct { - tfdt uint64 // the source video track's first decode time - opus []byte // the Opus track's canonical bytes + tfdt uint64 // the source video track's first decode time + opus map[string][]byte // the main group's Opus track and each stem's, canonical bytes by track id } -// rememberSourceAudio keeps a replicated source segment's Opus track, keyed -// by its video track's first decode time, for PublishRenditionsForPlayback. +// rememberSourceAudio keeps a replicated source segment's Opus tracks — the +// main group's and each stem's, so a rendition viewer gets the stems too — +// keyed by its video track's first decode time, for +// PublishRenditionsForPlayback. func (mm *MediaManager) rememberSourceAudio(ctx context.Context, did string, seg []byte) { events, err := unwrapMuxlEvents(ctx, seg) if err != nil { @@ -33,18 +33,33 @@ func (mm *MediaManager) rememberSourceAudio(ctx context.Context, did string, seg if cat == nil || cat.Video == nil || cat.Audio == nil || tracks == nil { return } - var videoID, opusID string + var videoID string for _, v := range cat.Video.Renditions { videoID = strconv.FormatUint(uint64(v.TrackID()), 10) break } - for _, a := range cat.Audio.Renditions { + all := audioTracks(cat) + var opusIDs []uint32 + for _, a := range mainAudioTracks(all) { if isOpusCodec(a.Codec) { - opusID = strconv.FormatUint(uint64(a.TrackID()), 10) + opusIDs = append(opusIDs, a.TrackID()) break } } - if videoID == "" || opusID == "" || len(tracks[opusID]) == 0 { + if videoID == "" || len(opusIDs) == 0 { + return + } + for _, a := range stemOpusTracks(all) { + opusIDs = append(opusIDs, a.TrackID()) + } + opus := map[string][]byte{} + for _, id := range opusIDs { + key := strconv.FormatUint(uint64(id), 10) + if len(tracks[key]) > 0 { + opus[key] = tracks[key] + } + } + if len(opus) == 0 { return } tfdt, ok := firstTfdt(tracks[videoID]) @@ -65,13 +80,13 @@ func (mm *MediaManager) rememberSourceAudio(ctx context.Context, did string, seg replaced := false for i := range list { if list[i].tfdt == tfdt { - list[i].opus = tracks[opusID] + list[i].opus = opus replaced = true break } } if !replaced { - list = append(list, sourceAudio{tfdt: tfdt, opus: tracks[opusID]}) + list = append(list, sourceAudio{tfdt: tfdt, opus: opus}) } if len(list) > sourceAudioKeep { list = list[len(list)-sourceAudioKeep:] @@ -79,7 +94,7 @@ func (mm *MediaManager) rememberSourceAudio(ctx context.Context, did string, seg mm.sourceAudios[did] = list } -func (mm *MediaManager) sourceAudioAt(did string, tfdt uint64) []byte { +func (mm *MediaManager) sourceAudioAt(did string, tfdt uint64) map[string][]byte { mm.sourceAudioMu.Lock() defer mm.sourceAudioMu.Unlock() list := mm.sourceAudios[did] @@ -95,11 +110,10 @@ func (mm *MediaManager) sourceAudioAt(did string, tfdt uint64) []byte { // MintVideoRenditions) playable over WebRTC on this node: each rendition, // paired with the Opus audio of the source segment it was transcoded from // (the renditions are re-timed onto the source video's decode time, which -// is the pairing key), is presented, packetized and published on the bus -// under the rendition's name ("720p"), as the transcoding node publishes -// its own. A node that only syndicates the stream received the renditions -// for HLS but had nothing on its bus, so a WebRTC viewer who picked one -// got no video. +// is the pairing key), is packetized and published on the bus under the +// rendition's name ("720p"), as the transcoding node publishes its own. A +// node that only syndicates the stream received the renditions for HLS but +// had nothing on its bus, so a WebRTC viewer who picked one got no video. func (mm *MediaManager) PublishRenditionsForPlayback(ctx context.Context, did string, addendum []byte, published bool) { if mm.bus == nil || len(addendum) == 0 { return @@ -138,18 +152,18 @@ func (mm *MediaManager) PublishRenditionsForPlayback(ctx context.Context, did st continue } name := names[id] - var flat bytes.Buffer - if err := muxl.RunMuxlWrap(ctx, bytes.NewReader(append(append([]byte{}, video...), opus...)), "flat", &flat); err != nil { - log.Warn(ctx, "rendition playback: could not present rendition", "streamer", did, "rendition", name, "error", err) - continue + // The rendition's video plus the source's Opus tracks, in track-id + // order, is a bare canonical segment, which Packetize presents and + // selects from (the mix and the stems) like the source's own. + m4s := map[string][]byte{id: video} + for key, track := range opus { + m4s[key] = track } - seg := &bus.Seg{Data: flat.Bytes(), Published: published} - packet, err := Packetize(ctx, mm.cli, seg) + packet, err := Packetize(ctx, mm.cli, &bus.Seg{Muxl: concatTracksByID(m4s)}) if err != nil { log.Warn(ctx, "rendition playback: could not packetize rendition", "streamer", did, "rendition", name, "error", err) continue } - seg.PacketizedData = packet - mm.bus.PublishSegment(ctx, did, name, seg) + mm.bus.PublishSegment(ctx, did, name, &bus.Seg{PacketizedData: packet, Published: published}) } } diff --git a/pkg/media/rtmp_ingest.go b/pkg/media/rtmp_ingest.go index 2e5fb1d80..d022d90b6 100644 --- a/pkg/media/rtmp_ingest.go +++ b/pkg/media/rtmp_ingest.go @@ -68,7 +68,7 @@ func RTMPIngestUnpublished(ctx context.Context, cli *config.CLI, rtmpURL string, } } signStream := func(c context.Context, input io.Reader, events chan *muxl.MuxlEvent) error { - err := ms.SignSegmentStream(c, input, events) + err := ms.SignSegmentStream(c, input, events, false) fail(err) return err } diff --git a/pkg/media/thumbnail.go b/pkg/media/thumbnail.go index a7b6cb349..756435117 100644 --- a/pkg/media/thumbnail.go +++ b/pkg/media/thumbnail.go @@ -185,7 +185,7 @@ func Thumbnail(ctx context.Context, r io.Reader, w io.Writer, format string) err return fmt.Errorf("error creating Thumbnail pipeline: parsed description is not a pipeline") } - demuxBin, err := ConcatDemuxBin(ctx, seg, false) + demuxBin, err := ConcatDemuxBin(ctx, seg, false, 1) if err != nil { return fmt.Errorf("failed to create demux bin: %w", err) } diff --git a/pkg/media/transcode.go b/pkg/media/transcode.go index cdee7b7ac..f165f2ef3 100644 --- a/pkg/media/transcode.go +++ b/pkg/media/transcode.go @@ -22,6 +22,87 @@ import ( func isAACCodec(codec string) bool { return strings.HasPrefix(codec, "mp4a") } func isOpusCodec(codec string) bool { return strings.HasPrefix(codec, "opus") } +// Audio track selection. A segment's audio tracks are grouped by their +// catalog alternateGroup: tracks sharing a group carry the same sound in +// different codecs (the AAC and Opus encodes of one mix), and different +// groups are independent sounds — the OBS plugin's full mix and its stems +// ("Music"). Tracks with no group (0) all share one group, which is every +// segment that isn't the plugin's. +// +// The main audio is the lowest group. Every consumer that wants one audio +// track — HLS, RTMP push, VOD, thumbnails, the WebRTC mix, codec completion — +// picks from the main group by codec; the stems reach WebRTC viewers only +// (see Packetize), who mix them at their own volumes. + +// maxStems bounds how many stem groups a segment contributes to playback: +// the plugin sends at most two (six encoders per output, two per track). +const maxStems = 2 + +// audioTracks returns a catalog's audio renditions in track-id order. +func audioTracks(cat *muxl.MuxlCatalog) []muxl.MuxlAudioConfig { + if cat == nil || cat.Audio == nil { + return nil + } + out := make([]muxl.MuxlAudioConfig, 0, len(cat.Audio.Renditions)) + for _, a := range cat.Audio.Renditions { + out = append(out, a) + } + sort.Slice(out, func(i, j int) bool { return out[i].TrackID() < out[j].TrackID() }) + return out +} + +// mainAudioGroup is the lowest alternateGroup among tracks: 0 for an +// ungrouped segment. +func mainAudioGroup(tracks []muxl.MuxlAudioConfig) uint32 { + var g uint32 + for i, a := range tracks { + if i == 0 || a.AlternateGroup < g { + g = a.AlternateGroup + } + } + return g +} + +// mainAudioTracks returns the main group's tracks (see mainAudioGroup), in +// track-id order. +func mainAudioTracks(tracks []muxl.MuxlAudioConfig) []muxl.MuxlAudioConfig { + g := mainAudioGroup(tracks) + var out []muxl.MuxlAudioConfig + for _, a := range tracks { + if a.AlternateGroup == g { + out = append(out, a) + } + } + return out +} + +// stemOpusTracks returns one Opus track per stem group — the groups above the +// main one, ascending, at most maxStems — in that order. WebRTC carries Opus +// only, so a stem group without an Opus encode is skipped. +func stemOpusTracks(tracks []muxl.MuxlAudioConfig) []muxl.MuxlAudioConfig { + main := mainAudioGroup(tracks) + byGroup := map[uint32]muxl.MuxlAudioConfig{} + var groups []uint32 + for _, a := range tracks { + if a.AlternateGroup <= main || !isOpusCodec(a.Codec) { + continue + } + if _, seen := byGroup[a.AlternateGroup]; !seen { + groups = append(groups, a.AlternateGroup) + byGroup[a.AlternateGroup] = a + } + } + sort.Slice(groups, func(i, j int) bool { return groups[i] < groups[j] }) + if len(groups) > maxStems { + groups = groups[:maxStems] + } + out := make([]muxl.MuxlAudioConfig, 0, len(groups)) + for _, g := range groups { + out = append(out, byGroup[g]) + } + return out +} + // transcodeManifest is the C2PA manifest for a transcode-completed audio // track: it opens the source segment as an ingredient and records a // c2pa.transcoded action referencing it, both by TranscodeIngredientLabel. @@ -90,11 +171,20 @@ func (mm *MediaManager) audioCompletionTarget(ctx context.Context, seg []byte) ( return "", false } cat, _ := catalogAndTracks(events) - if cat == nil || cat.Audio == nil || cat.Video == nil { + return completionTarget(cat) +} + +// completionTarget is audioCompletionTarget over an already-read catalog. +// Only the main audio group is completed: a stem (a higher group) is +// distributed as sent, since the plugin that labels stems encodes both codecs +// for each, and no HLS or RTMP consumer ever picks a stem. A segment whose +// main group already has both codecs needs nothing, whatever its stems carry. +func completionTarget(cat *muxl.MuxlCatalog) (target string, need bool) { + if cat == nil || cat.Video == nil { return "", false } var haveAAC, haveOpus bool - for _, a := range cat.Audio.Renditions { + for _, a := range mainAudioTracks(audioTracks(cat)) { if isAACCodec(a.Codec) { haveAAC = true } @@ -400,18 +490,36 @@ func concatTrackAcrossSegments(events []*muxl.MuxlEvent, tid uint32) []byte { // filterSegmentToCodec returns a bare canonical .m4s containing every video // track plus the single audio track matching the requested codec (Opus when // wantOpus, else AAC) — how an output consumer "asks for the audio it needs" -// from a dual-codec segment. Track bytes are carried verbatim (signatures -// intact) in ascending track-id order. If no audio matches the requested -// codec, any one audio track is kept (degraded but playable); with no audio -// info at all, the segment is returned unchanged. +// from a dual-codec segment. See filterSegmentAudio. func filterSegmentToCodec(ctx context.Context, seg []byte, wantOpus bool) ([]byte, error) { + out, _, err := filterSegmentAudio(ctx, seg, wantOpus, false) + return out, err +} + +// playbackAudio describes one audio track kept by filterSegmentAudio. +type playbackAudio struct { + TrackID uint32 + Group uint32 // the track's alternateGroup + Main bool // the main group's track; the others are stems +} + +// filterSegmentAudio returns a bare canonical .m4s containing every video +// track plus the main group's audio track matching the requested codec (Opus +// when wantOpus, else AAC) and, with stems, one Opus track per stem group +// (see stemOpusTracks). Track bytes are carried verbatim (signatures intact) +// in ascending track-id order; the kept audio tracks are described in that +// same order, which is the order their streams appear in a presentation of +// the result. If no main audio matches the requested codec, any one main +// track is kept (degraded but playable); with no audio info at all, the +// segment is returned unchanged. +func filterSegmentAudio(ctx context.Context, seg []byte, wantOpus, stems bool) ([]byte, []playbackAudio, error) { events, err := unwrapMuxlEvents(ctx, seg) if err != nil { - return nil, fmt.Errorf("unwrap segment for codec filter: %w", err) + return nil, nil, fmt.Errorf("unwrap segment for codec filter: %w", err) } cat, tracks := catalogAndTracks(events) if cat == nil { - return seg, nil + return seg, nil, nil } keep := map[uint32]bool{} @@ -420,25 +528,26 @@ func filterSegmentToCodec(ctx context.Context, seg []byte, wantOpus bool) ([]byt keep[v.TrackID()] = true } } - if cat.Audio != nil && len(cat.Audio.Renditions) > 0 { - var chosen uint32 - found := false - for _, a := range cat.Audio.Renditions { + var audio []playbackAudio + all := audioTracks(cat) + if main := mainAudioTracks(all); len(main) > 0 { + chosen := main[0] // no exact codec match — keep some audio rather than none + for _, a := range main { if (wantOpus && isOpusCodec(a.Codec)) || (!wantOpus && isAACCodec(a.Codec)) { - chosen, found = a.TrackID(), true + chosen = a break } } - if !found { // no exact codec match — keep some audio rather than none - for _, a := range cat.Audio.Renditions { - chosen, found = a.TrackID(), true - break - } - } - if found { - keep[chosen] = true + keep[chosen.TrackID()] = true + audio = append(audio, playbackAudio{TrackID: chosen.TrackID(), Group: chosen.AlternateGroup, Main: true}) + } + if stems { + for _, a := range stemOpusTracks(all) { + keep[a.TrackID()] = true + audio = append(audio, playbackAudio{TrackID: a.TrackID(), Group: a.AlternateGroup}) } } + sort.Slice(audio, func(i, j int) bool { return audio[i].TrackID < audio[j].TrackID }) ids := make([]uint32, 0, len(keep)) for id := range keep { @@ -451,7 +560,7 @@ func filterSegmentToCodec(ctx context.Context, seg []byte, wantOpus bool) ([]byt out = append(out, tracks[strconv.FormatUint(uint64(id), 10)]...) } if len(out) == 0 { - return seg, nil + return seg, nil, nil } - return out, nil + return out, audio, nil } diff --git a/pkg/media/transcode_stream.go b/pkg/media/transcode_stream.go index 69476bffb..2959eabd1 100644 --- a/pkg/media/transcode_stream.go +++ b/pkg/media/transcode_stream.go @@ -440,7 +440,16 @@ func (mm *MediaManager) finishTranscodedSegment(ctx context.Context, srcSeg, tra if cat == nil || cat.Audio == nil { return nil, fmt.Errorf("source segment has no audio track") } - var srcAudioTID, maxTID uint32 + // The transcoder re-encodes the demux's first audio stream, the main + // group's lowest track (completion runs only when the main group lacks a + // codec, see completionTarget); the completed track joins that group so + // the main-group consumers find it beside the source track. + main := mainAudioTracks(audioTracks(cat)) + if len(main) == 0 { + return nil, fmt.Errorf("source segment has no audio track") + } + srcAudio := main[0] + var maxTID uint32 if cat.Video != nil { for _, v := range cat.Video.Renditions { maxTID = maxU32(maxTID, v.TrackID()) @@ -448,13 +457,16 @@ func (mm *MediaManager) finishTranscodedSegment(ctx context.Context, srcSeg, tra } for _, a := range cat.Audio.Renditions { maxTID = maxU32(maxTID, a.TrackID()) - srcAudioTID = a.TrackID() } - sourceAudio := tracks[strconv.FormatUint(uint64(srcAudioTID), 10)] + sourceAudio := tracks[strconv.FormatUint(uint64(srcAudio.TrackID()), 10)] if len(sourceAudio) == 0 { - return nil, fmt.Errorf("source audio track %d missing", srcAudioTID) + return nil, fmt.Errorf("source audio track %d missing", srcAudio.TrackID()) } freeTID := maxTID + 1 + var meta []muxl.CanonicalizeOption + if srcAudio.AlternateGroup != 0 { + meta = append(meta, muxl.WithTrackMeta(map[uint32]muxl.TrackMeta{freeTID: {AlternateGroup: srcAudio.AlternateGroup, Label: srcAudio.Label}})) + } // Relabel transTID → freeTID: wrap the full segment to fMP4, canonicalize // with the remap (the video keyframe anchors one segment per GoP), extract @@ -465,7 +477,7 @@ func (mm *MediaManager) finishTranscodedSegment(ctx context.Context, srcSeg, tra if err := muxl.RunMuxlWrap(ctx, bytes.NewReader(transSeg), "fmp4", &fmp4); err != nil { return nil, fmt.Errorf("wrap transcoded segment: %w", err) } - canon, err := muxl.RunMuxlCanonicalize(ctx, fmp4.Bytes(), map[uint32]uint32{transTID: freeTID}) + canon, err := muxl.RunMuxlCanonicalize(ctx, fmp4.Bytes(), map[uint32]uint32{transTID: freeTID}, meta...) if err != nil { return nil, fmt.Errorf("relabel transcoded audio: %w", err) } @@ -495,13 +507,12 @@ func (mm *MediaManager) finishTranscodedSegment(ctx context.Context, srcSeg, tra return completed, nil } -// audioTrackID returns the (single) audio rendition's track id in a catalog. +// audioTrackID returns the lowest audio track id in a catalog: the transcoded +// catalog's only audio track. func audioTrackID(cat *muxl.MuxlCatalog) (uint32, bool) { - if cat == nil || cat.Audio == nil { + tracks := audioTracks(cat) + if len(tracks) == 0 { return 0, false } - for _, a := range cat.Audio.Renditions { - return a.TrackID(), true - } - return 0, false + return tracks[0].TrackID(), true } diff --git a/pkg/media/transcode_stream_test.go b/pkg/media/transcode_stream_test.go index 647ee3343..cd5bfcb36 100644 --- a/pkg/media/transcode_stream_test.go +++ b/pkg/media/transcode_stream_test.go @@ -127,7 +127,7 @@ func allSignedBareSegments(t *testing.T, ctx context.Context, ms *MediaSignerLoc eventCh := make(chan *muxl.MuxlEvent, 16) errCh := make(chan error, 1) go func() { - err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh) + err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh, false) close(eventCh) errCh <- err }() diff --git a/pkg/media/transcode_video_test.go b/pkg/media/transcode_video_test.go index cb0820fd0..0cde0e5e0 100644 --- a/pkg/media/transcode_video_test.go +++ b/pkg/media/transcode_video_test.go @@ -66,7 +66,7 @@ func sourceSegmentForTest(t *testing.T) []byte { eventCh := make(chan *muxl.MuxlEvent, 16) errCh := make(chan error, 1) go func() { - err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh) + err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh, false) close(eventCh) errCh <- err }() @@ -177,7 +177,7 @@ func TestMintVideoRenditionsRetimes(t *testing.T) { eventCh := make(chan *muxl.MuxlEvent, 16) errCh := make(chan error, 1) go func() { - err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh) + err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh, false) close(eventCh) errCh <- err }() diff --git a/pkg/media/validate.go b/pkg/media/validate.go index e9e9018b1..b86cc5744 100644 --- a/pkg/media/validate.go +++ b/pkg/media/validate.go @@ -83,7 +83,7 @@ func (mm *MediaManager) validateMP4(ctx context.Context, input io.Reader, local // dual-codec segment asynchronously (~1 GoP later, gapless). Everything else // — already dual-codec, no audio, audio-only, an exotic codec, or no node // signer — is distributed as-is, immediately. - if target, need := mm.audioCompletionTarget(ctx, buf); need { + if target, need := completionTarget(vs.catalog); need { if cert, keyPEM, serr := mm.transcodeSigner(); serr == nil { return mm.feedStreamTranscoder(ctx, vs, buf, target, cert, keyPEM) } else { @@ -100,6 +100,7 @@ func (mm *MediaManager) validateMP4(ctx context.Context, input io.Reader, local type validatedSegment struct { meta *SegmentMetadata mediaData *localdb.SegmentMediaData + catalog *muxl.MuxlCatalog label string repoDID string signingKeyDID string @@ -218,6 +219,7 @@ func (mm *MediaManager) validateSource(ctx context.Context, buf []byte, local bo return &validatedSegment{ meta: meta, mediaData: valid.MediaData, + catalog: valid.Catalog, label: label, repoDID: repoDID, signingKeyDID: signingKeyDID, @@ -376,6 +378,7 @@ type ValidationResult struct { Pub *atcrypto.PublicKeyK256 Meta *SegmentMetadata MediaData *localdb.SegmentMediaData + Catalog *muxl.MuxlCatalog // the segment's track configs Manifest *c2patypes.Manifest Cert string } @@ -383,6 +386,7 @@ type ValidationResult struct { func ValidateMP4Media(ctx context.Context, buf []byte) (*ValidationResult, error) { g, ctx := errgroup.WithContext(ctx) var mediaData *localdb.SegmentMediaData + var catalog *muxl.MuxlCatalog var validationResult *ValidationResult g.Go(func() error { // gstreamer needs a parseable MP4, but the bytes on the wire/disk are @@ -397,6 +401,16 @@ func ValidateMP4Media(ctx context.Context, buf []byte) (*ValidationResult, error mediaData, err = ValidateMP4MediaData(ctx, flat.Bytes()) return err }) + g.Go(func() error { + // The catalog describes every audio track (codec, group, label); + // the gst parse above only reads the first one. + events, err := unwrapMuxlEvents(ctx, buf) + if err != nil { + return fmt.Errorf("unwrap segment for catalog: %w", err) + } + catalog, _ = catalogAndTracks(events) + return nil + }) g.Go(func() error { var err error validationResult, err = ValidateMP4MediaC2PA(ctx, buf) @@ -406,7 +420,11 @@ func ValidateMP4Media(ctx context.Context, buf []byte) (*ValidationResult, error if err != nil { return nil, err } + if audio := audioMediaData(catalog); len(audio) > 0 { + mediaData.Audio = audio + } validationResult.MediaData = mediaData + validationResult.Catalog = catalog return validationResult, nil } diff --git a/pkg/media/validate_bare_test.go b/pkg/media/validate_bare_test.go index 040421a46..552db7bd9 100644 --- a/pkg/media/validate_bare_test.go +++ b/pkg/media/validate_bare_test.go @@ -44,7 +44,7 @@ func TestValidateMP4MediaBareSegment(t *testing.T) { eventCh := make(chan *muxl.MuxlEvent, 16) errCh := make(chan error, 1) go func() { - err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh) + err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh, false) close(eventCh) errCh <- err }() @@ -91,7 +91,7 @@ func TestFeedLiveWindow(t *testing.T) { eventCh := make(chan *muxl.MuxlEvent, 16) errCh := make(chan error, 1) go func() { - err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh) + err := ms.SignSegmentStream(ctx, bytes.NewReader(frag), eventCh, false) close(eventCh) errCh <- err }() diff --git a/pkg/media/webrtc_playback2.go b/pkg/media/webrtc_playback2.go index e4e728d8a..d6dd23140 100644 --- a/pkg/media/webrtc_playback2.go +++ b/pkg/media/webrtc_playback2.go @@ -31,6 +31,10 @@ func (mm *MediaManager) WebRTCPlayback2(ctx context.Context, user string, rendit } audioOnly := rendition == renditions.AudioRendition.Name + busRendition := rendition + if audioOnly { + busRendition = "source" + } var videoTrack *webrtc.TrackLocalStaticSample var videoRTPSender *webrtc.RTPSender @@ -45,14 +49,34 @@ func (mm *MediaManager) WebRTCPlayback2(ctx context.Context, user string, rendit } } + // Audio tracks: the main audio (the full mix) first, then one track per + // stem group the stream currently carries, "audio-g", ascending. + // pion binds our audio senders to the offer's audio transceivers in the + // order we add them, so this order is the client's: it offers one + // transceiver per track it can take, and we send what fits. audioTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, "audio", "pion") if err != nil { return nil, fmt.Errorf("failed to create audio track: %w", err) } + audioSenders := []*webrtc.RTPSender{} audioRTPSender, err := peerConnection.AddTrack(audioTrack) if err != nil { return nil, fmt.Errorf("failed to add audio track to peer connection: %w", err) } + audioSenders = append(audioSenders, audioRTPSender) + stemTracks := map[uint32]*webrtc.TrackLocalStaticSample{} + for _, group := range stemGroupsFor(mm.bus.LatestSegment(user, busRendition), offeredAudioTracks(offer)-1) { + track, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, fmt.Sprintf("audio-g%d", group), "pion") + if err != nil { + return nil, fmt.Errorf("failed to create stem track: %w", err) + } + sender, err := peerConnection.AddTrack(track) + if err != nil { + return nil, fmt.Errorf("failed to add stem track to peer connection: %w", err) + } + audioSenders = append(audioSenders, sender) + stemTracks[group] = track + } close := func() { if cErr := peerConnection.Close(); cErr != nil { @@ -120,10 +144,6 @@ func (mm *MediaManager) WebRTCPlayback2(ctx context.Context, user string, rendit packetQueue := make(chan *bus.PacketizedSegment, 1024) go func() { - busRendition := rendition - if audioOnly { - busRendition = "source" - } segChan := mm.bus.SubscribeSegmentBuf(ctx, user, busRendition, 2) defer mm.bus.UnsubscribeSegment(ctx, user, busRendition, segChan) for { @@ -179,32 +199,43 @@ func (mm *MediaManager) WebRTCPlayback2(ctx context.Context, user string, rendit // timing regressions are immediately audible as garbling. // Video is the track that needs per-sample durations (its // spacing goes non-uniform when an encoder sheds frames). - var audioDur time.Duration - if len(packet.Audio) > 0 { - audioDur = packet.Duration / time.Duration(len(packet.Audio)) + // The stems are paced the same way against the same + // segment duration, so they stay aligned with the mix. + writeOpus := func(track *webrtc.TrackLocalStaticSample, samples []bus.PacketizedSample) error { + audioDur := packet.Duration / time.Duration(len(samples)) + ticker := time.NewTicker(time.Duration(float64(audioDur) * (1 / scalar))) + defer ticker.Stop() + for _, audio := range samples { + err := track.WriteSample(media.Sample{Data: audio.Data, Duration: audioDur}) + if err != nil { + return fmt.Errorf("failed to write audio sample: %w", err) + } + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + continue + } + } + return nil } - if audioDur > 0 { + if len(packet.Audio) > 0 && packet.Duration > 0 { wroteAny = true - g.Go(func() error { - ticker := time.NewTicker(time.Duration(float64(audioDur) * (1 / scalar))) - defer ticker.Stop() - for _, audio := range packet.Audio { - err := audioTrack.WriteSample(media.Sample{Data: audio.Data, Duration: audioDur}) - if err != nil { - return fmt.Errorf("failed to write audio sample: %w", err) - } - select { - case <-ctx.Done(): - return nil - case <-ticker.C: - continue - } - } - return nil - }) + g.Go(func() error { return writeOpus(audioTrack, packet.Audio) }) } else { log.Warn(ctx, "no audio samples to write") } + for _, stem := range packet.Stems { + // A stem this session has no track for (it appeared + // after the offer, or the client offered fewer + // transceivers) waits for the viewer's next session. + track := stemTracks[stem.Group] + if track == nil || len(stem.Audio) == 0 || packet.Duration == 0 { + continue + } + wroteAny = true + g.Go(func() error { return writeOpus(track, stem.Audio) }) + } if wroteAny { if err := g.Wait(); err != nil { log.Error(ctx, "failed to write samples", "error", err) @@ -226,14 +257,16 @@ func (mm *MediaManager) WebRTCPlayback2(ctx context.Context, user string, rendit }() } - go func() { - rtcpBuf := make([]byte, 1500) - for { - if _, _, rtcpErr := audioRTPSender.Read(rtcpBuf); rtcpErr != nil { - return + for _, sender := range audioSenders { + go func() { + rtcpBuf := make([]byte, 1500) + for { + if _, _, rtcpErr := sender.Read(rtcpBuf); rtcpErr != nil { + return + } } - } - }() + }() + } // Set the handler for ICE connection state // This will notify you when the peer has connected/disconnected @@ -272,6 +305,40 @@ func (mm *MediaManager) WebRTCPlayback2(ctx context.Context, user string, rendit } } +// offeredAudioTracks counts the audio transceivers in a viewer's offer: how +// many audio tracks the answer can carry. 1 when the offer can't be read +// (SetRemoteDescription reports why). +func offeredAudioTracks(offer *webrtc.SessionDescription) int { + parsed, err := offer.Unmarshal() + if err != nil { + return 1 + } + n := 0 + for _, md := range parsed.MediaDescriptions { + if md.MediaName.Media == "audio" { + n++ + } + } + return max(n, 1) +} + +// stemGroupsFor lists the stem groups a playback session sends, ascending: +// the stems of the stream's latest packetized segment (none before the first +// segment, or for a stream without stems), at most limit of them. +func stemGroupsFor(latest *bus.Seg, limit int) []uint32 { + if latest == nil || latest.PacketizedData == nil || limit <= 0 { + return nil + } + var groups []uint32 + for _, stem := range latest.PacketizedData.Stems { + if len(groups) == limit { + break + } + groups = append(groups, stem.Group) + } + return groups +} + // writeSamples writes one track's samples paced by their real durations — // the source timeline, so non-uniform frame spacing (bursts and gaps from an // encoder shedding frames) reaches the viewer intact. The stamped duration diff --git a/pkg/muxl/muxl.go b/pkg/muxl/muxl.go index 11b2166af..aaa36f74f 100644 --- a/pkg/muxl/muxl.go +++ b/pkg/muxl/muxl.go @@ -18,15 +18,29 @@ import ( // aliases, so values (events, catalogs, signer inputs) flow between this package // and upstream with no conversion and the existing call sites are unchanged. type ( - MuxlEvent = upstream.Event - MuxlCatalog = upstream.Catalog - MuxlCatalogVideo = upstream.CatalogVideo - MuxlCatalogAudio = upstream.CatalogAudio - MuxlVideoConfig = upstream.VideoConfig - MuxlAudioConfig = upstream.AudioConfig - MuxlContainer = upstream.Container - SignerInput = upstream.SignerInput - TranscodeInput = upstream.TranscodeInput + MuxlEvent = upstream.Event + MuxlCatalog = upstream.Catalog + MuxlCatalogVideo = upstream.CatalogVideo + MuxlCatalogAudio = upstream.CatalogAudio + MuxlVideoConfig = upstream.VideoConfig + MuxlAudioConfig = upstream.AudioConfig + MuxlContainer = upstream.Container + SignerInput = upstream.SignerInput + TranscodeInput = upstream.TranscodeInput + TrackMeta = upstream.TrackMeta + SegmentEventsOption = upstream.SegmentEventsOption + CanonicalizeOption = upstream.CanonicalizeOption +) + +// Audio track metadata options: an alternateGroup and label per OUTPUT track +// id, set explicitly (WithSegmentTrackMeta while segmenting, WithTrackMeta +// while canonicalizing) or read from the source's tkhd.alternate_group and +// trak/udta/name (WithSegmentSourceTrackMeta, or SignerInput.SourceTrackMeta +// while signing). See the upstream docs. +var ( + WithSegmentTrackMeta = upstream.WithSegmentTrackMeta + WithSegmentSourceTrackMeta = upstream.WithSegmentSourceTrackMeta + WithTrackMeta = upstream.WithTrackMeta ) // TranscodeIngredientLabel is the C2PA ingredient label SignTranscode assigns @@ -196,12 +210,12 @@ func RunMuxlReadSegmentsAtFileOffset(ctx context.Context, src io.ReaderAt, size, // RunMuxlSegmenterEvents segments an fMP4 stream into per-GoP canonical MUXL // events (unsigned). -func RunMuxlSegmenterEvents(ctx context.Context, input io.Reader, eventCh chan *MuxlEvent) error { +func RunMuxlSegmenterEvents(ctx context.Context, input io.Reader, eventCh chan *MuxlEvent, opts ...SegmentEventsOption) error { eng, err := getEngine() if err != nil { return err } - return eng.SegmentEvents(ctx, input, eventCh) + return eng.SegmentEvents(ctx, input, eventCh, opts...) } // RunMuxlSignSegment segments an fMP4 stream and S2PA-signs each canonical @@ -218,12 +232,12 @@ func RunMuxlSignSegment(ctx context.Context, input io.Reader, in SignerInput, in // output) into a canonical MUXL fMP4. trackRemap (may be nil) reassigns track // IDs in the output — used to mint a transcoded rendition at a free id so it // can join the source's tracks in one multi-track segment without colliding. -func RunMuxlCanonicalize(ctx context.Context, mp4 []byte, trackRemap map[uint32]uint32) ([]byte, error) { +// opts (e.g. WithTrackMeta) apply to the remapped ids. +func RunMuxlCanonicalize(ctx context.Context, mp4 []byte, trackRemap map[uint32]uint32, opts ...CanonicalizeOption) ([]byte, error) { eng, err := getEngine() if err != nil { return nil, err } - var opts []upstream.CanonicalizeOption if len(trackRemap) > 0 { opts = append(opts, upstream.WithTrackRemap(trackRemap)) } diff --git a/pkg/replication/websocketrep/moq_test.go b/pkg/replication/websocketrep/moq_test.go index 88274f131..25243ca26 100644 --- a/pkg/replication/websocketrep/moq_test.go +++ b/pkg/replication/websocketrep/moq_test.go @@ -37,7 +37,7 @@ func signedSegments(t *testing.T, ctx context.Context) [][]byte { events := make(chan *muxl.MuxlEvent, 16) signed := make(chan error, 1) go func() { - signed <- ms.SignSegmentStream(ctx, bytes.NewReader(frag), events) + signed <- ms.SignSegmentStream(ctx, bytes.NewReader(frag), events, false) close(events) }() var segments [][]byte diff --git a/pkg/vod/clip_test.go b/pkg/vod/clip_test.go index 1c8e57c6e..960b42b61 100644 --- a/pkg/vod/clip_test.go +++ b/pkg/vod/clip_test.go @@ -32,7 +32,7 @@ func liveSegments(t *testing.T) [][]byte { events := make(chan *muxl.MuxlEvent, 16) errCh := make(chan error, 1) go func() { - errCh <- ms.SignSegmentStream(context.Background(), bytes.NewReader(frag), events) + errCh <- ms.SignSegmentStream(context.Background(), bytes.NewReader(frag), events, false) close(events) }() var segs [][]byte -- 2.54.0