diff --git a/api/openapi.yaml b/api/openapi.yaml index 20540cf5..90cb47d9 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -95,6 +95,12 @@ components: - read - publish + MoQVersion: + type: string + enum: + - moqt-18 + - moqt-19 + OKStatus: type: string enum: @@ -805,6 +811,8 @@ components: $ref: "#/components/schemas/MoQSessionState" userAgent: type: string + version: + $ref: "#/components/schemas/MoQVersion" MoQSessionList: type: object diff --git a/docs/3-publish/01-moq-clients.md b/docs/3-publish/01-moq-clients.md index 91e15a18..e30b3352 100644 --- a/docs/3-publish/01-moq-clients.md +++ b/docs/3-publish/01-moq-clients.md @@ -9,7 +9,7 @@ Media-over-QUIC is a streaming protocol built upon cutting edge protocols (QUIC, Media-over-QUIC has a wide range of features and variants, most of them in active development. We currently support the following: -- We support `draft-18` of the [main specification](https://datatracker.ietf.org/doc/html/draft-ietf-moq-transport-18). +- The server supports `draft-19` and `draft-18` of the [main specification](https://datatracker.ietf.org/doc/html/draft-ietf-moq-transport-19), and prefers `draft-19` when both are offered during negotiation. - We only support using Media-over-QUIC through browsers and in particular through the WebTransport API. We do not support using QUIC directly. - We support the `PUBLISH` and `SUBSCRIBE` messages only, which are the ones meant to be used with a routing solution like _MediaMTX_. - We use the MOQT Streaming Format (MSF) to advertise tracks, described in [this specification](https://datatracker.ietf.org/doc/html/draft-ietf-moq-msf-00). diff --git a/internal/api/api_moq_test.go b/internal/api/api_moq_test.go index 3cbaa4a6..df644f04 100644 --- a/internal/api/api_moq_test.go +++ b/internal/api/api_moq_test.go @@ -56,6 +56,7 @@ func TestMoQSessionsList(t *testing.T) { State: defs.APIMoQSessionStatePublish, Path: "stream1", Query: "token=abc", + Version: defs.APIMoQVersionDraft19, InboundBytes: 1000, OutboundBytes: 2000, }, @@ -66,6 +67,7 @@ func TestMoQSessionsList(t *testing.T) { State: defs.APIMoQSessionStateRead, Path: "stream2", Query: "", + Version: defs.APIMoQVersionDraft18, InboundBytes: 500, OutboundBytes: 1500, }, @@ -109,6 +111,7 @@ func TestMoQSessionsGet(t *testing.T) { State: defs.APIMoQSessionStatePublish, Path: "mystream", Query: "key=value", + Version: defs.APIMoQVersionDraft19, InboundBytes: 999999, OutboundBytes: 888888, }, @@ -155,6 +158,7 @@ func TestMoQSessionsKick(t *testing.T) { State: defs.APIMoQSessionStatePublish, Path: "mystream", Query: "", + Version: defs.APIMoQVersionDraft19, InboundBytes: 1000, OutboundBytes: 2000, }, diff --git a/internal/apidocsgen/enums.go b/internal/apidocsgen/enums.go index 2699680c..8a8bef2a 100644 --- a/internal/apidocsgen/enums.go +++ b/internal/apidocsgen/enums.go @@ -47,6 +47,11 @@ var enums = []struct { internalName: "APIMoQSessionState", File: filepath.Join("internal", "defs", "api_moq.go"), }, + { + externalName: "MoQVersion", + internalName: "APIMoQVersion", + File: filepath.Join("internal", "defs", "api_moq.go"), + }, { externalName: "OKStatus", internalName: "APIOKStatus", diff --git a/internal/apidocsgen/structs.go b/internal/apidocsgen/structs.go index d18e519e..bc42be8a 100644 --- a/internal/apidocsgen/structs.go +++ b/internal/apidocsgen/structs.go @@ -366,6 +366,9 @@ func isStructEnum(rt reflect.Type) bool { case reflect.TypeOf(defs.APIMoQSessionState("")): return true + case reflect.TypeOf(defs.APIMoQVersion("")): + return true + case reflect.TypeOf(defs.APISRTConnState("")): return true } diff --git a/internal/defs/api_moq.go b/internal/defs/api_moq.go index 772269ed..4a9f10f4 100644 --- a/internal/defs/api_moq.go +++ b/internal/defs/api_moq.go @@ -23,6 +23,15 @@ const ( APIMoQSessionStatePublish APIMoQSessionState = "publish" ) +// APIMoQVersion is a MoQ protocol version. +type APIMoQVersion string + +// protocol versions. +const ( + APIMoQVersionDraft18 APIMoQVersion = "moqt-18" + APIMoQVersionDraft19 APIMoQVersion = "moqt-19" +) + // APIMoQSessionList is a list of MoQ sessions. type APIMoQSessionList struct { ItemCount int `json:"itemCount"` @@ -39,6 +48,7 @@ type APIMoQSession struct { Path string `json:"path"` Query string `json:"query"` UserAgent string `json:"userAgent"` + Version APIMoQVersion `json:"version"` InboundBytes uint64 `json:"inboundBytes"` OutboundBytes uint64 `json:"outboundBytes"` } diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index 7ae42469..f766c380 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -339,6 +339,7 @@ func (dummyMoQServer) APISessionsList() (*defs.APIMoQSessionList, error) { RemoteAddr: "127.0.0.2:3456", State: defs.APIMoQSessionStatePublish, Path: "mypath", + Version: defs.APIMoQVersionDraft19, InboundBytes: 321, OutboundBytes: 654, }}, diff --git a/internal/protocols/moq/from_stream.go b/internal/protocols/moq/from_stream.go index f3eec12c..35fe9eae 100644 --- a/internal/protocols/moq/from_stream.go +++ b/internal/protocols/moq/from_stream.go @@ -4,6 +4,7 @@ package moq import ( "encoding/base64" "encoding/hex" + "strconv" "strings" "github.com/bluenviron/gortsplib/v5/pkg/description" @@ -16,262 +17,322 @@ import ( "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg4audio" "github.com/bluenviron/mediacommon/v2/pkg/codecs/vp8" "github.com/bluenviron/mediacommon/v2/pkg/codecs/vp9" + "github.com/bluenviron/mediamtx/internal/protocols/moq/catalog" + "github.com/bluenviron/mediamtx/internal/stream" "github.com/bluenviron/mediamtx/internal/unit" ) -// FromStream maps a MediaMTX stream to a Media-over-QUIC stream. -func FromStream(desc *description.Session) ([]*Track, error) { - var tracks []*Track +type writeDataFunc func(payload []byte, pts int64) error + +// SetupTrackFunc is a function that sets up a track in a MediaMTX stream. +type SetupTrackFunc func(r *stream.Reader, writeData writeDataFunc) + +// FromStream maps a MediaMTX stream to a Media-over-QUIC catalog and subscribed tracks. +func FromStream(desc *description.Session) (*catalog.Catalog, []SetupTrackFunc, error) { + cat := &catalog.Catalog{ + Version: 1, + } + + var setupTracks []SetupTrackFunc + + addTrack := func( + media *description.Media, + forma format.Format, + track catalog.Track, + genParsePayload func(writeData writeDataFunc) func(u *unit.Unit) error, + ) { + track.Name = strconv.Itoa(len(cat.Tracks)) + track.Packaging = "loc" + track.IsLive = true + + setup := func(r *stream.Reader, writeData writeDataFunc) { + parsePayload := genParsePayload(writeData) + + r.OnData(media, forma, func(u *unit.Unit) error { + return parsePayload(u) + }) + } + + cat.Tracks = append(cat.Tracks, track) + setupTracks = append(setupTracks, setup) + } for _, media := range desc.Medias { for _, forma := range media.Formats { switch forma := forma.(type) { case *format.AV1: - firstRandomAccess := false - - track := &Track{ - Codec: "av01.0.04M.08", - Media: media, - Format: forma, - OnData: func(u *unit.Unit, wrapped func([]byte, int64) error) error { - if u.NilPayload() { - return nil - } - - if !firstRandomAccess && !av1.IsRandomAccess2(u.Payload.(unit.PayloadAV1)) { - return nil - } - firstRandomAccess = true - - payload, err := av1.Bitstream([][]byte(u.Payload.(unit.PayloadAV1))).Marshal() - if err != nil { - return err - } - - return wrapped(payload, u.PTS) + addTrack( + media, + forma, + catalog.Track{ + Codec: "av01.0.04M.08", }, - } - tracks = append(tracks, track) + func(writeData writeDataFunc) func(u *unit.Unit) error { + firstRandomAccess := false - case *format.VP9: - firstRandomAccess := false + return func(u *unit.Unit) error { + if u.NilPayload() { + return nil + } - track := &Track{ - Codec: "vp09.00.10.08", - Media: media, - Format: forma, - OnData: func(u *unit.Unit, wrapped func([]byte, int64) error) error { - if u.NilPayload() { - return nil - } + if !firstRandomAccess && !av1.IsRandomAccess2(u.Payload.(unit.PayloadAV1)) { + return nil + } + firstRandomAccess = true - if !firstRandomAccess && !vp9.IsRandomAccess(u.Payload.(unit.PayloadVP9)) { - return nil - } - firstRandomAccess = true - - return wrapped(u.Payload.(unit.PayloadVP9), u.PTS) - }, - } - tracks = append(tracks, track) - - case *format.VP8: - firstRandomAccess := false - - track := &Track{ - Codec: "vp8", - Media: media, - Format: forma, - OnData: func(u *unit.Unit, wrapped func([]byte, int64) error) error { - if u.NilPayload() { - return nil - } - - if !firstRandomAccess && !vp8.IsRandomAccess(u.Payload.(unit.PayloadVP8)) { - return nil - } - firstRandomAccess = true - - return wrapped(u.Payload.(unit.PayloadVP8), u.PTS) - }, - } - tracks = append(tracks, track) - - case *format.H265: - firstRandomAccess := false - - track := &Track{ - Codec: "hev1.1.6.L93.B0", - Media: media, - Format: forma, - OnData: func(u *unit.Unit, wrapped func([]byte, int64) error) error { - if u.NilPayload() { - return nil - } - - if !firstRandomAccess && !h265.IsRandomAccess(u.Payload.(unit.PayloadH265)) { - return nil - } - firstRandomAccess = true - - payload, err := h264.AVCC(u.Payload.(unit.PayloadH265)).Marshal() - if err != nil { - return err - } - - return wrapped(payload, u.PTS) - }, - } - tracks = append(tracks, track) - - case *format.H264: - firstRandomAccess := false - - track := &Track{ - Codec: "avc3.640028", - Media: media, - Format: forma, - OnData: func(u *unit.Unit, wrapped func([]byte, int64) error) error { - if u.NilPayload() { - return nil - } - - if !firstRandomAccess && !h264.IsRandomAccess(u.Payload.(unit.PayloadH264)) { - return nil - } - firstRandomAccess = true - - payload, err := h264.AVCC(u.Payload.(unit.PayloadH264)).Marshal() - if err != nil { - return err - } - - return wrapped(payload, u.PTS) - }, - } - tracks = append(tracks, track) - - case *format.Opus: - track := &Track{ - Codec: "opus", - Samplerate: 48000, - Channels: forma.ChannelCount, - Media: media, - Format: forma, - OnData: func(u *unit.Unit, wrapped func([]byte, int64) error) error { - if u.NilPayload() { - return nil - } - - for _, pkt := range u.Payload.(unit.PayloadOpus) { - err := wrapped(pkt, u.PTS) + payload, err := av1.Bitstream([][]byte(u.Payload.(unit.PayloadAV1))).Marshal() if err != nil { return err } + + return writeData(payload, u.PTS) } - return nil }, - } - tracks = append(tracks, track) + ) + + case *format.VP9: + addTrack( + media, + forma, + catalog.Track{ + Codec: "vp09.00.10.08", + }, + func(writeData writeDataFunc) func(u *unit.Unit) error { + firstRandomAccess := false + + return func(u *unit.Unit) error { + if u.NilPayload() { + return nil + } + + if !firstRandomAccess && !vp9.IsRandomAccess(u.Payload.(unit.PayloadVP9)) { + return nil + } + firstRandomAccess = true + + return writeData(u.Payload.(unit.PayloadVP9), u.PTS) + } + }, + ) + + case *format.VP8: + addTrack( + media, + forma, + catalog.Track{ + Codec: "vp8", + }, + func(writeData writeDataFunc) func(u *unit.Unit) error { + firstRandomAccess := false + + return func(u *unit.Unit) error { + if u.NilPayload() { + return nil + } + + if !firstRandomAccess && !vp8.IsRandomAccess(u.Payload.(unit.PayloadVP8)) { + return nil + } + firstRandomAccess = true + + return writeData(u.Payload.(unit.PayloadVP8), u.PTS) + } + }, + ) + + case *format.H265: + addTrack( + media, + forma, + catalog.Track{ + Codec: "hev1.1.6.L93.B0", + }, + func(writeData writeDataFunc) func(u *unit.Unit) error { + firstRandomAccess := false + + return func(u *unit.Unit) error { + if u.NilPayload() { + return nil + } + + if !firstRandomAccess && !h265.IsRandomAccess(u.Payload.(unit.PayloadH265)) { + return nil + } + firstRandomAccess = true + + payload, err := h264.AVCC(u.Payload.(unit.PayloadH265)).Marshal() + if err != nil { + return err + } + + return writeData(payload, u.PTS) + } + }, + ) + + case *format.H264: + addTrack( + media, + forma, + catalog.Track{ + Codec: "avc3.640028", + }, + func(writeData writeDataFunc) func(u *unit.Unit) error { + firstRandomAccess := false + + return func(u *unit.Unit) error { + if u.NilPayload() { + return nil + } + + if !firstRandomAccess && !h264.IsRandomAccess(u.Payload.(unit.PayloadH264)) { + return nil + } + firstRandomAccess = true + + payload, err := h264.AVCC(u.Payload.(unit.PayloadH264)).Marshal() + if err != nil { + return err + } + + return writeData(payload, u.PTS) + } + }, + ) + + case *format.Opus: + addTrack( + media, + forma, + catalog.Track{ + Codec: "opus", + Samplerate: 48000, + Channels: forma.ChannelCount, + }, + func(writeData writeDataFunc) func(u *unit.Unit) error { + return func(u *unit.Unit) error { + if u.NilPayload() { + return nil + } + + for _, pkt := range u.Payload.(unit.PayloadOpus) { + err := writeData(pkt, u.PTS) + if err != nil { + return err + } + } + return nil + } + }, + ) case *format.Generic: if strings.HasPrefix(strings.ToLower(forma.RTPMap()), "flac/") { enc, err := hex.DecodeString(forma.FMT["streaminfo"]) if err != nil { - return nil, err + return nil, nil, err } var streamInfo flac.StreamInfo err = streamInfo.Unmarshal(enc) if err != nil { - return nil, err + return nil, nil, err } - track := &Track{ - Codec: "flac", - Samplerate: int(streamInfo.SampleRate), - Channels: int(streamInfo.ChannelCount), - InitData: base64.StdEncoding.EncodeToString(enc), - Media: media, - Format: forma, - OnData: func(u *unit.Unit, wrapped func([]byte, int64) error) error { - if u.NilPayload() { - return nil - } - - return wrapped(u.Payload.(unit.PayloadFLAC), u.PTS) + addTrack( + media, + forma, + catalog.Track{ + Codec: "flac", + Samplerate: int(streamInfo.SampleRate), + Channels: int(streamInfo.ChannelCount), + InitData: base64.StdEncoding.EncodeToString(enc), }, - } - tracks = append(tracks, track) + func(writeData writeDataFunc) func(u *unit.Unit) error { + return func(u *unit.Unit) error { + if u.NilPayload() { + return nil + } + + return writeData(u.Payload.(unit.PayloadFLAC), u.PTS) + } + }, + ) } case *format.MPEG4Audio: if forma.Config != nil { enc, err := forma.Config.Marshal() if err != nil { - return nil, err + return nil, nil, err } - track := &Track{ - Codec: "mp4a.40.2", - Samplerate: forma.Config.SampleRate, - Channels: int(forma.Config.ChannelConfig), - InitData: base64.StdEncoding.EncodeToString(enc), - Media: media, - Format: forma, - OnData: func(u *unit.Unit, wrapped func([]byte, int64) error) error { + addTrack( + media, + forma, + catalog.Track{ + Codec: "mp4a.40.2", + Samplerate: forma.Config.SampleRate, + Channels: int(forma.Config.ChannelConfig), + InitData: base64.StdEncoding.EncodeToString(enc), + }, + func(writeData writeDataFunc) func(u *unit.Unit) error { + return func(u *unit.Unit) error { + if u.NilPayload() { + return nil + } + + pts := u.PTS + + for _, au := range u.Payload.(unit.PayloadMPEG4Audio) { + err2 := writeData(au, pts) + if err2 != nil { + return err2 + } + + pts += mpeg4audio.SamplesPerAccessUnit + } + return nil + } + }, + ) + } + + case *format.G711: + addTrack( + media, + forma, + catalog.Track{ + Codec: "pcm-s16", + Samplerate: forma.SampleRate, + Channels: forma.ChannelCount, + }, + func(writeData writeDataFunc) func(u *unit.Unit) error { + return func(u *unit.Unit) error { if u.NilPayload() { return nil } - pts := u.PTS - - for _, au := range u.Payload.(unit.PayloadMPEG4Audio) { - err2 := wrapped(au, pts) - if err2 != nil { - return err2 - } - - pts += mpeg4audio.SamplesPerAccessUnit + var bigEndian []byte + if forma.MULaw { + var mu g711.Mulaw + mu.Unmarshal(u.Payload.(unit.PayloadG711)) + bigEndian = mu + } else { + var al g711.Alaw + al.Unmarshal(u.Payload.(unit.PayloadG711)) + bigEndian = al } - return nil - }, - } - tracks = append(tracks, track) - } - case *format.G711: - track := &Track{ - Codec: "pcm-s16", - Samplerate: forma.SampleRate, - Channels: forma.ChannelCount, - Media: media, - Format: forma, - OnData: func(u *unit.Unit, wrapped func([]byte, int64) error) error { - if u.NilPayload() { - return nil + swapped := make([]byte, len(bigEndian)) + for i := 0; i+2 <= len(bigEndian); i += 2 { + swapped[i], swapped[i+1] = bigEndian[i+1], bigEndian[i] + } + + return writeData(swapped, u.PTS) } - - var bigEndian []byte - if forma.MULaw { - var mu g711.Mulaw - mu.Unmarshal(u.Payload.(unit.PayloadG711)) - bigEndian = mu - } else { - var al g711.Alaw - al.Unmarshal(u.Payload.(unit.PayloadG711)) - bigEndian = al - } - - swapped := make([]byte, len(bigEndian)) - for i := 0; i+2 <= len(bigEndian); i += 2 { - swapped[i], swapped[i+1] = bigEndian[i+1], bigEndian[i] - } - - return wrapped(swapped, u.PTS) }, - } - tracks = append(tracks, track) + ) case *format.LPCM: var codec string @@ -286,33 +347,36 @@ func FromStream(desc *description.Session) ([]*Track, error) { codec = "pcm-s32" } - track := &Track{ - Codec: codec, - Samplerate: forma.SampleRate, - Channels: forma.ChannelCount, - Media: media, - Format: forma, - OnData: func(u *unit.Unit, onData func([]byte, int64) error) error { - if u.NilPayload() { - return nil - } - - src := []byte(u.Payload.(unit.PayloadLPCM)) - byteDepth := forma.BitDepth / 8 - swapped := make([]byte, len(src)) - for i := 0; i+byteDepth <= len(src); i += byteDepth { - for j := range byteDepth { - swapped[i+j] = src[i+byteDepth-1-j] - } - } - - return onData(swapped, u.PTS) + addTrack( + media, + forma, + catalog.Track{ + Codec: codec, + Samplerate: forma.SampleRate, + Channels: forma.ChannelCount, }, - } - tracks = append(tracks, track) + func(writeData writeDataFunc) func(u *unit.Unit) error { + return func(u *unit.Unit) error { + if u.NilPayload() { + return nil + } + + src := []byte(u.Payload.(unit.PayloadLPCM)) + byteDepth := forma.BitDepth / 8 + swapped := make([]byte, len(src)) + for i := 0; i+byteDepth <= len(src); i += byteDepth { + for j := range byteDepth { + swapped[i+j] = src[i+byteDepth-1-j] + } + } + + return writeData(swapped, u.PTS) + } + }, + ) } } } - return tracks, nil + return cat, setupTracks, nil } diff --git a/internal/protocols/moq/from_stream_test.go b/internal/protocols/moq/from_stream_test.go index c2fb4dac..dbd17ce2 100644 --- a/internal/protocols/moq/from_stream_test.go +++ b/internal/protocols/moq/from_stream_test.go @@ -9,6 +9,7 @@ import ( "github.com/bluenviron/gortsplib/v5/pkg/format" "github.com/bluenviron/mediacommon/v2/pkg/codecs/flac" "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg4audio" + "github.com/bluenviron/mediamtx/internal/protocols/moq/catalog" "github.com/stretchr/testify/require" ) @@ -59,53 +60,84 @@ func TestFromStream(t *testing.T) { }, } - tracks, err := FromStream(desc) + cat, _, err := FromStream(desc) require.NoError(t, err) - for _, track := range tracks { - track.OnData = nil - } - - require.Equal(t, []*Track{ - { - Codec: "av01.0.04M.08", Media: desc.Medias[0], - Format: desc.Medias[0].Formats[0], + require.Equal(t, &catalog.Catalog{ + Version: 1, + Tracks: []catalog.Track{ + { + Name: "0", + Packaging: "loc", + IsLive: true, + Codec: "av01.0.04M.08", + }, + { + Name: "1", + Packaging: "loc", + IsLive: true, + Codec: "vp09.00.10.08", + }, + { + Name: "2", + Packaging: "loc", + IsLive: true, + Codec: "vp8", + }, + { + Name: "3", + Packaging: "loc", + IsLive: true, + Codec: "hev1.1.6.L93.B0", + }, + { + Name: "4", + Packaging: "loc", + IsLive: true, + Codec: "avc3.640028", + }, + { + Name: "5", + Packaging: "loc", + IsLive: true, + Codec: "opus", + Samplerate: 48000, + Channels: 2, + }, + { + Name: "6", + Packaging: "loc", + IsLive: true, + Codec: "flac", + Samplerate: 44100, + Channels: 2, + InitData: base64.StdEncoding.EncodeToString(flacEnc), + }, + { + Name: "7", + Packaging: "loc", + IsLive: true, + Codec: "mp4a.40.2", + Samplerate: 44100, + Channels: 2, + InitData: base64.StdEncoding.EncodeToString(mpeg4Enc), + }, + { + Name: "8", + Packaging: "loc", + IsLive: true, + Codec: "pcm-s16", + Samplerate: 8000, + Channels: 1, + }, + { + Name: "9", + Packaging: "loc", + IsLive: true, + Codec: "pcm-s16", + Samplerate: 44100, + Channels: 2, + }, }, - { - Codec: "vp09.00.10.08", Media: desc.Medias[1], - Format: desc.Medias[1].Formats[0], - }, - { - Codec: "vp8", Media: desc.Medias[2], - Format: desc.Medias[2].Formats[0], - }, - { - Codec: "hev1.1.6.L93.B0", Media: desc.Medias[3], - Format: desc.Medias[3].Formats[0], - }, - { - Codec: "avc3.640028", Media: desc.Medias[4], - Format: desc.Medias[4].Formats[0], - }, - { - Codec: "opus", Samplerate: 48000, Channels: 2, - Media: desc.Medias[5], Format: desc.Medias[5].Formats[0], - }, - { - Codec: "flac", Samplerate: 44100, Channels: 2, - InitData: base64.StdEncoding.EncodeToString(flacEnc), Media: desc.Medias[6], Format: desc.Medias[6].Formats[0], - }, - { - Codec: "mp4a.40.2", Samplerate: 44100, Channels: 2, - InitData: base64.StdEncoding.EncodeToString(mpeg4Enc), Media: desc.Medias[7], Format: desc.Medias[7].Formats[0], - }, - { - Codec: "pcm-s16", Samplerate: 8000, Channels: 1, - Media: desc.Medias[8], Format: desc.Medias[8].Formats[0], - }, - { - Codec: "pcm-s16", Samplerate: 44100, Channels: 2, - Media: desc.Medias[9], Format: desc.Medias[9].Formats[0], - }, - }, tracks) + }, cat) } diff --git a/internal/protocols/moq/to_stream.go b/internal/protocols/moq/to_stream.go index e69672c9..472e1709 100644 --- a/internal/protocols/moq/to_stream.go +++ b/internal/protocols/moq/to_stream.go @@ -25,11 +25,8 @@ func findTimestamp(props []property.Property) (int64, bool) { return 0, false } -// ToStream maps a Media-over-QUIC stream to a MediaMTX stream. -func ToStream( - cat *catalog.Catalog, - subStream **stream.SubStream, -) ( +// ToStream maps a Media-over-QUIC catalog and published tracks to a MediaMTX stream. +func ToStream(cat *catalog.Catalog, subStream **stream.SubStream) ( []*description.Media, map[uint64]func(sg *subgroup.SubGroup) error, error, diff --git a/internal/protocols/moq/track.go b/internal/protocols/moq/track.go deleted file mode 100644 index 8c011caa..00000000 --- a/internal/protocols/moq/track.go +++ /dev/null @@ -1,18 +0,0 @@ -package moq - -import ( - "github.com/bluenviron/gortsplib/v5/pkg/description" - "github.com/bluenviron/gortsplib/v5/pkg/format" - "github.com/bluenviron/mediamtx/internal/unit" -) - -// Track is a Media-over-QUIC track. -type Track struct { - Codec string - Samplerate int - Channels int - InitData string - Media *description.Media - Format format.Format - OnData func(u *unit.Unit, wrapped func([]byte, int64) error) error -} diff --git a/internal/servers/moq/http_server.go b/internal/servers/moq/http_server.go index e88fd07b..5bab92fd 100644 --- a/internal/servers/moq/http_server.go +++ b/internal/servers/moq/http_server.go @@ -37,27 +37,41 @@ var readerJS []byte var publisherJS []byte const ( - moqtVersion = "moqt-18" wtProtocolHeader = "WT-Protocol" wtAvailableProtocolsHeader = "WT-Available-Protocols" ) +// ordered from most preferred to least preferred +var supportedMoqtVersions = []defs.APIMoQVersion{ + defs.APIMoQVersionDraft19, + defs.APIMoQVersionDraft18, +} + type ginUnwrapper interface { Unwrap() http.ResponseWriter } -func containsMoqtVersion(header string) bool { +func selectMoqtVersion(header string) defs.APIMoQVersion { + available := make(map[string]struct{}) + for item := range strings.SplitSeq(header, ",") { item = strings.TrimSpace(item) if i := strings.IndexByte(item, ';'); i >= 0 { item = item[:i] } item = strings.Trim(strings.TrimSpace(item), `"`) - if item == moqtVersion { - return true + if item != "" { + available[item] = struct{}{} } } - return false + + for _, version := range supportedMoqtVersions { + if _, ok := available[string(version)]; ok { + return version + } + } + + return "" } func trailingSlashLocation(rawPath string, rawQuery string) string { @@ -308,19 +322,22 @@ func (s *httpServer) onRequestHTTPS3(ctx *gin.Context) { return } - if offered := ctx.Request.Header.Get(wtAvailableProtocolsHeader); offered != "" { - if !containsMoqtVersion(offered) { - s.writeErrorNoLog(ctx, http.StatusBadRequest, - fmt.Errorf("no supported MoQ version in %s", wtAvailableProtocolsHeader)) - return - } + offered := ctx.Request.Header.Get(wtAvailableProtocolsHeader) + if offered == "" { + s.writeErrorNoLog(ctx, http.StatusBadRequest, + fmt.Errorf("missing %s header", wtAvailableProtocolsHeader)) + return + } + + version := selectMoqtVersion(offered) + if version == "" { + s.writeErrorNoLog(ctx, http.StatusBadRequest, + fmt.Errorf("no supported MoQ version in %s", wtAvailableProtocolsHeader)) + return } w := ctx.Writer.(ginUnwrapper).Unwrap() - - if offered := ctx.Request.Header.Get(wtAvailableProtocolsHeader); offered != "" { - w.Header().Set(wtProtocolHeader, `"`+moqtVersion+`"`) - } + w.Header().Set(wtProtocolHeader, `"`+string(version)+`"`) wt, err := s.innerHTTP3.Upgrade(w, ctx.Request) if err != nil { @@ -332,6 +349,7 @@ func (s *httpServer) onRequestHTTPS3(ctx *gin.Context) { pathName: pathName, query: ctx.Request.URL.RawQuery, userAgent: ctx.Request.Header.Get("User-Agent"), + version: version, wt: wt, }) if res.err != nil { diff --git a/internal/servers/moq/publisher.js b/internal/servers/moq/publisher.js index ec9e1641..afa64e66 100644 --- a/internal/servers/moq/publisher.js +++ b/internal/servers/moq/publisher.js @@ -34,7 +34,7 @@ class MediaMTXMoQPublisher { static #RETRY_PAUSE = 2000; - static #MOQT_VERSION = "moqt-18"; + static #MOQT_VERSION = "moqt-19"; static #SETUP_TYPE = 0x2f00n; static #MSG_PUBLISH = 0x1dn; diff --git a/internal/servers/moq/reader.js b/internal/servers/moq/reader.js index ed5f7b9d..de9452b4 100644 --- a/internal/servers/moq/reader.js +++ b/internal/servers/moq/reader.js @@ -33,7 +33,7 @@ class MediaMTXMoQReader { static #RETRY_PAUSE = 2000; - static #MOQT_VERSION = "moqt-18"; + static #MOQT_VERSION = "moqt-19"; static #SETUP_TYPE = 0x2f00n; static #MSG_SUBSCRIBE = 0x03n; diff --git a/internal/servers/moq/server.go b/internal/servers/moq/server.go index 9de34348..d388b440 100644 --- a/internal/servers/moq/server.go +++ b/internal/servers/moq/server.go @@ -37,6 +37,7 @@ type newSessionReq struct { pathName string query string userAgent string + version defs.APIMoQVersion wt *webtransport.Session res chan newSessionRes } @@ -180,6 +181,7 @@ outer: pathName: req.pathName, query: req.query, userAgent: req.userAgent, + version: req.version, pathManager: s.PathManager, parent: s, } diff --git a/internal/servers/moq/server_test.go b/internal/servers/moq/server_test.go index 13b64056..0988c74b 100644 --- a/internal/servers/moq/server_test.go +++ b/internal/servers/moq/server_test.go @@ -153,7 +153,7 @@ func TestAuthError(t *testing.T) { EnableDatagrams: true, EnableStreamResetPartialDelivery: true, }, - ApplicationProtocols: []string{"moqt-18"}, + ApplicationProtocols: []string{"moqt-19"}, } defer d.Close() //nolint:errcheck @@ -263,33 +263,226 @@ func TestAuthError(t *testing.T) { } func TestServer(t *testing.T) { - desc := &description.Session{Medias: []*description.Media{test.UniqueMediaH264()}} - strm := &stream.Stream{ - OrigDesc: desc, - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Parent: test.NilLogger, - } - err := strm.Initialize() - require.NoError(t, err) - defer strm.Close() - - subStream := &stream.SubStream{ - Stream: strm, - UseRTPPackets: false, - } - err = subStream.Initialize() - require.NoError(t, err) - - pm := &test.PathManager{ - FindPathConfImpl: func(_ defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { - return &defs.PathFindPathConfRes{Conf: &conf.Path{}}, nil + for _, ca := range []struct { + name string + clientProtocols []string + expectedVersion defs.APIMoQVersion + }{ + { + name: "draft-18", + clientProtocols: []string{"moqt-18"}, + expectedVersion: defs.APIMoQVersionDraft18, }, - AddReaderImpl: func(_ defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { - return &defs.PathAddReaderRes{Path: &serverDummyPath{}, Stream: strm}, nil + { + name: "draft-19", + clientProtocols: []string{"moqt-19"}, + expectedVersion: defs.APIMoQVersionDraft19, }, - } + { + name: "draft-19-preferred", + clientProtocols: []string{"moqt-19", "moqt-18"}, + expectedVersion: defs.APIMoQVersionDraft19, + }, + } { + t.Run(ca.name, func(t *testing.T) { + desc := &description.Session{Medias: []*description.Media{test.UniqueMediaH264()}} + strm := &stream.Stream{ + OrigDesc: desc, + WriteQueueSize: 512, + RTPMaxPayloadSize: 1450, + Parent: test.NilLogger, + } + err := strm.Initialize() + require.NoError(t, err) + defer strm.Close() + subStream := &stream.SubStream{ + Stream: strm, + UseRTPPackets: false, + } + err = subStream.Initialize() + require.NoError(t, err) + + pm := &test.PathManager{ + FindPathConfImpl: func(_ defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { + return &defs.PathFindPathConfRes{Conf: &conf.Path{}}, nil + }, + AddReaderImpl: func(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { + require.Equal(t, ca.expectedVersion, req.Author.(*session).version) + return &defs.PathAddReaderRes{Path: &serverDummyPath{}, Stream: strm}, nil + }, + } + + serverCertFile := test.CreateTempFile(t, test.TLSCertPub) + serverKeyFile := test.CreateTempFile(t, test.TLSCertKey) + + s := &Server{ + HTTP2Address: "127.0.0.1:19895", + HTTP3Address: "127.0.0.1:19896", + ServerCert: serverCertFile, + ServerKey: serverKeyFile, + AllowOrigins: []string{"*"}, + TrustedProxies: conf.IPNetworks{}, + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + PathManager: pm, + Parent: test.NilLogger, + } + err = s.Initialize() + require.NoError(t, err) + defer s.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + d := &webtransport.Dialer{ + TLSClientConfig: &tls.Config{ + InsecureSkipVerify: true, //nolint:gosec + }, + QUICConfig: &quic.Config{ + EnableDatagrams: true, + EnableStreamResetPartialDelivery: true, + }, + ApplicationProtocols: ca.clientProtocols, + } + defer d.Close() //nolint:errcheck + + res, sx, err := d.Dial(ctx, "https://127.0.0.1:19896/teststream/moq", nil) + require.NoError(t, err) + defer sx.CloseWithError(0, "") //nolint:errcheck + defer res.Body.Close() //nolint:errcheck + + require.Equal(t, `"`+string(ca.expectedVersion)+`"`, res.Header.Get("WT-Protocol")) + + setupStream, err := sx.AcceptUniStream(ctx) + require.NoError(t, err) + + setupMsg, err := controlmessage.Read(setupStream) + require.NoError(t, err) + require.Equal(t, &controlmessage.Setup{}, setupMsg) + + clientSetup, err := sx.OpenUniStreamSync(ctx) + require.NoError(t, err) + + _, err = clientSetup.Write(controlmessage.Setup{}.Marshal()) + require.NoError(t, err) + + catalogBidi, err := sx.OpenStreamSync(ctx) + require.NoError(t, err) + + _, err = catalogBidi.Write(controlmessage.Subscribe{ + RequestID: 1, + TrackName: ".catalog", + }.Marshal()) + require.NoError(t, err) + + catalogOkMsg, err := controlmessage.Read(catalogBidi) + require.NoError(t, err) + require.Equal(t, &controlmessage.SubscribeOk{TrackAlias: 1}, catalogOkMsg) + + catalogDataStream, err := sx.AcceptUniStream(ctx) + require.NoError(t, err) + + var catalogSG subgroup.SubGroup + err = catalogSG.Read(catalogDataStream) + require.NoError(t, err) + + var cat catalog.Catalog + err = json.Unmarshal(catalogSG.Objects[0].Payload, &cat) + require.NoError(t, err) + + require.Equal(t, catalog.Catalog{ + Version: 1, + Tracks: []catalog.Track{{ + Name: "0", + Packaging: "loc", + IsLive: true, + Codec: "avc3.640028", + }}, + }, cat) + + trackBidi, err := sx.OpenStreamSync(ctx) + require.NoError(t, err) + + _, err = trackBidi.Write(controlmessage.Subscribe{ + RequestID: 2, + TrackName: "0", + }.Marshal()) + require.NoError(t, err) + + trackOkMsg, err := controlmessage.Read(trackBidi) + require.NoError(t, err) + require.Equal(t, &controlmessage.SubscribeOk{TrackAlias: 2}, trackOkMsg) + + trackBidi2, err := sx.OpenStreamSync(ctx) + require.NoError(t, err) + + _, err = trackBidi2.Write(controlmessage.Subscribe{ + RequestID: 3, + TrackName: "0", + }.Marshal()) + require.NoError(t, err) + + trackOkMsg2, err := controlmessage.Read(trackBidi2) + require.NoError(t, err) + require.Equal(t, &controlmessage.SubscribeOk{TrackAlias: 3}, trackOkMsg2) + + go func() { + time.Sleep(200 * time.Millisecond) + subStream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.Unit{ + PTS: 0, + Payload: unit.PayloadH264{{5, 1}}, + }) + }() + + frameStream, err := sx.AcceptUniStream(ctx) + require.NoError(t, err) + + var frameSG subgroup.SubGroup + err = frameSG.Read(frameStream) + require.NoError(t, err) + + frameStream2, err := sx.AcceptUniStream(ctx) + require.NoError(t, err) + + var frameSG2 subgroup.SubGroup + err = frameSG2.Read(frameStream2) + require.NoError(t, err) + + expectedPayload, err2 := mch264.AVCC([][]byte{test.FormatH264.SPS, test.FormatH264.PPS, {5, 1}}).Marshal() + require.NoError(t, err2) + require.Equal(t, expectedPayload, frameSG.Objects[0].Payload) + require.Equal(t, expectedPayload, frameSG2.Objects[0].Payload) + require.ElementsMatch(t, []uint64{2, 3}, []uint64{frameSG.Header.TrackAlias, frameSG2.Header.TrackAlias}) + + trackBidi.Close() //nolint:errcheck + time.Sleep(100 * time.Millisecond) + + go func() { + time.Sleep(200 * time.Millisecond) + subStream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.Unit{ + PTS: 1, + Payload: unit.PayloadH264{{5, 2}}, + }) + }() + + frameStream3, err := sx.AcceptUniStream(ctx) + require.NoError(t, err) + + var frameSG3 subgroup.SubGroup + err = frameSG3.Read(frameStream3) + require.NoError(t, err) + + expectedPayload2, err2 := mch264.AVCC([][]byte{test.FormatH264.SPS, test.FormatH264.PPS, {5, 2}}).Marshal() + require.NoError(t, err2) + require.Equal(t, expectedPayload2, frameSG3.Objects[0].Payload) + require.Equal(t, uint64(3), frameSG3.Header.TrackAlias) + trackBidi2.Close() //nolint:errcheck + }) + } +} + +func TestServerUnsupportedVersion(t *testing.T) { serverCertFile := test.CreateTempFile(t, test.TLSCertPub) serverKeyFile := test.CreateTempFile(t, test.TLSCertKey) @@ -302,10 +495,10 @@ func TestServer(t *testing.T) { TrustedProxies: conf.IPNetworks{}, ReadTimeout: conf.Duration(10 * time.Second), WriteTimeout: conf.Duration(10 * time.Second), - PathManager: pm, + PathManager: &test.PathManager{}, Parent: test.NilLogger, } - err = s.Initialize() + err := s.Initialize() require.NoError(t, err) defer s.Close() @@ -313,98 +506,16 @@ func TestServer(t *testing.T) { defer cancel() d := &webtransport.Dialer{ - TLSClientConfig: &tls.Config{ - InsecureSkipVerify: true, //nolint:gosec - }, + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec QUICConfig: &quic.Config{ EnableDatagrams: true, EnableStreamResetPartialDelivery: true, }, - ApplicationProtocols: []string{"moqt-18"}, + ApplicationProtocols: []string{"moqt-20"}, } defer d.Close() //nolint:errcheck - res, sx, err := d.Dial(ctx, "https://127.0.0.1:19896/teststream/moq", nil) - require.NoError(t, err) - defer sx.CloseWithError(0, "") //nolint:errcheck - defer res.Body.Close() //nolint:errcheck - - setupStream, err := sx.AcceptUniStream(ctx) - require.NoError(t, err) - - setupMsg, err := controlmessage.Read(setupStream) - require.NoError(t, err) - require.Equal(t, &controlmessage.Setup{}, setupMsg) - - clientSetup, err := sx.OpenUniStreamSync(ctx) - require.NoError(t, err) - - _, err = clientSetup.Write(controlmessage.Setup{}.Marshal()) - require.NoError(t, err) - - catalogBidi, err := sx.OpenStreamSync(ctx) - require.NoError(t, err) - - _, err = catalogBidi.Write(controlmessage.Subscribe{ - RequestID: 1, - TrackName: ".catalog", - }.Marshal()) - require.NoError(t, err) - - catalogOkMsg, err := controlmessage.Read(catalogBidi) - require.NoError(t, err) - require.Equal(t, &controlmessage.SubscribeOk{TrackAlias: 1}, catalogOkMsg) - - catalogDataStream, err := sx.AcceptUniStream(ctx) - require.NoError(t, err) - - var catalogSG subgroup.SubGroup - err = catalogSG.Read(catalogDataStream) - require.NoError(t, err) - - var cat catalog.Catalog - err = json.Unmarshal(catalogSG.Objects[0].Payload, &cat) - require.NoError(t, err) - - require.Equal(t, catalog.Catalog{ - Version: 1, - Tracks: []catalog.Track{{ - Name: "0", - Packaging: "loc", - IsLive: true, - Codec: "avc3.640028", - }}, - }, cat) - - trackBidi, err := sx.OpenStreamSync(ctx) - require.NoError(t, err) - - _, err = trackBidi.Write(controlmessage.Subscribe{ - RequestID: 2, - TrackName: "0", - }.Marshal()) - require.NoError(t, err) - - trackOkMsg, err := controlmessage.Read(trackBidi) - require.NoError(t, err) - require.Equal(t, &controlmessage.SubscribeOk{TrackAlias: 2}, trackOkMsg) - - go func() { - time.Sleep(200 * time.Millisecond) - subStream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.Unit{ - PTS: 0, - Payload: unit.PayloadH264{{5, 1}}, - }) - }() - - frameStream, err := sx.AcceptUniStream(ctx) - require.NoError(t, err) - - var frameSG subgroup.SubGroup - err = frameSG.Read(frameStream) - require.NoError(t, err) - - expectedPayload, err2 := mch264.AVCC([][]byte{test.FormatH264.SPS, test.FormatH264.PPS, {5, 1}}).Marshal() - require.NoError(t, err2) - require.Equal(t, expectedPayload, frameSG.Objects[0].Payload) + res, _, err := d.Dial(ctx, "https://127.0.0.1:19896/teststream/moq", nil) + require.Error(t, err) + defer res.Body.Close() } diff --git a/internal/servers/moq/session.go b/internal/servers/moq/session.go index 4357a0ac..1575907f 100644 --- a/internal/servers/moq/session.go +++ b/internal/servers/moq/session.go @@ -27,7 +27,6 @@ import ( "github.com/bluenviron/mediamtx/internal/protocols/moq/property" "github.com/bluenviron/mediamtx/internal/protocols/moq/subgroup" "github.com/bluenviron/mediamtx/internal/stream" - "github.com/bluenviron/mediamtx/internal/unit" "github.com/google/uuid" "github.com/quic-go/webtransport-go" "golang.org/x/sync/errgroup" @@ -72,28 +71,6 @@ func credentialsFromAuthorizationToken(authorization *parameter.AuthorizationTok return &auth.Credentials{} } -func tracksToCatalog(tracks []*moq.Track) (catalog.Catalog, error) { - cat := catalog.Catalog{ - Version: 1, - Tracks: make([]catalog.Track, len(tracks)), - } - - for i, track := range tracks { - ct := catalog.Track{ - Name: strconv.Itoa(i), - Packaging: "loc", - IsLive: true, - Codec: track.Codec, - Samplerate: track.Samplerate, - Channels: track.Channels, - InitData: track.InitData, - } - cat.Tracks[i] = ct - } - - return cat, nil -} - func isSubGroupStream(b byte) bool { return (b & 0x90) == 0x10 } @@ -108,22 +85,22 @@ type session struct { pathName string query string userAgent string + version defs.APIMoQVersion pathManager serverPathManager parent sessionParent - ctx context.Context - ctxCancel context.CancelFunc - created time.Time - uuid uuid.UUID - mutex sync.Mutex - state defs.APIMoQSessionState - path defs.Path - stream *stream.Stream // read only - tracks []*moq.Track // read only - trackSubscriptions map[int]struct{} // read only - catalogReceived chan []byte // publish only - publishReady chan struct{} // publish only - inboundTracks map[uint64]*inboundTrack // publish only + ctx context.Context + ctxCancel context.CancelFunc + created time.Time + uuid uuid.UUID + mutex sync.Mutex + state defs.APIMoQSessionState + path defs.Path + stream *stream.Stream // read only + setupTracks []moq.SetupTrackFunc // read only + catalogReceived chan *catalog.Catalog // publish only + publishReady chan struct{} // publish only + inboundTracks map[uint64]*inboundTrack // publish only inboundBytes atomic.Uint64 outboundBytes atomic.Uint64 @@ -137,9 +114,8 @@ func (s *session) initialize() { s.created = time.Now() s.uuid = uuid.New() s.state = defs.APIMoQSessionStateIdle - s.trackSubscriptions = make(map[int]struct{}) - s.catalogReceived = make(chan []byte, 1) + s.catalogReceived = make(chan *catalog.Catalog, 1) s.publishReady = make(chan struct{}) s.setupReceived = make(chan struct{}) @@ -379,9 +355,7 @@ func (s *session) onSubscribeCatalog(wstream *webtransport.Stream, m *controlmes return err } - tracks, err := moq.FromStream( - addRes.Stream.OrigDesc, - ) + cat, setupTracks, err := moq.FromStream(addRes.Stream.OrigDesc) if err != nil { addRes.Path.RemoveReader(defs.PathRemoveReaderReq{Author: s}) return err @@ -390,7 +364,7 @@ func (s *session) onSubscribeCatalog(wstream *webtransport.Stream, m *controlmes s.mutex.Lock() s.path = addRes.Path s.stream = addRes.Stream - s.tracks = tracks + s.setupTracks = setupTracks s.mutex.Unlock() s.Log(logger.Info, "is reading from path %s", s.pathName) @@ -400,11 +374,6 @@ func (s *session) onSubscribeCatalog(wstream *webtransport.Stream, m *controlmes return err } - cat, err := tracksToCatalog(tracks) - if err != nil { - return err - } - enc, err := json.Marshal(cat) if err != nil { return err @@ -456,17 +425,10 @@ func (s *session) onSubscribeTrack(wstream *webtransport.Stream, m *controlmessa return fmt.Errorf("stream not ready") } - if trackID >= len(s.tracks) { + if trackID >= len(s.setupTracks) { return fmt.Errorf("track index %d out of range", trackID) } - _, ok := s.trackSubscriptions[trackID] - if ok { - return fmt.Errorf("already subscribed to track %d", trackID) - } - - s.trackSubscriptions[trackID] = struct{}{} - return nil }() if err != nil { @@ -475,10 +437,9 @@ func (s *session) onSubscribeTrack(wstream *webtransport.Stream, m *controlmessa r := &stream.Reader{Parent: s} - track := s.tracks[trackID] groupID := uint64(0) - wrapped := func(payload []byte, pts int64) error { + writeData := func(payload []byte, pts int64) error { wstream, err2 := s.wt.OpenUniStreamSync(context.Background()) if err2 != nil { return err2 @@ -509,9 +470,7 @@ func (s *session) onSubscribeTrack(wstream *webtransport.Stream, m *controlmessa return err2 } - r.OnData(track.Media, track.Format, func(u *unit.Unit) error { - return track.OnData(u, wrapped) - }) + s.setupTracks[trackID](r, writeData) s.stream.AddReader(r) defer s.stream.RemoveReader(r) @@ -531,7 +490,7 @@ func (s *session) onSubscribeTrack(wstream *webtransport.Stream, m *controlmessa case err = <-r.Error(): return err case <-streamClosed: - return fmt.Errorf("SUBSCRIBE track stream closed") + return nil case <-s.ctx.Done(): return fmt.Errorf("terminated") } @@ -547,16 +506,10 @@ func (s *session) onPublishCatalog(wstream *webtransport.Stream, m *controlmessa s.mutex.Unlock() select { - case catalogData := <-s.catalogReceived: - var cat catalog.Catalog - err := json.Unmarshal(catalogData, &cat) - if err != nil { - return fmt.Errorf("failed to parse catalog JSON: %w", err) - } - + case cat := <-s.catalogReceived: var subStream *stream.SubStream - medias, writeFuncs, err := moq.ToStream(&cat, &subStream) + medias, writeFuncs, err := moq.ToStream(cat, &subStream) if err != nil { return err } @@ -666,8 +619,14 @@ func (s *session) onUniSubGroup(r io.Reader) error { } func (s *session) onDataCatalog(r io.Reader, sg *subgroup.SubGroup) error { + var cat catalog.Catalog + err := json.Unmarshal(sg.Objects[0].Payload, &cat) + if err != nil { + return fmt.Errorf("failed to parse catalog JSON: %w", err) + } + select { - case s.catalogReceived <- sg.Objects[0].Payload: + case s.catalogReceived <- &cat: default: return fmt.Errorf("catalog already received") } @@ -714,6 +673,7 @@ func (s *session) apiItem() defs.APIMoQSession { Path: s.pathName, Query: s.query, UserAgent: s.userAgent, + Version: s.version, InboundBytes: s.inboundBytes.Load(), OutboundBytes: s.outboundBytes.Load(), }