From 7e576d077f46f2bc78f1ac1aba7618c7c08a72bd Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Wed, 10 Jun 2026 23:46:02 +0200 Subject: [PATCH] rtsp: use ServerStream.ReloadDesc to propagate description changes to RTSP servers (#5856) --- docs/2-features/06-authentication.md | 3 +- go.mod | 2 +- go.sum | 4 +- internal/core/path.go | 23 ++- internal/defs/api_path_track.go | 19 +- internal/defs/api_path_track_codec_props.go | 54 +++--- internal/protocols/hls/from_stream.go | 73 +++++--- internal/protocols/hls/from_stream_test.go | 8 +- internal/protocols/hls/to_stream_test.go | 2 +- internal/protocols/rtmp/from_stream.go | 152 +++++++-------- internal/protocols/rtmp/from_stream_test.go | 12 +- internal/protocols/webrtc/from_stream_test.go | 16 +- internal/recorder/format_fmp4.go | 139 +++++++------- internal/recorder/format_mpegts.go | 4 +- internal/recorder/recorder_test.go | 14 +- internal/servers/hls/muxer_instance.go | 6 +- internal/servers/hls/server_test.go | 8 +- internal/servers/hls/session.go | 5 +- internal/servers/moq/server_test.go | 5 +- internal/servers/moq/session.go | 2 +- internal/servers/rtmp/conn.go | 8 +- internal/servers/rtmp/server_test.go | 8 +- internal/servers/rtsp/conn.go | 10 +- internal/servers/rtsp/server_test.go | 16 +- internal/servers/rtsp/session.go | 29 ++- internal/servers/srt/conn.go | 2 +- internal/servers/srt/server_test.go | 8 +- internal/servers/webrtc/server_test.go | 14 +- internal/servers/webrtc/session.go | 2 +- internal/staticsources/rpicamera/source.go | 4 +- internal/stream/format_updater.go | 35 ++-- internal/stream/offline_sub_stream.go | 6 +- internal/stream/reader.go | 8 +- internal/stream/stream.go | 173 ++++++++---------- .../stream/stream_alwaysavailable_test.go | 32 ++-- internal/stream/stream_format.go | 15 +- internal/stream/stream_media.go | 28 ++- internal/stream/stream_standard_test.go | 86 ++++----- internal/stream/sub_stream.go | 26 +-- internal/stream/sub_stream_format.go | 53 +++--- internal/stream/sub_stream_media.go | 12 +- internal/test/static_source_parent.go | 2 +- 42 files changed, 596 insertions(+), 532 deletions(-) diff --git a/docs/2-features/06-authentication.md b/docs/2-features/06-authentication.md index dada730c..b96ad0e9 100644 --- a/docs/2-features/06-authentication.md +++ b/docs/2-features/06-authentication.md @@ -79,6 +79,7 @@ Authentication can be delegated to an external HTTP server: ```yml authMethod: http authHTTPAddress: http://myauthserver/auth +authHTTPExclude: [] # explicitly clear authHTTPExclude to validate every action ``` Each time a user needs to be authenticated, the specified URL will be requested with the POST method and this payload: @@ -109,7 +110,7 @@ If the URL returns a status code that begins with `20` (i.e. `200`), authenticat This happens because RTSP clients don't provide credentials until they are asked to. In order to receive the credentials, the authentication server must reply with status code `401`, then the client will send credentials. -Some actions can be excluded from the process: +Some actions can be excluded from the validation process: ```yml # Actions to exclude from HTTP-based authentication. diff --git a/go.mod b/go.mod index dc3af077..a61edb46 100644 --- a/go.mod +++ b/go.mod @@ -12,7 +12,7 @@ require ( github.com/asticode/go-astits v1.15.0 github.com/bluenviron/gohlslib/v2 v2.4.0 github.com/bluenviron/gortmplib v0.4.0 - github.com/bluenviron/gortsplib/v5 v5.5.4 + github.com/bluenviron/gortsplib/v5 v5.5.5-0.20260610213428-fb7a9dca6ded github.com/bluenviron/mediacommon/v2 v2.9.0 github.com/datarhei/gosrt v0.11.0 github.com/fsnotify/fsnotify v1.10.1 diff --git a/go.sum b/go.sum index 43db3fba..70a27711 100644 --- a/go.sum +++ b/go.sum @@ -39,8 +39,8 @@ github.com/bluenviron/gohlslib/v2 v2.4.0 h1:O0bmUvyOMVPuRx3givLqfGWUzOKicKwkSCIn github.com/bluenviron/gohlslib/v2 v2.4.0/go.mod h1:iqU6QJ5geVuCDpUeAbK/Si2RtDC3ocBTp8LnPHBkN6U= github.com/bluenviron/gortmplib v0.4.0 h1:ydwTrOUxFOJkKagzXON2IetEP2yVQwRByzClpapSODc= github.com/bluenviron/gortmplib v0.4.0/go.mod h1:6AAQoyC8vSHnw02fCW6Kjll93Hk1rEH4+S0uy5NFyDM= -github.com/bluenviron/gortsplib/v5 v5.5.4 h1:ooXvRBGPmKHCK9yh90MooeZKZMmaZ63nfITCzTo/vS4= -github.com/bluenviron/gortsplib/v5 v5.5.4/go.mod h1:sT0KqICjZKSpHmx7R3xFRY1tfgP2iMe2bF3i3MxX/jE= +github.com/bluenviron/gortsplib/v5 v5.5.5-0.20260610213428-fb7a9dca6ded h1:RH7gq7G2k4xHpZU9fgBigXvM78NEmE0iJdKDn4wzhCU= +github.com/bluenviron/gortsplib/v5 v5.5.5-0.20260610213428-fb7a9dca6ded/go.mod h1:sT0KqICjZKSpHmx7R3xFRY1tfgP2iMe2bF3i3MxX/jE= github.com/bluenviron/mediacommon/v2 v2.9.0 h1:8gBeby10B/qD8ctnesKTHx3pgdsxiZcaQZe+RR6Djx4= github.com/bluenviron/mediacommon/v2 v2.9.0/go.mod h1:syqw4j2RmBCKZgUtGFtB/7gcnh1aaX8I3vd24+bgXWc= github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= diff --git a/internal/core/path.go b/internal/core/path.go index 2ac04811..272c5b25 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -427,9 +427,11 @@ func (pa *path) doSourceStaticSetReady(req defs.PathSourceStaticSetReadyReq) { Stream: pa.stream, UseRTPPackets: req.UseRTPPackets, } + if pa.conf.AlwaysAvailable { - subStream.CurDesc = req.Desc + subStream.InDesc = req.Desc } + err := subStream.Initialize() if err != nil { req.Res <- defs.PathSourceStaticSetReadyRes{Err: err} @@ -547,9 +549,11 @@ func (pa *path) doAddPublisher(req defs.PathAddPublisherReq) { Stream: pa.stream, UseRTPPackets: req.UseRTPPackets, } + if pa.conf.AlwaysAvailable { - subStream.CurDesc = req.Desc + subStream.InDesc = req.Desc } + err := subStream.Initialize() if err != nil { req.Res <- defs.PathAddPublisherRes{Err: err} @@ -621,6 +625,11 @@ func (pa *path) doRemoveReader(req defs.PathRemoveReaderReq) { } func (pa *path) doAPIPathsGet(req pathAPIPathsGetReq) { + var outDesc *description.Session + if pa.isAvailable() { + outDesc = pa.stream.OutDescCopy() + } + req.res <- pathAPIPathsGetRes{ data: &defs.APIPath{ Name: pa.name, @@ -660,13 +669,13 @@ func (pa *path) doAPIPathsGet(req pathAPIPathsGetReq) { if !pa.isAvailable() { return []defs.APIPathTrackCodec{} } - return formatlabel.MediasToLabels(pa.stream.Desc.Medias) + return formatlabel.MediasToLabels(outDesc.Medias) }(), Tracks2: func() []defs.APIPathTrack { if !pa.isAvailable() { return []defs.APIPathTrack{} } - return defs.MediasToTracks(pa.stream.Desc.Medias) + return defs.MediasToTracks(outDesc.Medias) }(), InboundBytes: func() uint64 { if !pa.isAvailable() { @@ -820,7 +829,7 @@ func (pa *path) setAvailable( replaceNTP bool, ) error { pa.stream = &stream.Stream{ - Desc: desc, + OrigDesc: desc, AlwaysAvailable: pa.conf.AlwaysAvailable, AlwaysAvailableTracks: pa.conf.AlwaysAvailableTracks, AlwaysAvailableFile: pa.conf.AlwaysAvailableFile, @@ -859,9 +868,9 @@ func (pa *path) setAvailable( }) if pa.conf.AlwaysAvailable { - pa.Log(logger.Info, "stream is available, %s", defs.MediasInfo(pa.stream.Desc.Medias)) + pa.Log(logger.Info, "stream is available, %s", defs.MediasInfo(pa.stream.OrigDesc.Medias)) } else { - pa.Log(logger.Info, "stream is available and online, %s", defs.MediasInfo(pa.stream.Desc.Medias)) + pa.Log(logger.Info, "stream is available and online, %s", defs.MediasInfo(pa.stream.OrigDesc.Medias)) } pa.parent.setPathReady(pa) diff --git a/internal/defs/api_path_track.go b/internal/defs/api_path_track.go index 4e893ae2..dff34940 100644 --- a/internal/defs/api_path_track.go +++ b/internal/defs/api_path_track.go @@ -12,10 +12,10 @@ type APIPathTrack struct { CodecProps APIPathTrackCodecProps `json:"codecProps"` } -func formatToTrack(forma format.Format) APIPathTrack { +func formatToTrack(outFormat format.Format) APIPathTrack { return APIPathTrack{ - Codec: formatlabel.FormatToLabel(forma), - CodecProps: formatToTrackCodecProps(forma), + Codec: formatlabel.FormatToLabel(outFormat), + CodecProps: formatToTrackCodecProps(outFormat), } } @@ -39,18 +39,17 @@ func gatherFormats(medias []*description.Media) []format.Format { return formats } -// FormatsToTracks returns tracks of given formats. -func FormatsToTracks(formats []format.Format) []APIPathTrack { - ret := make([]APIPathTrack, len(formats)) +func formatsToTracks(outFormats []format.Format) []APIPathTrack { + ret := make([]APIPathTrack, len(outFormats)) - for i, forma := range formats { - ret[i] = formatToTrack(forma) + for i, outFormat := range outFormats { + ret[i] = formatToTrack(outFormat) } return ret } // MediasToTracks returns tracks of given medias. -func MediasToTracks(medias []*description.Media) []APIPathTrack { - return FormatsToTracks(gatherFormats(medias)) +func MediasToTracks(outMedias []*description.Media) []APIPathTrack { + return formatsToTracks(gatherFormats(outMedias)) } diff --git a/internal/defs/api_path_track_codec_props.go b/internal/defs/api_path_track_codec_props.go index 0bec31e9..2581aa2f 100644 --- a/internal/defs/api_path_track_codec_props.go +++ b/internal/defs/api_path_track_codec_props.go @@ -160,19 +160,19 @@ func h265LevelString(levelIdc uint8) string { return fmt.Sprintf("%.1f", level) } -func formatToTrackCodecProps(forma format.Format) APIPathTrackCodecProps { - switch forma := forma.(type) { +func formatToTrackCodecProps(outFormat format.Format) APIPathTrackCodecProps { + switch outFormat := outFormat.(type) { case *format.AV1: props := &APIPathTrackCodecPropsAV1{} - if forma.Profile != nil { - props.Profile = *forma.Profile + if outFormat.Profile != nil { + props.Profile = *outFormat.Profile } - if forma.LevelIdx != nil { - props.Level = *forma.LevelIdx + if outFormat.LevelIdx != nil { + props.Level = *outFormat.LevelIdx } - if forma.Tier != nil { - props.Tier = *forma.Tier + if outFormat.Tier != nil { + props.Tier = *outFormat.Tier } return props @@ -180,8 +180,8 @@ func formatToTrackCodecProps(forma format.Format) APIPathTrackCodecProps { case *format.VP9: props := &APIPathTrackCodecPropsVP9{} - if forma.ProfileID != nil { - props.Profile = *forma.ProfileID + if outFormat.ProfileID != nil { + props.Profile = *outFormat.ProfileID } return props @@ -189,10 +189,9 @@ func formatToTrackCodecProps(forma format.Format) APIPathTrackCodecProps { case *format.H265: props := &APIPathTrackCodecPropsH265{} - _, sps, _ := forma.SafeParams() - if sps != nil { + if outFormat.SPS != nil { var s codecsh265.SPS - if err := s.Unmarshal(sps); err == nil { + if err := s.Unmarshal(outFormat.SPS); err == nil { props.Width = s.Width() props.Height = s.Height() props.Profile = h265ProfileString(s.ProfileTierLevel.GeneralProfileIdc) @@ -205,10 +204,9 @@ func formatToTrackCodecProps(forma format.Format) APIPathTrackCodecProps { case *format.H264: props := &APIPathTrackCodecPropsH264{} - sps, _ := forma.SafeParams() - if sps != nil { + if outFormat.SPS != nil { var s codecsh264.SPS - if err := s.Unmarshal(sps); err == nil { + if err := s.Unmarshal(outFormat.SPS); err == nil { props.Width = s.Width() props.Height = s.Height() props.Profile = h264ProfileString(s.ProfileIdc) @@ -220,37 +218,37 @@ func formatToTrackCodecProps(forma format.Format) APIPathTrackCodecProps { case *format.Opus: return &APIPathTrackCodecPropsOpus{ - ChannelCount: forma.ChannelCount, + ChannelCount: outFormat.ChannelCount, } case *format.MPEG4Audio: props := &APIPathTrackCodecPropsMPEG4Audio{} - if forma.Config != nil { - props.SampleRate = forma.Config.SampleRate - props.ChannelCount = int(forma.Config.ChannelConfig) + if outFormat.Config != nil { + props.SampleRate = outFormat.Config.SampleRate + props.ChannelCount = int(outFormat.Config.ChannelConfig) } return props case *format.AC3: return &APIPathTrackCodecPropsAC3{ - SampleRate: forma.SampleRate, - ChannelCount: forma.ChannelCount, + SampleRate: outFormat.SampleRate, + ChannelCount: outFormat.ChannelCount, } case *format.G711: return &APIPathTrackCodecPropsG711{ - MULaw: forma.MULaw, - SampleRate: forma.SampleRate, - ChannelCount: forma.ChannelCount, + MULaw: outFormat.MULaw, + SampleRate: outFormat.SampleRate, + ChannelCount: outFormat.ChannelCount, } case *format.LPCM: return &APIPathTrackCodecPropsLPCM{ - BitDepth: forma.BitDepth, - SampleRate: forma.SampleRate, - ChannelCount: forma.ChannelCount, + BitDepth: outFormat.BitDepth, + SampleRate: outFormat.SampleRate, + ChannelCount: outFormat.ChannelCount, } default: diff --git a/internal/protocols/hls/from_stream.go b/internal/protocols/hls/from_stream.go index 20f53d43..2fd49207 100644 --- a/internal/protocols/hls/from_stream.go +++ b/internal/protocols/hls/from_stream.go @@ -5,6 +5,7 @@ import ( "encoding/hex" "errors" "fmt" + "reflect" "slices" "strings" @@ -24,8 +25,30 @@ import ( var ErrNoSupportedCodecs = errors.New( "the stream doesn't contain any supported codec, which are currently AV1, VP9, H265, H264, Opus, MPEG-4 Audio, KLV") +func findFormatAndIndexInMedia(media *description.Media, forma any) int { + for i, forma2 := range media.Formats { + if reflect.TypeOf(forma2) == reflect.TypeOf(forma).Elem() { + reflect.ValueOf(forma).Elem().Set(reflect.ValueOf(forma2)) + return i + } + } + + return -1 +} + +func findFormatAndIndexes(d *description.Session, forma any) (*description.Media, int, int) { + for i, media := range d.Medias { + j := findFormatAndIndexInMedia(media, forma) + if j >= 0 { + return media, i, j + } + } + return nil, -1, -1 +} + func setupVideoTrack( - desc *description.Session, + origDesc *description.Session, + outDesc *description.Session, r *stream.Reader, muxer *gohlslib.Muxer, ) { @@ -40,7 +63,7 @@ func setupVideoTrack( } var videoFormatAV1 *format.AV1 - videoMedia := desc.FindFormat(&videoFormatAV1) + videoMedia, _, _ := findFormatAndIndexes(origDesc, &videoFormatAV1) if videoFormatAV1 != nil { track := &gohlslib.Track{ @@ -73,7 +96,7 @@ func setupVideoTrack( } var videoFormatVP9 *format.VP9 - videoMedia = desc.FindFormat(&videoFormatVP9) + videoMedia, _, _ = findFormatAndIndexes(origDesc, &videoFormatVP9) if videoFormatVP9 != nil { track := &gohlslib.Track{ @@ -106,15 +129,16 @@ func setupVideoTrack( } var videoFormatH265 *format.H265 - videoMedia = desc.FindFormat(&videoFormatH265) + videoMedia, i, j := findFormatAndIndexes(origDesc, &videoFormatH265) if videoFormatH265 != nil { - vps, sps, pps := videoFormatH265.SafeParams() + outFormat := outDesc.Medias[i].Formats[j].(*format.H265) + track := &gohlslib.Track{ Codec: &codecs.H265{ - VPS: vps, - SPS: sps, - PPS: pps, + VPS: outFormat.VPS, + SPS: outFormat.SPS, + PPS: outFormat.PPS, }, ClockRate: videoFormatH265.ClockRate(), } @@ -144,14 +168,15 @@ func setupVideoTrack( } var videoFormatH264 *format.H264 - videoMedia = desc.FindFormat(&videoFormatH264) + videoMedia, i, j = findFormatAndIndexes(origDesc, &videoFormatH264) if videoFormatH264 != nil { - sps, pps := videoFormatH264.SafeParams() + outFormat := outDesc.Medias[i].Formats[j].(*format.H264) + track := &gohlslib.Track{ Codec: &codecs.H264{ - SPS: sps, - PPS: pps, + SPS: outFormat.SPS, + PPS: outFormat.PPS, }, ClockRate: videoFormatH264.ClockRate(), } @@ -182,7 +207,8 @@ func setupVideoTrack( } func setupAudioTracks( - desc *description.Session, + origDesc *description.Session, + _ *description.Session, r *stream.Reader, muxer *gohlslib.Muxer, ) error { @@ -196,7 +222,7 @@ func setupAudioTracks( r.OnData(medi, forma, onData) } - for _, media := range desc.Medias { + for _, media := range origDesc.Medias { for _, forma := range media.Formats { switch forma := forma.(type) { case *format.Opus: @@ -335,7 +361,8 @@ func setupAudioTracks( } func setupDataTracks( - desc *description.Session, + origDesc *description.Session, + _ *description.Session, r *stream.Reader, muxer *gohlslib.Muxer, ) { @@ -349,7 +376,7 @@ func setupDataTracks( r.OnData(media, forma, onData) } - for _, media := range desc.Medias { + for _, media := range origDesc.Medias { for _, forma := range media.Formats { if forma, ok := forma.(*format.KLV); ok && muxer.Variant == gohlslib.MuxerVariantMPEGTS { track := &gohlslib.Track{ @@ -384,18 +411,21 @@ func setupDataTracks( // FromStream maps a MediaMTX stream to a HLS muxer. func FromStream( - desc *description.Session, + origDesc *description.Session, + outDesc *description.Session, r *stream.Reader, muxer *gohlslib.Muxer, ) error { setupVideoTrack( - desc, + origDesc, + outDesc, r, muxer, ) err := setupAudioTracks( - desc, + origDesc, + outDesc, r, muxer, ) @@ -404,7 +434,8 @@ func FromStream( } setupDataTracks( - desc, + origDesc, + outDesc, r, muxer, ) @@ -416,7 +447,7 @@ func FromStream( setuppedFormats := r.Formats() n := 1 - for _, media := range desc.Medias { + for _, media := range origDesc.Medias { for _, forma := range media.Formats { if !slices.Contains(setuppedFormats, forma) { r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, formatlabel.FormatToLabel(forma)) diff --git a/internal/protocols/hls/from_stream_test.go b/internal/protocols/hls/from_stream_test.go index 6c7a2fe4..7e4ae2c2 100644 --- a/internal/protocols/hls/from_stream_test.go +++ b/internal/protocols/hls/from_stream_test.go @@ -27,7 +27,7 @@ func TestFromStreamNoSupportedCodecs(t *testing.T) { m := &gohlslib.Muxer{} - err := FromStream(desc, r, m) + err := FromStream(desc, desc, r, m) require.Equal(t, ErrNoSupportedCodecs, err) } @@ -64,7 +64,7 @@ func TestFromStreamSkipUnsupportedTracks(t *testing.T) { }), } - err := FromStream(desc, r, m) + err := FromStream(desc, desc, r, m) require.NoError(t, err) require.Equal(t, 2, n) @@ -85,7 +85,7 @@ func TestFromStreamKLVRequiresMPEGTSVariant(t *testing.T) { m := &gohlslib.Muxer{Variant: gohlslib.MuxerVariantFMP4} - err := FromStream(desc, r, m) + err := FromStream(desc, desc, r, m) require.Equal(t, ErrNoSupportedCodecs, err) }) @@ -113,7 +113,7 @@ func TestFromStreamKLVRequiresMPEGTSVariant(t *testing.T) { m := &gohlslib.Muxer{Variant: gohlslib.MuxerVariantFMP4} - err := FromStream(desc, r, m) + err := FromStream(desc, desc, r, m) require.NoError(t, err) require.Equal(t, 1, n) require.Equal(t, 1, len(m.Tracks)) diff --git a/internal/protocols/hls/to_stream_test.go b/internal/protocols/hls/to_stream_test.go index 9ead53e3..fa064548 100644 --- a/internal/protocols/hls/to_stream_test.go +++ b/internal/protocols/hls/to_stream_test.go @@ -144,7 +144,7 @@ func TestToStream(t *testing.T) { } strm = &stream.Stream{ - Desc: &description.Session{Medias: medias}, + OrigDesc: &description.Session{Medias: medias}, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, diff --git a/internal/protocols/rtmp/from_stream.go b/internal/protocols/rtmp/from_stream.go index 14597a3d..98e62b63 100644 --- a/internal/protocols/rtmp/from_stream.go +++ b/internal/protocols/rtmp/from_stream.go @@ -43,7 +43,8 @@ func timestampToDuration(t int64, clockRate int) time.Duration { // FromStream maps a MediaMTX stream to a RTMP stream. func FromStream( - desc *description.Session, + origDesc *description.Session, + outDesc *description.Session, r *stream.Reader, conn *gortmplib.ServerConn, nconn net.Conn, @@ -56,9 +57,9 @@ func FromStream( legacyVideoTrackCount := 0 legacyAudioTrackCount := 0 - for _, media := range desc.Medias { - for _, forma := range media.Formats { - switch forma := forma.(type) { + for i, origMedia := range origDesc.Medias { + for j, origFormat := range origMedia.Formats { + switch origFormat := origFormat.(type) { case *format.AV1: if slices.Contains(conn.FourCcList, any(fourCCToString(message.FourCCAV1))) { track := &gortmplib.Track{ @@ -67,8 +68,8 @@ func FromStream( tracks = append(tracks, track) r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -77,7 +78,7 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) return (*w).WriteAV1( track, - timestampToDuration(u.PTS, forma.ClockRate()), + timestampToDuration(u.PTS, origFormat.ClockRate()), u.Payload.(unit.PayloadAV1)) }) } @@ -90,8 +91,8 @@ func FromStream( tracks = append(tracks, track) r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -100,19 +101,20 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) return (*w).WriteVP9( track, - timestampToDuration(u.PTS, forma.ClockRate()), + timestampToDuration(u.PTS, origFormat.ClockRate()), u.Payload.(unit.PayloadVP9)) }) } case *format.H265: if slices.Contains(conn.FourCcList, any(fourCCToString(message.FourCCHEVC))) { - vps, sps, pps := forma.SafeParams() + outFormat := outDesc.Medias[i].Formats[j].(*format.H265) + track := &gortmplib.Track{ Codec: &codecs.H265{ - VPS: vps, - SPS: sps, - PPS: pps, + VPS: outFormat.VPS, + SPS: outFormat.SPS, + PPS: outFormat.PPS, }, } tracks = append(tracks, track) @@ -120,8 +122,8 @@ func FromStream( var videoDTSExtractor *h265.DTSExtractor r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -143,8 +145,8 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) return (*w).WriteH265( track, - timestampToDuration(u.PTS, forma.ClockRate()), - timestampToDuration(dts, forma.ClockRate()), + timestampToDuration(u.PTS, origFormat.ClockRate()), + timestampToDuration(dts, origFormat.ClockRate()), u.Payload.(unit.PayloadH265)) }) } @@ -152,11 +154,13 @@ func FromStream( case *format.H264: if isEnhanced || legacyVideoTrackCount == 0 { legacyVideoTrackCount++ - sps, pps := forma.SafeParams() + + outFormat := outDesc.Medias[i].Formats[j].(*format.H264) + track := &gortmplib.Track{ Codec: &codecs.H264{ - SPS: sps, - PPS: pps, + SPS: outFormat.SPS, + PPS: outFormat.PPS, }, } tracks = append(tracks, track) @@ -164,8 +168,8 @@ func FromStream( var videoDTSExtractor *h264.DTSExtractor r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -205,8 +209,8 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) return (*w).WriteH264( track, - timestampToDuration(u.PTS, forma.ClockRate()), - timestampToDuration(dts, forma.ClockRate()), + timestampToDuration(u.PTS, origFormat.ClockRate()), + timestampToDuration(dts, origFormat.ClockRate()), u.Payload.(unit.PayloadH264)) }) } @@ -217,7 +221,7 @@ func FromStream( Codec: &codecs.Opus{ IDHeader: &opus.IDHeader{ Version: 0x1, - ChannelCount: uint8(forma.ChannelCount), + ChannelCount: uint8(origFormat.ChannelCount), PreSkip: 3840, }, }, @@ -225,8 +229,8 @@ func FromStream( tracks = append(tracks, track) r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -238,7 +242,7 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) err := (*w).WriteOpus( track, - timestampToDuration(pts, forma.ClockRate()), + timestampToDuration(pts, origFormat.ClockRate()), pkt, ) if err != nil { @@ -257,14 +261,14 @@ func FromStream( legacyAudioTrackCount++ track := &gortmplib.Track{ Codec: &codecs.MPEG4Audio{ - Config: forma.Config, + Config: origFormat.Config, }, } tracks = append(tracks, track) r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -276,7 +280,7 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) err := (*w).WriteMPEG4Audio( track, - timestampToDuration(pts, forma.ClockRate()), + timestampToDuration(pts, origFormat.ClockRate()), au, ) if err != nil { @@ -289,25 +293,25 @@ func FromStream( } case *format.MPEG4AudioLATM: - if !forma.CPresent && (isEnhanced || legacyAudioTrackCount == 0) { + if !origFormat.CPresent && (isEnhanced || legacyAudioTrackCount == 0) { legacyAudioTrackCount++ track := &gortmplib.Track{ Codec: &codecs.MPEG4Audio{ - Config: forma.StreamMuxConfig.Programs[0].Layers[0].AudioSpecificConfig, + Config: origFormat.StreamMuxConfig.Programs[0].Layers[0].AudioSpecificConfig, }, } tracks = append(tracks, track) r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil } var ame mpeg4audio.AudioMuxElement - ame.StreamMuxConfig = forma.StreamMuxConfig + ame.StreamMuxConfig = origFormat.StreamMuxConfig err := ame.Unmarshal(u.Payload.(unit.PayloadMPEG4AudioLATM)) if err != nil { return err @@ -316,7 +320,7 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) return (*w).WriteMPEG4Audio( track, - timestampToDuration(u.PTS, forma.ClockRate()), + timestampToDuration(u.PTS, origFormat.ClockRate()), ame.Payloads[0][0][0], ) }) @@ -331,8 +335,8 @@ func FromStream( tracks = append(tracks, track) r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -350,14 +354,14 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) err = (*w).WriteMPEG1Audio( track, - timestampToDuration(pts, forma.ClockRate()), + timestampToDuration(pts, origFormat.ClockRate()), frame) if err != nil { return err } pts += int64(h.SampleCount()) * - int64(forma.ClockRate()) / int64(h.SampleRate) + int64(origFormat.ClockRate()) / int64(h.SampleRate) } return nil @@ -368,15 +372,15 @@ func FromStream( if slices.Contains(conn.FourCcList, any(fourCCToString(message.FourCCAC3))) { track := &gortmplib.Track{ Codec: &codecs.AC3{ - SampleRate: forma.SampleRate, - ChannelCount: forma.ChannelCount, + SampleRate: origFormat.SampleRate, + ChannelCount: origFormat.ChannelCount, }, } tracks = append(tracks, track) r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -388,7 +392,7 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) err := (*w).WriteAC3( track, - timestampToDuration(pts, forma.ClockRate()), + timestampToDuration(pts, origFormat.ClockRate()), frame) if err != nil { return err @@ -400,19 +404,19 @@ func FromStream( } case *format.G711: - if forma.SampleRate == 8000 && (isEnhanced || legacyAudioTrackCount == 0) { + if origFormat.SampleRate == 8000 && (isEnhanced || legacyAudioTrackCount == 0) { legacyAudioTrackCount++ track := &gortmplib.Track{ Codec: &codecs.G711{ - MULaw: forma.MULaw, - ChannelCount: forma.ChannelCount, + MULaw: origFormat.MULaw, + ChannelCount: origFormat.ChannelCount, }, } tracks = append(tracks, track) r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -421,32 +425,32 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) return (*w).WriteG711( track, - timestampToDuration(u.PTS, forma.ClockRate()), + timestampToDuration(u.PTS, origFormat.ClockRate()), u.Payload.(unit.PayloadG711), ) }) } case *format.LPCM: - if (forma.ChannelCount == 1 || forma.ChannelCount == 2) && - (forma.SampleRate == 5512 || - forma.SampleRate == 11025 || - forma.SampleRate == 22050 || - forma.SampleRate == 44100) && + if (origFormat.ChannelCount == 1 || origFormat.ChannelCount == 2) && + (origFormat.SampleRate == 5512 || + origFormat.SampleRate == 11025 || + origFormat.SampleRate == 22050 || + origFormat.SampleRate == 44100) && (isEnhanced || legacyAudioTrackCount == 0) { legacyAudioTrackCount++ track := &gortmplib.Track{ Codec: &codecs.LPCM{ - BitDepth: forma.BitDepth, - SampleRate: forma.SampleRate, - ChannelCount: forma.ChannelCount, + BitDepth: origFormat.BitDepth, + SampleRate: origFormat.SampleRate, + ChannelCount: origFormat.ChannelCount, }, } tracks = append(tracks, track) r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -455,16 +459,16 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) return (*w).WriteLPCM( track, - timestampToDuration(u.PTS, forma.ClockRate()), + timestampToDuration(u.PTS, origFormat.ClockRate()), u.Payload.(unit.PayloadLPCM), ) }) } case *format.Generic: - if strings.HasPrefix(strings.ToLower(forma.RTPMap()), "flac/") && + if strings.HasPrefix(strings.ToLower(origFormat.RTPMap()), "flac/") && slices.Contains(conn.FourCcList, any(fourCCToString(message.FourCCFLAC))) { - enc, err := hex.DecodeString(forma.FMT["streaminfo"]) + enc, err := hex.DecodeString(origFormat.FMT["streaminfo"]) if err != nil { return err } @@ -483,8 +487,8 @@ func FromStream( tracks = append(tracks, track) r.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -493,7 +497,7 @@ func FromStream( nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) return (*w).WriteFLAC( track, - timestampToDuration(u.PTS, forma.ClockRate()), + timestampToDuration(u.PTS, origFormat.ClockRate()), u.Payload.(unit.PayloadFLAC), ) }) @@ -518,10 +522,10 @@ func FromStream( setuppedFormats := r.Formats() n := 1 - for _, media := range desc.Medias { - for _, forma := range media.Formats { - if !slices.Contains(setuppedFormats, forma) { - r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, formatlabel.FormatToLabel(forma)) + for _, origMedia := range origDesc.Medias { + for _, origFormat := range origMedia.Formats { + if !slices.Contains(setuppedFormats, origFormat) { + r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, formatlabel.FormatToLabel(origFormat)) } n++ } diff --git a/internal/protocols/rtmp/from_stream_test.go b/internal/protocols/rtmp/from_stream_test.go index 9c9a3658..574162f0 100644 --- a/internal/protocols/rtmp/from_stream_test.go +++ b/internal/protocols/rtmp/from_stream_test.go @@ -616,7 +616,7 @@ func TestFromStream(t *testing.T) { medias := tc.medias strm := &stream.Stream{ - Desc: &description.Session{Medias: medias}, + OrigDesc: &description.Session{Medias: medias}, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -673,7 +673,7 @@ func TestFromStream(t *testing.T) { r := &stream.Reader{Parent: test.NilLogger} - err = FromStream(strm.Desc, r, conn, nconn, 10*time.Second) + err = FromStream(strm.OrigDesc, strm.OutDescCopy(), r, conn, nconn, 10*time.Second) require.NoError(t, err) strm.AddReader(r) @@ -751,7 +751,7 @@ func TestFromStreamLegacyClientMultipleTracks(t *testing.T) { } strm := &stream.Stream{ - Desc: &description.Session{Medias: medias}, + OrigDesc: &description.Session{Medias: medias}, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -819,7 +819,7 @@ func TestFromStreamLegacyClientMultipleTracks(t *testing.T) { r := &stream.Reader{Parent: test.NilLogger} - err = FromStream(strm.Desc, r, conn, nconn, 10*time.Second) + err = FromStream(strm.OrigDesc, strm.OutDescCopy(), r, conn, nconn, 10*time.Second) require.NoError(t, err) strm.AddReader(r) @@ -857,7 +857,7 @@ func TestFromStreamNoSupportedCodecs(t *testing.T) { conn := &gortmplib.ServerConn{} - err := FromStream(desc, r, conn, nil, 0) + err := FromStream(desc, desc, r, conn, nil, 0) require.Equal(t, errNoSupportedCodecsFrom, err) } @@ -913,7 +913,7 @@ func TestFromStreamSkipUnsupportedTracks(t *testing.T) { err = conn.Accept() require.NoError(t, err) - err = FromStream(desc, r, conn, nil, 0) + err = FromStream(desc, desc, r, conn, nil, 0) require.NoError(t, err) require.Equal(t, 1, n) diff --git a/internal/protocols/webrtc/from_stream_test.go b/internal/protocols/webrtc/from_stream_test.go index 7529cdff..9f509fc2 100644 --- a/internal/protocols/webrtc/from_stream_test.go +++ b/internal/protocols/webrtc/from_stream_test.go @@ -87,7 +87,7 @@ func TestFromStream(t *testing.T) { func TestFromStreamResampleOpus(t *testing.T) { strm := &stream.Stream{ - Desc: &description.Session{Medias: []*description.Media{ + OrigDesc: &description.Session{Medias: []*description.Media{ { Type: description.MediaTypeAudio, Formats: []format.Format{&format.Opus{ @@ -129,7 +129,7 @@ func TestFromStreamResampleOpus(t *testing.T) { r := &stream.Reader{Parent: nil} - err = FromStream(strm.Desc, r, pc2) + err = FromStream(strm.OrigDesc, r, pc2) require.NoError(t, err) err = pc2.Start() @@ -154,7 +154,7 @@ func TestFromStreamResampleOpus(t *testing.T) { strm.AddReader(r) defer strm.RemoveReader(r) - subStream.WriteUnit(strm.Desc.Medias[0], strm.Desc.Medias[0].Formats[0], &unit.Unit{ + subStream.WriteUnit(strm.OrigDesc.Medias[0], strm.OrigDesc.Medias[0].Formats[0], &unit.Unit{ PTS: 0, NTP: time.Now(), RTPPackets: []*rtp.Packet{{ @@ -170,7 +170,7 @@ func TestFromStreamResampleOpus(t *testing.T) { }}, }) - subStream.WriteUnit(strm.Desc.Medias[0], strm.Desc.Medias[0].Formats[0], &unit.Unit{ + subStream.WriteUnit(strm.OrigDesc.Medias[0], strm.OrigDesc.Medias[0].Formats[0], &unit.Unit{ PTS: 0, NTP: time.Now(), RTPPackets: []*rtp.Packet{{ @@ -215,7 +215,7 @@ func TestFromStreamResampleOpus(t *testing.T) { func TestFromStreamResampleOpusAbsoluteTimestamp(t *testing.T) { strm := &stream.Stream{ - Desc: &description.Session{Medias: []*description.Media{ + OrigDesc: &description.Session{Medias: []*description.Media{ { Type: description.MediaTypeAudio, Formats: []format.Format{&format.Opus{ @@ -257,7 +257,7 @@ func TestFromStreamResampleOpusAbsoluteTimestamp(t *testing.T) { r := &stream.Reader{Parent: nil} - err = FromStream(strm.Desc, r, pcPublisher) + err = FromStream(strm.OrigDesc, r, pcPublisher) require.NoError(t, err) err = pcPublisher.Start() @@ -286,7 +286,7 @@ func TestFromStreamResampleOpusAbsoluteTimestamp(t *testing.T) { step := 20 * time.Millisecond // prime the pipeline to allow track gathering - subStream.WriteUnit(strm.Desc.Medias[0], strm.Desc.Medias[0].Formats[0], &unit.Unit{ + subStream.WriteUnit(strm.OrigDesc.Medias[0], strm.OrigDesc.Medias[0].Formats[0], &unit.Unit{ PTS: 0, NTP: baseNTP, RTPPackets: []*rtp.Packet{{ @@ -355,7 +355,7 @@ func TestFromStreamResampleOpusAbsoluteTimestamp(t *testing.T) { seq := startSeq + i expected, _ := expectedNTP(seq) - subStream.WriteUnit(strm.Desc.Medias[0], strm.Desc.Medias[0].Formats[0], &unit.Unit{ + subStream.WriteUnit(strm.OrigDesc.Medias[0], strm.OrigDesc.Medias[0].Formats[0], &unit.Unit{ PTS: 0, NTP: expected, RTPPackets: []*rtp.Packet{{ diff --git a/internal/recorder/format_fmp4.go b/internal/recorder/format_fmp4.go index f573e859..4f1b1fc3 100644 --- a/internal/recorder/format_fmp4.go +++ b/internal/recorder/format_fmp4.go @@ -176,22 +176,24 @@ func (f *formatFMP4) initialize() bool { return track } - for _, media := range f.ri.stream.Desc.Medias { - for _, forma := range media.Formats { - clockRate := forma.ClockRate() + outDesc := f.ri.stream.OutDescCopy() - switch forma := forma.(type) { + for i, origMedia := range f.ri.stream.OrigDesc.Medias { + for j, origFormat := range origMedia.Formats { + clockRate := origFormat.ClockRate() + + switch origFormat := origFormat.(type) { case *rtspformat.AV1: codec := &mcodecs.AV1{ SequenceHeader: av1DefaultSequenceHeader, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) firstReceived := false f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -245,13 +247,13 @@ func (f *formatFMP4) initialize() bool { ChromaSubsampling: 1, ColorRange: false, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) firstReceived := false f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -320,7 +322,12 @@ func (f *formatFMP4) initialize() bool { // TODO case *rtspformat.H265: - vps, sps, pps := forma.SafeParams() + outFormat := outDesc.Medias[i].Formats[j].(*rtspformat.H265) + + vps := outFormat.VPS + sps := outFormat.SPS + pps := outFormat.PPS + if vps == nil || sps == nil || pps == nil { vps = h265DefaultVPS sps = h265DefaultSPS @@ -332,13 +339,13 @@ func (f *formatFMP4) initialize() bool { SPS: sps, PPS: pps, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) var dtsExtractor *h265.DTSExtractor f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -405,7 +412,11 @@ func (f *formatFMP4) initialize() bool { }) case *rtspformat.H264: - sps, pps := forma.SafeParams() + outFormat := outDesc.Medias[i].Formats[j].(*rtspformat.H264) + + sps := outFormat.SPS + pps := outFormat.PPS + if sps == nil || pps == nil { sps = h264DefaultSPS pps = h264DefaultPPS @@ -415,13 +426,13 @@ func (f *formatFMP4) initialize() bool { SPS: sps, PPS: pps, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) var dtsExtractor *h264.DTSExtractor f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -481,7 +492,9 @@ func (f *formatFMP4) initialize() bool { }) case *rtspformat.MPEG4Video: - config := forma.SafeParams() + outFormat := outDesc.Medias[i].Formats[j].(*rtspformat.MPEG4Video) + + config := outFormat.Config if config == nil { config = mpeg4VideoDefaultConfig @@ -490,14 +503,14 @@ func (f *formatFMP4) initialize() bool { codec := &mcodecs.MPEG4Video{ Config: config, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) firstReceived := false var lastPTS int64 f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -544,14 +557,14 @@ func (f *formatFMP4) initialize() bool { codec := &mcodecs.MPEG1Video{ Config: mpeg1VideoDefaultConfig, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) firstReceived := false var lastPTS int64 f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -596,13 +609,13 @@ func (f *formatFMP4) initialize() bool { Width: 800, Height: 600, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) parsed := false f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -630,13 +643,13 @@ func (f *formatFMP4) initialize() bool { case *rtspformat.Opus: codec := &mcodecs.Opus{ - ChannelCount: forma.ChannelCount, + ChannelCount: origFormat.ChannelCount, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -664,13 +677,13 @@ func (f *formatFMP4) initialize() bool { case *rtspformat.MPEG4Audio: codec := &mcodecs.MPEG4Audio{ - Config: *forma.Config, + Config: *origFormat.Config, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -695,22 +708,22 @@ func (f *formatFMP4) initialize() bool { }) case *rtspformat.MPEG4AudioLATM: - if !forma.CPresent { + if !origFormat.CPresent { codec := &mcodecs.MPEG4Audio{ - Config: *forma.StreamMuxConfig.Programs[0].Layers[0].AudioSpecificConfig, + Config: *origFormat.StreamMuxConfig.Programs[0].Layers[0].AudioSpecificConfig, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil } var ame mpeg4audio.AudioMuxElement - ame.StreamMuxConfig = forma.StreamMuxConfig + ame.StreamMuxConfig = origFormat.StreamMuxConfig err := ame.Unmarshal(u.Payload.(unit.PayloadMPEG4AudioLATM)) if err != nil { return err @@ -731,13 +744,13 @@ func (f *formatFMP4) initialize() bool { SampleRate: 32000, ChannelCount: 2, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) parsed := false f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -779,8 +792,8 @@ func (f *formatFMP4) initialize() bool { case *rtspformat.AC3: codec := &mcodecs.AC3{ - SampleRate: forma.SampleRate, - ChannelCount: forma.ChannelCount, + SampleRate: origFormat.SampleRate, + ChannelCount: origFormat.ChannelCount, Fscod: 0, Bsid: 8, Bsmod: 0, @@ -788,13 +801,13 @@ func (f *formatFMP4) initialize() bool { LfeOn: true, BitRateCode: 7, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) parsed := false f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -849,21 +862,21 @@ func (f *formatFMP4) initialize() bool { codec := &mcodecs.LPCM{ LittleEndian: false, BitDepth: 16, - SampleRate: forma.SampleRate, - ChannelCount: forma.ChannelCount, + SampleRate: origFormat.SampleRate, + ChannelCount: origFormat.ChannelCount, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil } var lpcm []byte - if forma.MULaw { + if origFormat.MULaw { var mu g711.Mulaw mu.Unmarshal(u.Payload.(unit.PayloadG711)) lpcm = mu @@ -885,15 +898,15 @@ func (f *formatFMP4) initialize() bool { case *rtspformat.LPCM: codec := &mcodecs.LPCM{ LittleEndian: false, - BitDepth: forma.BitDepth, - SampleRate: forma.SampleRate, - ChannelCount: forma.ChannelCount, + BitDepth: origFormat.BitDepth, + SampleRate: origFormat.SampleRate, + ChannelCount: origFormat.ChannelCount, } - track := addTrack(forma, codec) + track := addTrack(origFormat, codec) f.ri.reader.OnData( - media, - forma, + origMedia, + origFormat, func(u *unit.Unit) error { if u.NilPayload() { return nil @@ -919,7 +932,7 @@ func (f *formatFMP4) initialize() bool { setuppedFormats := f.ri.reader.Formats() n := 1 - for _, medi := range f.ri.stream.Desc.Medias { + for _, medi := range f.ri.stream.OrigDesc.Medias { for _, forma := range medi.Formats { if !slices.Contains(setuppedFormats, forma) { f.ri.Log(logger.Warn, "skipping track %d (%s)", n, formatlabel.FormatToLabel(forma)) diff --git a/internal/recorder/format_mpegts.go b/internal/recorder/format_mpegts.go index 7eaafb92..675ec5ae 100644 --- a/internal/recorder/format_mpegts.go +++ b/internal/recorder/format_mpegts.go @@ -79,7 +79,7 @@ func (f *formatMPEGTS) initialize() bool { return track } - for _, media := range f.ri.stream.Desc.Medias { + for _, media := range f.ri.stream.OrigDesc.Medias { for _, forma := range media.Formats { clockRate := forma.ClockRate() @@ -422,7 +422,7 @@ func (f *formatMPEGTS) initialize() bool { setuppedFormats := f.ri.reader.Formats() n := 1 - for _, medi := range f.ri.stream.Desc.Medias { + for _, medi := range f.ri.stream.OrigDesc.Medias { for _, forma := range medi.Formats { if !slices.Contains(setuppedFormats, forma) { f.ri.Log(logger.Warn, "skipping track %d (%s)", n, formatlabel.FormatToLabel(forma)) diff --git a/internal/recorder/recorder_test.go b/internal/recorder/recorder_test.go index efa6cd04..3ab81192 100644 --- a/internal/recorder/recorder_test.go +++ b/internal/recorder/recorder_test.go @@ -130,7 +130,7 @@ func TestRecorder(t *testing.T) { for _, ca := range []string{"fmp4", "mpegts"} { t.Run(ca, func(t *testing.T) { strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -423,7 +423,7 @@ func TestRecorderFMP4NegativeInitialDTS(t *testing.T) { }} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -516,7 +516,7 @@ func TestRecorderFMP4NegativeDTSDiff(t *testing.T) { }} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -614,7 +614,7 @@ func TestRecorderSkipTracksPartial(t *testing.T) { }} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -680,7 +680,7 @@ func TestRecorderSkipTracksFull(t *testing.T) { }} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -748,7 +748,7 @@ func TestRecorderFMP4SegmentSwitch(t *testing.T) { }} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -875,7 +875,7 @@ func TestRecorderTimeDriftDetector(t *testing.T) { }} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, diff --git a/internal/servers/hls/muxer_instance.go b/internal/servers/hls/muxer_instance.go index 97986c6c..9823a204 100644 --- a/internal/servers/hls/muxer_instance.go +++ b/internal/servers/hls/muxer_instance.go @@ -76,7 +76,11 @@ func (mi *muxerInstance) initialize() error { Parent: mi, } - err := hls.FromStream(mi.stream.Desc, mi.reader, mi.hmuxer) + err := hls.FromStream( + mi.stream.OrigDesc, + mi.stream.OutDescCopy(), + mi.reader, + mi.hmuxer) if err != nil { return err } diff --git a/internal/servers/hls/server_test.go b/internal/servers/hls/server_test.go index fbcc9515..b8989f7e 100644 --- a/internal/servers/hls/server_test.go +++ b/internal/servers/hls/server_test.go @@ -319,7 +319,7 @@ func TestServerRead(t *testing.T) { }} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, ReplaceNTP: false, @@ -489,7 +489,7 @@ func TestServerDirectory(t *testing.T) { desc := &description.Session{Medias: []*description.Media{test.MediaH264}} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -544,7 +544,7 @@ func TestServerDynamicAlwaysRemux(t *testing.T) { desc := &description.Session{Medias: []*description.Media{test.MediaH264}} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -705,7 +705,7 @@ func TestServerNoSupportedCodecs(t *testing.T) { }}} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, diff --git a/internal/servers/hls/session.go b/internal/servers/hls/session.go index f413264c..e1fadd79 100644 --- a/internal/servers/hls/session.go +++ b/internal/servers/hls/session.go @@ -108,8 +108,9 @@ func (s *session) initialize(ctx *gin.Context) error { Parent: s, } - // all of this is needed to allow Stream to increase outbound bytes for every HLS session - for _, medi := range res.Stream.Desc.Medias { + // this is needed to increase stream outbound bytes for every HLS session, + // even if HLS sessions are not directly attached to streams (they are through muxers). + for _, medi := range res.Stream.OrigDesc.Medias { for _, forma := range medi.Formats { if slices.Contains(muxerFormats, forma) { s.reader.OnData(medi, forma, func(_ *unit.Unit) error { diff --git a/internal/servers/moq/server_test.go b/internal/servers/moq/server_test.go index a58383b5..d3ee7498 100644 --- a/internal/servers/moq/server_test.go +++ b/internal/servers/moq/server_test.go @@ -34,7 +34,7 @@ func (p *serverDummyPath) RemoveReader(_ defs.PathRemoveReaderReq) {} func TestServer(t *testing.T) { desc := &description.Session{Medias: []*description.Media{test.UniqueMediaH264()}} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -172,8 +172,7 @@ func TestServer(t *testing.T) { err = frameSG.Read(frameStream) require.NoError(t, err) - sps, pps := test.FormatH264.SafeParams() - expectedPayload, err2 := mch264.AVCC([][]byte{sps, pps, {5, 1}}).Marshal() + 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) } diff --git a/internal/servers/moq/session.go b/internal/servers/moq/session.go index 51df53d5..7427389d 100644 --- a/internal/servers/moq/session.go +++ b/internal/servers/moq/session.go @@ -379,7 +379,7 @@ func (s *session) onSubscribeCatalog(wstream *webtransport.Stream, m *controlmes } tracks, err := moq.FromStream( - addRes.Stream.Desc, + addRes.Stream.OrigDesc, ) if err != nil { addRes.Path.RemoveReader(defs.PathRemoveReaderReq{Author: s}) diff --git a/internal/servers/rtmp/conn.go b/internal/servers/rtmp/conn.go index 03466a09..c90451e0 100644 --- a/internal/servers/rtmp/conn.go +++ b/internal/servers/rtmp/conn.go @@ -185,7 +185,13 @@ func (c *conn) runRead() error { r := &stream.Reader{Parent: c} - err = rtmp.FromStream(res.Stream.Desc, r, c.rconn, c.nconn, time.Duration(c.writeTimeout)) + err = rtmp.FromStream( + res.Stream.OrigDesc, + res.Stream.OutDescCopy(), + r, + c.rconn, + c.nconn, + time.Duration(c.writeTimeout)) if err != nil { return err } diff --git a/internal/servers/rtmp/server_test.go b/internal/servers/rtmp/server_test.go index 2bef6122..8e2060b3 100644 --- a/internal/servers/rtmp/server_test.go +++ b/internal/servers/rtmp/server_test.go @@ -80,7 +80,7 @@ func TestServerPublish(t *testing.T) { require.Equal(t, "mypass", req.AccessRequest.Credentials.Pass) strm = &stream.Stream{ - Desc: req.Desc, + OrigDesc: req.Desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -98,8 +98,8 @@ func TestServerPublish(t *testing.T) { reader = &stream.Reader{Parent: test.NilLogger} reader.OnData( - strm.Desc.Medias[0], - strm.Desc.Medias[0].Formats[0], + strm.OrigDesc.Medias[0], + strm.OrigDesc.Medias[0].Formats[0], func(u *unit.Unit) error { switch n { case 0: @@ -261,7 +261,7 @@ func TestServerRead(t *testing.T) { desc := &description.Session{Medias: []*description.Media{test.MediaH264}} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, diff --git a/internal/servers/rtsp/conn.go b/internal/servers/rtsp/conn.go index b4d047b3..a6224630 100644 --- a/internal/servers/rtsp/conn.go +++ b/internal/servers/rtsp/conn.go @@ -190,9 +190,15 @@ func (c *conn) onDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx, var strm *gortsplib.ServerStream if !c.encryption { - strm = res.Stream.RTSPStream(c.rserver) + strm, err = res.Stream.RTSPStream(c.rserver) } else { - strm = res.Stream.RTSPSStream(c.rserver) + strm, err = res.Stream.RTSPSStream(c.rserver) + } + + if err != nil { + return &base.Response{ + StatusCode: base.StatusBadRequest, + }, nil, err } return &base.Response{ diff --git a/internal/servers/rtsp/server_test.go b/internal/servers/rtsp/server_test.go index b174a942..e4f3d107 100644 --- a/internal/servers/rtsp/server_test.go +++ b/internal/servers/rtsp/server_test.go @@ -110,7 +110,7 @@ func TestServerPublish(t *testing.T) { require.True(t, req.AccessRequest.SkipAuth) strm = &stream.Stream{ - Desc: req.Desc, + OrigDesc: req.Desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -128,8 +128,8 @@ func TestServerPublish(t *testing.T) { reader = &stream.Reader{Parent: test.NilLogger} reader.OnData( - strm.Desc.Medias[0], - strm.Desc.Medias[0].Formats[0], + strm.OrigDesc.Medias[0], + strm.OrigDesc.Medias[0].Formats[0], func(u *unit.Unit) error { require.Equal(t, unit.PayloadH264{ test.FormatH264.SPS, @@ -302,7 +302,7 @@ func TestServerPublishMPEGTS(t *testing.T) { }}}, req.Desc) strm = &stream.Stream{ - Desc: req.Desc, + OrigDesc: req.Desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -321,8 +321,8 @@ func TestServerPublishMPEGTS(t *testing.T) { n := 0 reader.OnData( - strm.Desc.Medias[0], - strm.Desc.Medias[0].Formats[0], + strm.OrigDesc.Medias[0], + strm.OrigDesc.Medias[0].Formats[0], func(u *unit.Unit) error { if n == 0 { require.Equal(t, unit.PayloadH264{ @@ -420,7 +420,7 @@ func TestServerRead(t *testing.T) { desc := &description.Session{Medias: []*description.Media{test.MediaH264}} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -597,7 +597,7 @@ func TestServerRedirect(t *testing.T) { desc := &description.Session{Medias: []*description.Media{test.MediaH264}} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, diff --git a/internal/servers/rtsp/session.go b/internal/servers/rtsp/session.go index 0a513cf7..b4c3423a 100644 --- a/internal/servers/rtsp/session.go +++ b/internal/servers/rtsp/session.go @@ -72,6 +72,7 @@ type session struct { pathConf *conf.Path // record only path defs.Path stream *stream.Stream + rtspStream *gortsplib.ServerStream subStream *stream.SubStream onUnreadHook func() inboundRTPPacketsLost *counterdumper.Dumper @@ -235,13 +236,6 @@ func (s *session) onAnnounce(c *conn, ctx *gortsplib.ServerHandlerOnAnnounceCtx) }, nil } -func (s *session) rtspStream() *gortsplib.ServerStream { - if !s.encryption { - return s.stream.RTSPStream(s.rserver) - } - return s.stream.RTSPSStream(s.rserver) -} - // onSetup is called by rtspServer. func (s *session) onSetup(c *conn, ctx *gortsplib.ServerHandlerOnSetupCtx, ) (*base.Response, *gortsplib.ServerStream, error) { @@ -310,8 +304,23 @@ func (s *session) onSetup(c *conn, ctx *gortsplib.ServerHandlerOnSetupCtx, }, nil, err } - s.path = res.Path + var rtspStream *gortsplib.ServerStream + if !s.encryption { + rtspStream, err = res.Stream.RTSPStream(s.rserver) + } else { + rtspStream, err = res.Stream.RTSPSStream(s.rserver) + } + + if err != nil { + res.Path.RemoveReader(defs.PathRemoveReaderReq{Author: s}) + return &base.Response{ + StatusCode: base.StatusBadRequest, + }, nil, err + } + s.stream = res.Stream + s.rtspStream = rtspStream + s.path = res.Path s.mutex.Lock() s.user = res.User @@ -320,12 +329,12 @@ func (s *session) onSetup(c *conn, ctx *gortsplib.ServerHandlerOnSetupCtx, return &base.Response{ StatusCode: base.StatusOK, - }, s.rtspStream(), nil + }, s.rtspStream, nil case gortsplib.ServerSessionStatePrePlay: // play, subsequent calls return &base.Response{ StatusCode: base.StatusOK, - }, s.rtspStream(), nil + }, s.rtspStream, nil default: // record return &base.Response{ diff --git a/internal/servers/srt/conn.go b/internal/servers/srt/conn.go index 1e197d2e..ce7291da 100644 --- a/internal/servers/srt/conn.go +++ b/internal/servers/srt/conn.go @@ -298,7 +298,7 @@ func (c *conn) runRead(streamID *streamID) error { r := &stream.Reader{Parent: c} - err = mpegts.FromStream(res.Stream.Desc, r, bw, sconn, time.Duration(c.writeTimeout)) + err = mpegts.FromStream(res.Stream.OrigDesc, r, bw, sconn, time.Duration(c.writeTimeout)) if err != nil { return err } diff --git a/internal/servers/srt/server_test.go b/internal/servers/srt/server_test.go index ef3d5d7a..fa42602f 100644 --- a/internal/servers/srt/server_test.go +++ b/internal/servers/srt/server_test.go @@ -66,7 +66,7 @@ func TestServerPublish(t *testing.T) { require.True(t, req.AccessRequest.SkipAuth) strm = &stream.Stream{ - Desc: req.Desc, + OrigDesc: req.Desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -84,8 +84,8 @@ func TestServerPublish(t *testing.T) { reader = &stream.Reader{Parent: test.NilLogger} reader.OnData( - strm.Desc.Medias[0], - strm.Desc.Medias[0].Formats[0], + strm.OrigDesc.Medias[0], + strm.OrigDesc.Medias[0].Formats[0], func(u *unit.Unit) error { switch n { case 0: @@ -257,7 +257,7 @@ func TestServerRead(t *testing.T) { desc := &description.Session{Medias: []*description.Media{test.MediaH264}} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, diff --git a/internal/servers/webrtc/server_test.go b/internal/servers/webrtc/server_test.go index 5d83c23f..5d8d0dc3 100644 --- a/internal/servers/webrtc/server_test.go +++ b/internal/servers/webrtc/server_test.go @@ -356,7 +356,7 @@ func TestServerPublish(t *testing.T) { require.True(t, req.AccessRequest.SkipAuth) strm = &stream.Stream{ - Desc: req.Desc, + OrigDesc: req.Desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -374,8 +374,8 @@ func TestServerPublish(t *testing.T) { reader = &stream.Reader{Parent: test.NilLogger} reader.OnData( - strm.Desc.Medias[0], - strm.Desc.Medias[0].Formats[0], + strm.OrigDesc.Medias[0], + strm.OrigDesc.Medias[0].Formats[0], func(u *unit.Unit) error { /* select { case <-recv: @@ -646,7 +646,7 @@ func TestServerRead(t *testing.T) { desc := &description.Session{Medias: ca.medias} strm := &stream.Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -920,7 +920,7 @@ func TestServerICERestart(t *testing.T) { }, AddPublisherImpl: func(req defs.PathAddPublisherReq) (*defs.PathAddPublisherRes, error) { strm = &stream.Stream{ - Desc: req.Desc, + OrigDesc: req.Desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: test.NilLogger, @@ -939,8 +939,8 @@ func TestServerICERestart(t *testing.T) { n := 0 reader.OnData( - strm.Desc.Medias[0], - strm.Desc.Medias[0].Formats[0], + strm.OrigDesc.Medias[0], + strm.OrigDesc.Medias[0].Formats[0], func(u *unit.Unit) error { switch n { case 0: diff --git a/internal/servers/webrtc/session.go b/internal/servers/webrtc/session.go index 8a5f24c1..bcd285fb 100644 --- a/internal/servers/webrtc/session.go +++ b/internal/servers/webrtc/session.go @@ -526,7 +526,7 @@ func (s *session) runRead(req *initialRequestReq) (int, error) { r := &stream.Reader{Parent: s} - err = webrtc.FromStream(res.Stream.Desc, r, pc) + err = webrtc.FromStream(res.Stream.OrigDesc, r, pc) if err != nil { return http.StatusBadRequest, err } diff --git a/internal/staticsources/rpicamera/source.go b/internal/staticsources/rpicamera/source.go index 91d50505..1f73e840 100644 --- a/internal/staticsources/rpicamera/source.go +++ b/internal/staticsources/rpicamera/source.go @@ -295,8 +295,8 @@ func (s *Source) runSecondary(params defs.StaticSourceRunParams) error { rdr := &stream.Reader{Parent: s} rdr.OnData( - primaryStream.Desc.Medias[1], - primaryStream.Desc.Medias[1].Formats[0], + primaryStream.OrigDesc.Medias[1], + primaryStream.OrigDesc.Medias[1].Formats[0], func(u *unit.Unit) error { pkt := u.RTPPackets[0] diff --git a/internal/stream/format_updater.go b/internal/stream/format_updater.go index 152fc893..b974c180 100644 --- a/internal/stream/format_updater.go +++ b/internal/stream/format_updater.go @@ -10,10 +10,10 @@ import ( "github.com/bluenviron/mediamtx/internal/unit" ) -type formatUpdater func(format.Format, unit.Payload) +type formatUpdater func(outFormat format.Format, payload unit.Payload, updateOutDesc func(func())) -func formatUpdaterH265(forma format.Format, payload unit.Payload) { - formatH265 := forma.(*format.H265) +func formatUpdaterH265(outFormat format.Format, payload unit.Payload, updateOutDesc func(func())) { + formatH265 := outFormat.(*format.H265) au := payload.(unit.PayloadH265) vps, sps, pps := formatH265.VPS, formatH265.SPS, formatH265.PPS @@ -44,12 +44,16 @@ func formatUpdaterH265(forma format.Format, payload unit.Payload) { } if update { - formatH265.SafeSetParams(vps, sps, pps) + updateOutDesc(func() { + formatH265.VPS = vps + formatH265.SPS = sps + formatH265.PPS = pps + }) } } -func formatUpdaterH264(forma format.Format, payload unit.Payload) { - formatH264 := forma.(*format.H264) +func formatUpdaterH264(outFormat format.Format, payload unit.Payload, updateOutDesc func(func())) { + formatH264 := outFormat.(*format.H264) au := payload.(unit.PayloadH264) sps, pps := formatH264.SPS, formatH264.PPS @@ -74,12 +78,15 @@ func formatUpdaterH264(forma format.Format, payload unit.Payload) { } if update { - formatH264.SafeSetParams(sps, pps) + updateOutDesc(func() { + formatH264.SPS = sps + formatH264.PPS = pps + }) } } -func formatUpdaterMPEG4Video(forma format.Format, payload unit.Payload) { - formatMPEG4Video := forma.(*format.MPEG4Video) +func formatUpdaterMPEG4Video(outFormat format.Format, payload unit.Payload, updateOutDesc func(func())) { + formatMPEG4Video := outFormat.(*format.MPEG4Video) frame := payload.(unit.PayloadMPEG4Video) if bytes.HasPrefix(frame, []byte{0, 0, 1, byte(mpeg4video.VisualObjectSequenceStartCode)}) { @@ -90,13 +97,15 @@ func formatUpdaterMPEG4Video(forma format.Format, payload unit.Payload) { conf := frame[:end+4] if !bytes.Equal(conf, formatMPEG4Video.Config) { - formatMPEG4Video.SafeSetParams(conf) + updateOutDesc(func() { + formatMPEG4Video.Config = conf + }) } } } -func newFormatUpdater(forma format.Format) formatUpdater { - switch forma.(type) { +func newFormatUpdater(outFormat format.Format) formatUpdater { + switch outFormat.(type) { case *format.H265: return formatUpdaterH265 @@ -107,7 +116,7 @@ func newFormatUpdater(forma format.Format) formatUpdater { return formatUpdaterMPEG4Video default: - return formatUpdater(func(_ format.Format, _ unit.Payload) { + return formatUpdater(func(_ format.Format, _ unit.Payload, _ func(func())) { }) } } diff --git a/internal/stream/offline_sub_stream.go b/internal/stream/offline_sub_stream.go index c1fb4700..003e1bd7 100644 --- a/internal/stream/offline_sub_stream.go +++ b/internal/stream/offline_sub_stream.go @@ -25,7 +25,7 @@ type offlineSubStream struct { func (o *offlineSubStream) initialize() error { o.subStream = &SubStream{ Stream: o.stream, - CurDesc: o.stream.offlineDesc, + InDesc: o.stream.offlineDesc, UseRTPPackets: false, } err := o.subStream.Initialize() @@ -36,9 +36,9 @@ func (o *offlineSubStream) initialize() error { o.ctx, o.ctxCancel = context.WithCancel(context.Background()) pos := 0 - o.tracks = make([]*offlineSubStreamTrack, len(o.subStream.CurDesc.Medias)) + o.tracks = make([]*offlineSubStreamTrack, len(o.subStream.InDesc.Medias)) - for _, media := range o.subStream.CurDesc.Medias { + for _, media := range o.subStream.InDesc.Medias { for _, forma := range media.Formats { t := &offlineSubStreamTrack{ wg: &o.wg, diff --git a/internal/stream/reader.go b/internal/stream/reader.go index 56b7386c..4186d615 100644 --- a/internal/stream/reader.go +++ b/internal/stream/reader.go @@ -29,14 +29,14 @@ type Reader struct { } // OnData registers a callback that is called when data from given format is available. -func (r *Reader) OnData(medi *description.Media, forma format.Format, cb OnDataFunc) { +func (r *Reader) OnData(origMedia *description.Media, origFormat format.Format, cb OnDataFunc) { if r.onDatas == nil { r.onDatas = make(map[*description.Media]map[format.Format]OnDataFunc) } - if r.onDatas[medi] == nil { - r.onDatas[medi] = make(map[format.Format]OnDataFunc) + if r.onDatas[origMedia] == nil { + r.onDatas[origMedia] = make(map[format.Format]OnDataFunc) } - r.onDatas[medi][forma] = cb + r.onDatas[origMedia][origFormat] = cb } // Formats returns all formats for which the reader has registered a OnData callback. diff --git a/internal/stream/stream.go b/internal/stream/stream.go index e4c2527b..d9a6253b 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -3,6 +3,7 @@ package stream import ( "os" + "reflect" "sync" "sync/atomic" "time" @@ -222,97 +223,38 @@ func mediasFromAlwaysAvailableTracks(alwaysAvailableTracks []conf.AlwaysAvailabl return medias } -// only fields filled by mediasFromAlwaysAvailableFile and mediasFromAlwaysAvailableTracks are cloned -func cloneFormat(forma format.Format) format.Format { - switch forma := forma.(type) { - case *format.AV1: - return &format.AV1{ - PayloadTyp: forma.PayloadTyp, - } - - case *format.VP9: - return &format.VP9{ - PayloadTyp: forma.PayloadTyp, - } - - case *format.H265: - return &format.H265{ - PayloadTyp: forma.PayloadTyp, - VPS: forma.VPS, - SPS: forma.SPS, - PPS: forma.PPS, - } - - case *format.H264: - return &format.H264{ - PayloadTyp: forma.PayloadTyp, - PacketizationMode: forma.PacketizationMode, - SPS: forma.SPS, - PPS: forma.PPS, - } - - case *format.Opus: - return &format.Opus{ - PayloadTyp: forma.PayloadTyp, - ChannelCount: forma.ChannelCount, - } - - case *format.MPEG4Audio: - return &format.MPEG4Audio{ - PayloadTyp: forma.PayloadTyp, - SizeLength: forma.SizeLength, - IndexLength: forma.IndexLength, - IndexDeltaLength: forma.IndexDeltaLength, - Config: forma.Config, - } - - case *format.G711: - return &format.G711{ - PayloadTyp: forma.PayloadTyp, - MULaw: forma.MULaw, - SampleRate: forma.SampleRate, - ChannelCount: forma.ChannelCount, - } - - case *format.LPCM: - return &format.LPCM{ - PayloadTyp: forma.PayloadTyp, - BitDepth: forma.BitDepth, - SampleRate: forma.SampleRate, - ChannelCount: forma.ChannelCount, - } - - default: - panic("unsupported format") - } +func cloneFormatShallow(forma format.Format) format.Format { + v := reflect.New(reflect.TypeOf(forma).Elem()) + v.Elem().Set(reflect.ValueOf(forma).Elem()) + return v.Interface().(format.Format) } -// only fields filled by mediasFromAlwaysAvailableFile and mediasFromAlwaysAvailableTracks are cloned func cloneDesc(desc *description.Session) *description.Session { - medias := make([]*description.Media, len(desc.Medias)) + out := &description.Session{ + Title: desc.Title, + Medias: make([]*description.Media, len(desc.Medias)), + } for i, media := range desc.Medias { formats := make([]format.Format, len(media.Formats)) for j, forma := range media.Formats { - formats[j] = cloneFormat(forma) + formats[j] = cloneFormatShallow(forma) } - medias[i] = &description.Media{ + out.Medias[i] = &description.Media{ Type: media.Type, Formats: formats, } } - return &description.Session{ - Medias: medias, - } + return out } // Stream is a media stream. // It stores tracks, readers and allows to write data to readers, remuxing it when needed. type Stream struct { - Desc *description.Session + OrigDesc *description.Session AlwaysAvailable bool AlwaysAvailableTracks []conf.AlwaysAvailableTrack AlwaysAvailableFile string @@ -321,6 +263,9 @@ type Stream struct { ReplaceNTP bool Parent logger.Writer + outDescMutex sync.RWMutex + outDesc *description.Session + offlineDesc *description.Session mutex sync.RWMutex subStream *SubStream @@ -344,7 +289,7 @@ type Stream struct { // Initialize initializes a Stream. func (s *Stream) Initialize() error { if s.AlwaysAvailable { - if s.Desc != nil { + if s.OrigDesc != nil { panic("should not happen") } if !s.ReplaceNTP { @@ -367,8 +312,7 @@ func (s *Stream) Initialize() error { Medias: medias, } - // clone the description since its parameters can be modified - s.Desc = cloneDesc(s.offlineDesc) + s.OrigDesc = cloneDesc(s.offlineDesc) } s.medias = make(map[*description.Media]*streamMedia) @@ -388,9 +332,14 @@ func (s *Stream) Initialize() error { s.lastSystemTime = time.Now() - for _, media := range s.Desc.Medias { + s.outDesc = &description.Session{ + Title: s.OrigDesc.Title, + Medias: make([]*description.Media, len(s.OrigDesc.Medias)), + } + + for i, origMedia := range s.OrigDesc.Medias { sm := &streamMedia{ - media: media, + origMedia: origMedia, alwaysAvailable: s.AlwaysAvailable, rtpMaxPayloadSize: s.RTPMaxPayloadSize, replaceNTP: s.ReplaceNTP, @@ -398,6 +347,7 @@ func (s *Stream) Initialize() error { outboundBytes: &s.outboundBytes, updateLastTime: s.updateLastTime, writeRTSP: s.writeRTSP, + updateOutDesc: s.updateOutDesc, inboundFramesInError: s.inboundFramesInError, parent: s.Parent, } @@ -405,7 +355,9 @@ func (s *Stream) Initialize() error { if err != nil { return err } - s.medias[media] = sm + + s.medias[origMedia] = sm + s.outDesc.Medias[i] = sm.outMedia } if s.AlwaysAvailable { @@ -487,39 +439,45 @@ func (s *Stream) InboundFramesInError() uint64 { } // RTSPStream returns the RTSP stream. -func (s *Stream) RTSPStream(server *gortsplib.Server) *gortsplib.ServerStream { +func (s *Stream) RTSPStream(server *gortsplib.Server) (*gortsplib.ServerStream, error) { s.mutex.Lock() defer s.mutex.Unlock() if s.rtspStream == nil { - s.rtspStream = &gortsplib.ServerStream{ + strm := &gortsplib.ServerStream{ Server: server, - Desc: s.Desc, + Desc: s.outDesc, } - err := s.rtspStream.Initialize() + err := strm.Initialize() if err != nil { - panic(err) + return nil, err } + + s.rtspStream = strm } - return s.rtspStream + + return s.rtspStream, nil } // RTSPSStream returns the RTSPS stream. -func (s *Stream) RTSPSStream(server *gortsplib.Server) *gortsplib.ServerStream { +func (s *Stream) RTSPSStream(server *gortsplib.Server) (*gortsplib.ServerStream, error) { s.mutex.Lock() defer s.mutex.Unlock() if s.rtspsStream == nil { - s.rtspsStream = &gortsplib.ServerStream{ + strm := &gortsplib.ServerStream{ Server: server, - Desc: s.Desc, + Desc: s.outDesc, } - err := s.rtspsStream.Initialize() + err := strm.Initialize() if err != nil { - panic(err) + return nil, err } + + s.rtspsStream = strm } - return s.rtspsStream + + return s.rtspsStream, nil } // AddReader adds a reader. @@ -530,11 +488,11 @@ func (s *Stream) AddReader(r *Reader) { s.readers[r] = struct{}{} - for medi, formats := range r.onDatas { - sm := s.medias[medi] + for origMedia, origFormats := range r.onDatas { + sm := s.medias[origMedia] - for forma, onData := range formats { - sf := sm.formats[forma] + for origFormat, onData := range origFormats { + sf := sm.formats[origFormat] sf.onDatas[r] = onData } } @@ -574,6 +532,29 @@ func (s *Stream) WaitForReaders() { <-s.hasReaders } +// OutDescCopy returns a copy of the output description. +func (s *Stream) OutDescCopy() *description.Session { + s.outDescMutex.RLock() + defer s.outDescMutex.RUnlock() + + return cloneDesc(s.outDesc) +} + +func (s *Stream) updateOutDesc(update func()) { + s.outDescMutex.Lock() + defer s.outDescMutex.Unlock() + + update() + + if s.rtspStream != nil { + s.rtspStream.ReloadDesc() + } + + if s.rtspsStream != nil { + s.rtspsStream.ReloadDesc() + } +} + func (s *Stream) updateLastTime(pts time.Duration) { s.timeMutex.Lock() defer s.timeMutex.Unlock() @@ -587,16 +568,16 @@ func (s *Stream) updateLastTime(pts time.Duration) { s.lastSystemTime = time.Now() } -func (s *Stream) writeRTSP(medi *description.Media, pkts []*rtp.Packet, ntp time.Time) { +func (s *Stream) writeRTSP(outMedia *description.Media, pkts []*rtp.Packet, ntp time.Time) { if s.rtspStream != nil { for _, pkt := range pkts { - s.rtspStream.WritePacketRTPWithNTP(medi, pkt, ntp) //nolint:errcheck + s.rtspStream.WritePacketRTPWithNTP(outMedia, pkt, ntp) //nolint:errcheck } } if s.rtspsStream != nil { for _, pkt := range pkts { - s.rtspsStream.WritePacketRTPWithNTP(medi, pkt, ntp) //nolint:errcheck + s.rtspsStream.WritePacketRTPWithNTP(outMedia, pkt, ntp) //nolint:errcheck } } } diff --git a/internal/stream/stream_alwaysavailable_test.go b/internal/stream/stream_alwaysavailable_test.go index dfa5790d..b8fd1684 100644 --- a/internal/stream/stream_alwaysavailable_test.go +++ b/internal/stream/stream_alwaysavailable_test.go @@ -117,7 +117,7 @@ func TestStreamAlwaysAvailableErrors(t *testing.T) { subStream := &SubStream{ Stream: strm, - CurDesc: ca.desc, + InDesc: ca.desc, UseRTPPackets: false, } err = subStream.Initialize() @@ -325,7 +325,7 @@ func TestStreamAlwaysAvailable(t *testing.T) { var lastPTSAV1 int64 var soAV1a sync.Once var soAV1b sync.Once - r.OnData(strm.Desc.Medias[n], strm.Desc.Medias[n].Formats[0], func(u *unit.Unit) error { + r.OnData(strm.OrigDesc.Medias[n], strm.OrigDesc.Medias[n].Formats[0], func(u *unit.Unit) error { require.GreaterOrEqual(t, u.PTS, lastPTSAV1) lastPTSAV1 = u.PTS @@ -348,7 +348,7 @@ func TestStreamAlwaysAvailable(t *testing.T) { var lastPTSVP9 int64 var soVP9a sync.Once var soVP9b sync.Once - r.OnData(strm.Desc.Medias[n], strm.Desc.Medias[n].Formats[0], func(u *unit.Unit) error { + r.OnData(strm.OrigDesc.Medias[n], strm.OrigDesc.Medias[n].Formats[0], func(u *unit.Unit) error { require.GreaterOrEqual(t, u.PTS, lastPTSVP9) lastPTSVP9 = u.PTS @@ -371,7 +371,7 @@ func TestStreamAlwaysAvailable(t *testing.T) { var lastPTSH265 int64 var soH265a sync.Once var soH265b sync.Once - r.OnData(strm.Desc.Medias[n], strm.Desc.Medias[n].Formats[0], func(u *unit.Unit) error { + r.OnData(strm.OrigDesc.Medias[n], strm.OrigDesc.Medias[n].Formats[0], func(u *unit.Unit) error { require.GreaterOrEqual(t, u.PTS, lastPTSH265) lastPTSH265 = u.PTS @@ -394,7 +394,7 @@ func TestStreamAlwaysAvailable(t *testing.T) { var lastPTSH264 int64 var soH264a sync.Once var soH264b sync.Once - r.OnData(strm.Desc.Medias[n], strm.Desc.Medias[n].Formats[0], func(u *unit.Unit) error { + r.OnData(strm.OrigDesc.Medias[n], strm.OrigDesc.Medias[n].Formats[0], func(u *unit.Unit) error { require.GreaterOrEqual(t, u.PTS, lastPTSH264) lastPTSH264 = u.PTS @@ -417,7 +417,7 @@ func TestStreamAlwaysAvailable(t *testing.T) { var lastPTSOpus int64 var soOpusa sync.Once var soOpusb sync.Once - r.OnData(strm.Desc.Medias[n], strm.Desc.Medias[n].Formats[0], func(u *unit.Unit) error { + r.OnData(strm.OrigDesc.Medias[n], strm.OrigDesc.Medias[n].Formats[0], func(u *unit.Unit) error { require.GreaterOrEqual(t, u.PTS, lastPTSOpus) lastPTSOpus = u.PTS @@ -440,7 +440,7 @@ func TestStreamAlwaysAvailable(t *testing.T) { var lastPTSMPEG4Audio int64 var soMPEG4Audioa sync.Once var soMPEG4Audiob sync.Once - r.OnData(strm.Desc.Medias[n], strm.Desc.Medias[n].Formats[0], func(u *unit.Unit) error { + r.OnData(strm.OrigDesc.Medias[n], strm.OrigDesc.Medias[n].Formats[0], func(u *unit.Unit) error { require.GreaterOrEqual(t, u.PTS, lastPTSMPEG4Audio) lastPTSMPEG4Audio = u.PTS @@ -463,7 +463,7 @@ func TestStreamAlwaysAvailable(t *testing.T) { var lastPTSLPCM int64 var soLPCMa sync.Once var soLPCMb sync.Once - r.OnData(strm.Desc.Medias[n], strm.Desc.Medias[n].Formats[0], func(u *unit.Unit) error { + r.OnData(strm.OrigDesc.Medias[n], strm.OrigDesc.Medias[n].Formats[0], func(u *unit.Unit) error { require.GreaterOrEqual(t, u.PTS, lastPTSLPCM) lastPTSLPCM = u.PTS @@ -488,7 +488,7 @@ func TestStreamAlwaysAvailable(t *testing.T) { subStream := &SubStream{ Stream: strm, - CurDesc: &description.Session{Medias: []*description.Media{ + InDesc: &description.Session{Medias: []*description.Media{ { Type: description.MediaTypeVideo, Formats: []format.Format{&format.AV1{}}, @@ -537,37 +537,37 @@ func TestStreamAlwaysAvailable(t *testing.T) { wg.Add(7) phase2.Store(true) - subStream.WriteUnit(subStream.CurDesc.Medias[0], subStream.CurDesc.Medias[0].Formats[0], &unit.Unit{ + subStream.WriteUnit(subStream.InDesc.Medias[0], subStream.InDesc.Medias[0].Formats[0], &unit.Unit{ PTS: 0, Payload: unit.PayloadAV1{{1, 2, 3, 4}}, }) - subStream.WriteUnit(subStream.CurDesc.Medias[1], subStream.CurDesc.Medias[1].Formats[0], &unit.Unit{ + subStream.WriteUnit(subStream.InDesc.Medias[1], subStream.InDesc.Medias[1].Formats[0], &unit.Unit{ PTS: 0, Payload: unit.PayloadVP9{1, 2, 3, 4}, }) - subStream.WriteUnit(subStream.CurDesc.Medias[2], subStream.CurDesc.Medias[2].Formats[0], &unit.Unit{ + subStream.WriteUnit(subStream.InDesc.Medias[2], subStream.InDesc.Medias[2].Formats[0], &unit.Unit{ PTS: 0, Payload: unit.PayloadH265{{1, 2, 3, 4}}, }) - subStream.WriteUnit(subStream.CurDesc.Medias[3], subStream.CurDesc.Medias[3].Formats[0], &unit.Unit{ + subStream.WriteUnit(subStream.InDesc.Medias[3], subStream.InDesc.Medias[3].Formats[0], &unit.Unit{ PTS: 0, Payload: unit.PayloadH264{{1, 2, 3, 4}}, }) - subStream.WriteUnit(subStream.CurDesc.Medias[4], subStream.CurDesc.Medias[4].Formats[0], &unit.Unit{ + subStream.WriteUnit(subStream.InDesc.Medias[4], subStream.InDesc.Medias[4].Formats[0], &unit.Unit{ PTS: 0, Payload: unit.PayloadOpus{{1, 2}}, }) - subStream.WriteUnit(subStream.CurDesc.Medias[5], subStream.CurDesc.Medias[5].Formats[0], &unit.Unit{ + subStream.WriteUnit(subStream.InDesc.Medias[5], subStream.InDesc.Medias[5].Formats[0], &unit.Unit{ PTS: 0, Payload: unit.PayloadMPEG4Audio{{1, 2, 3, 4}}, }) - subStream.WriteUnit(subStream.CurDesc.Medias[6], subStream.CurDesc.Medias[6].Formats[0], &unit.Unit{ + subStream.WriteUnit(subStream.InDesc.Medias[6], subStream.InDesc.Medias[6].Formats[0], &unit.Unit{ PTS: 0, Payload: unit.PayloadLPCM{1, 2, 3, 4}, }) diff --git a/internal/stream/stream_format.go b/internal/stream/stream_format.go index c057989b..f9ef50f5 100644 --- a/internal/stream/stream_format.go +++ b/internal/stream/stream_format.go @@ -5,7 +5,6 @@ import ( "sync/atomic" "time" - "github.com/bluenviron/gortsplib/v5/pkg/description" "github.com/bluenviron/gortsplib/v5/pkg/format" "github.com/pion/rtp" @@ -39,8 +38,7 @@ func randUint32() (uint32, error) { } type streamFormat struct { - format format.Format - media *description.Media + origFormat format.Format alwaysAvailable bool rtpMaxPayloadSize int replaceNTP bool @@ -48,9 +46,11 @@ type streamFormat struct { inboundBytes *atomic.Uint64 outboundBytes *atomic.Uint64 updateLastTime func(time.Duration) - writeRTSP func(*description.Media, []*rtp.Packet, time.Time) + writeRTSP func([]*rtp.Packet, time.Time) + updateOutDesc func(func()) parent logger.Writer + outFormat format.Format ptsOffset int64 formatUpdater formatUpdater unitRemuxer unitRemuxer @@ -61,12 +61,13 @@ type streamFormat struct { } func (sf *streamFormat) initialize() error { - sf.formatUpdater = newFormatUpdater(sf.format) - sf.unitRemuxer = newUnitRemuxer(sf.format) + sf.outFormat = cloneFormatShallow(sf.origFormat) + sf.formatUpdater = newFormatUpdater(sf.outFormat) + sf.unitRemuxer = newUnitRemuxer(sf.outFormat) if sf.replaceNTP { sf.ntpEstimator = &ntpestimator.Estimator{ - ClockRate: sf.format.ClockRate(), + ClockRate: sf.outFormat.ClockRate(), } } diff --git a/internal/stream/stream_media.go b/internal/stream/stream_media.go index f3ccb170..dac6afc8 100644 --- a/internal/stream/stream_media.go +++ b/internal/stream/stream_media.go @@ -12,7 +12,7 @@ import ( ) type streamMedia struct { - media *description.Media + origMedia *description.Media alwaysAvailable bool rtpMaxPayloadSize int replaceNTP bool @@ -20,19 +20,26 @@ type streamMedia struct { outboundBytes *atomic.Uint64 updateLastTime func(time.Duration) writeRTSP func(*description.Media, []*rtp.Packet, time.Time) + updateOutDesc func(func()) inboundFramesInError *errordumper.Dumper parent logger.Writer - formats map[format.Format]*streamFormat + outMedia *description.Media + formats map[format.Format]*streamFormat } func (sm *streamMedia) initialize() error { sm.formats = make(map[format.Format]*streamFormat) - for _, forma := range sm.media.Formats { + sm.outMedia = &description.Media{ + Type: sm.origMedia.Type, + ID: sm.origMedia.ID, + Formats: make([]format.Format, len(sm.origMedia.Formats)), + } + + for i, origFormat := range sm.origMedia.Formats { sf := &streamFormat{ - format: forma, - media: sm.media, + origFormat: origFormat, alwaysAvailable: sm.alwaysAvailable, rtpMaxPayloadSize: sm.rtpMaxPayloadSize, replaceNTP: sm.replaceNTP, @@ -40,15 +47,22 @@ func (sm *streamMedia) initialize() error { inboundBytes: sm.inboundBytes, outboundBytes: sm.outboundBytes, updateLastTime: sm.updateLastTime, - writeRTSP: sm.writeRTSP, + writeRTSP: sm.writeRTSPWrapper, + updateOutDesc: sm.updateOutDesc, parent: sm.parent, } err := sf.initialize() if err != nil { return err } - sm.formats[forma] = sf + + sm.formats[origFormat] = sf + sm.outMedia.Formats[i] = sf.outFormat } return nil } + +func (sm *streamMedia) writeRTSPWrapper(pkts []*rtp.Packet, ntp time.Time) { + sm.writeRTSP(sm.outMedia, pkts, ntp) +} diff --git a/internal/stream/stream_standard_test.go b/internal/stream/stream_standard_test.go index 3eb4386f..8935c765 100644 --- a/internal/stream/stream_standard_test.go +++ b/internal/stream/stream_standard_test.go @@ -29,7 +29,7 @@ func TestStream(t *testing.T) { }} strm := &Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, } @@ -82,7 +82,7 @@ func TestStreamSkipOutboundBytes(t *testing.T) { }} strm := &Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, } @@ -141,7 +141,7 @@ func TestStreamResizeOversizedRTPPackets(t *testing.T) { }} strm := &Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 400, Parent: &nilLogger{}, @@ -237,8 +237,6 @@ func TestStreamResizeOversizedRTPPackets(t *testing.T) { func TestStreamUpdateFormatParams(t *testing.T) { for _, ca := range []string{"h264", "h265", "mpeg4video"} { t.Run(ca, func(t *testing.T) { - var desc *description.Session - var media *description.Media var forma format.Format var u *unit.Unit @@ -250,19 +248,12 @@ func TestStreamUpdateFormatParams(t *testing.T) { } pps := []byte{0x08, 0x07, 0x08, 0x09} - formatH264 := &format.H264{ + forma = &format.H264{ PacketizationMode: 1, SPS: []byte{0x67, 0x42, 0xc0, 0x28}, PPS: []byte{0x08, 0x06}, } - desc = &description.Session{Medias: []*description.Media{{ - Type: description.MediaTypeVideo, - Formats: []format.Format{formatH264}, - }}} - media = desc.Medias[0] - forma = formatH264 - u = &unit.Unit{ PTS: 90000, Payload: unit.PayloadH264{ @@ -277,19 +268,12 @@ func TestStreamUpdateFormatParams(t *testing.T) { sps := []byte{0x42, 0x01, 0x01, 0x01, 0x60, 0x00, 0x00, 0x04} pps := []byte{0x44, 0x01, 0xc1, 0x73, 0xd1, 0x8a} - formatH265 := &format.H265{ + forma = &format.H265{ VPS: []byte{0x40, 0x01, 0x0c}, SPS: []byte{0x42, 0x01, 0x01}, PPS: []byte{0x44, 0x01, 0xc1}, } - desc = &description.Session{Medias: []*description.Media{{ - Type: description.MediaTypeVideo, - Formats: []format.Format{formatH265}, - }}} - media = desc.Medias[0] - forma = formatH265 - u = &unit.Unit{ PTS: 90000, Payload: unit.PayloadH265{ @@ -307,17 +291,10 @@ func TestStreamUpdateFormatParams(t *testing.T) { 0x14, 0x00, 0x00, 0x01, 0x00, } - formatMPEG4Video := &format.MPEG4Video{ + forma = &format.MPEG4Video{ Config: []byte{0x00, 0x00, 0x01, 0xb0, 0x01}, } - desc = &description.Session{Medias: []*description.Media{{ - Type: description.MediaTypeVideo, - Formats: []format.Format{formatMPEG4Video}, - }}} - media = desc.Medias[0] - forma = formatMPEG4Video - frame := make([]byte, 0, len(config)+20) frame = append(frame, config...) frame = append(frame, []byte{0x00, 0x00, 0x01, 0xb3}...) // Group of VOP @@ -329,8 +306,15 @@ func TestStreamUpdateFormatParams(t *testing.T) { } } + media := &description.Media{ + Type: description.MediaTypeVideo, + Formats: []format.Format{forma}, + } + + desc := &description.Session{Medias: []*description.Media{media}} + strm := &Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, } @@ -360,29 +344,35 @@ func TestStreamUpdateFormatParams(t *testing.T) { <-recv + outDesc := strm.OutDescCopy() + // Verify that format parameters were updated switch ca { case "h264": - formatH264 := forma.(*format.H264) - require.Equal(t, []byte{ - 0x67, 0x64, 0x00, 0x20, 0xac, 0xd9, 0x40, 0x78, - 0x02, 0x27, 0xe5, 0x9a, 0x80, 0x80, 0x80, 0xa0, - }, formatH264.SPS) - require.Equal(t, []byte{0x08, 0x07, 0x08, 0x09}, formatH264.PPS) + require.Equal(t, &format.H264{ + SPS: []byte{ + 0x67, 0x64, 0x00, 0x20, 0xac, 0xd9, 0x40, 0x78, + 0x02, 0x27, 0xe5, 0x9a, 0x80, 0x80, 0x80, 0xa0, + }, + PPS: []byte{0x08, 0x07, 0x08, 0x09}, + PacketizationMode: 1, + }, outDesc.Medias[0].Formats[0]) case "h265": - formatH265 := forma.(*format.H265) - require.Equal(t, []byte{0x40, 0x01, 0x0c, 0x01, 0xff, 0xfe}, formatH265.VPS) - require.Equal(t, []byte{0x42, 0x01, 0x01, 0x01, 0x60, 0x00, 0x00, 0x04}, formatH265.SPS) - require.Equal(t, []byte{0x44, 0x01, 0xc1, 0x73, 0xd1, 0x8a}, formatH265.PPS) + require.Equal(t, &format.H265{ + VPS: []byte{0x40, 0x01, 0x0c, 0x01, 0xff, 0xfe}, + SPS: []byte{0x42, 0x01, 0x01, 0x01, 0x60, 0x00, 0x00, 0x04}, + PPS: []byte{0x44, 0x01, 0xc1, 0x73, 0xd1, 0x8a}, + }, outDesc.Medias[0].Formats[0]) case "mpeg4video": - formatMPEG4Video := forma.(*format.MPEG4Video) - require.Equal(t, []byte{ - 0x00, 0x00, 0x01, 0xb0, - 0x02, 0x00, 0x00, 0x01, 0xb5, 0x8a, - 0x14, 0x00, 0x00, 0x01, 0x00, - }, formatMPEG4Video.Config) + require.Equal(t, &format.MPEG4Video{ + Config: []byte{ + 0x00, 0x00, 0x01, 0xb0, + 0x02, 0x00, 0x00, 0x01, 0xb5, 0x8a, + 0x14, 0x00, 0x00, 0x01, 0x00, + }, + }, outDesc.Medias[0].Formats[0]) } }) } @@ -1226,7 +1216,7 @@ func TestStreamDecode(t *testing.T) { }}} strm := &Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: &nilLogger{}, @@ -1276,7 +1266,7 @@ func TestStreamEncode(t *testing.T) { }}} strm := &Stream{ - Desc: desc, + OrigDesc: desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, Parent: &nilLogger{}, diff --git a/internal/stream/sub_stream.go b/internal/stream/sub_stream.go index e1acb224..57e83657 100644 --- a/internal/stream/sub_stream.go +++ b/internal/stream/sub_stream.go @@ -87,7 +87,7 @@ func mediasAreCompatible(medias1 []*description.Media, medias2 []*description.Me // SubStream is a Stream without interruptions. type SubStream struct { Stream *Stream - CurDesc *description.Session + InDesc *description.Session UseRTPPackets bool medias map[*description.Media]*subStreamMedia @@ -100,39 +100,39 @@ func (ss *SubStream) Initialize() error { panic("should not happen") } - if ss.CurDesc != nil { + if ss.InDesc != nil { panic("should not happen") } } else { - if ss.CurDesc == nil { + if ss.InDesc == nil { panic("should not happen") } - err := mediasAreCompatible(ss.Stream.Desc.Medias, ss.CurDesc.Medias) + err := mediasAreCompatible(ss.Stream.OrigDesc.Medias, ss.InDesc.Medias) if err != nil { return err } } if !ss.Stream.AlwaysAvailable { - ss.CurDesc = ss.Stream.Desc + ss.InDesc = ss.Stream.OrigDesc } ss.medias = make(map[*description.Media]*subStreamMedia) - for i, curMedia := range ss.CurDesc.Medias { - media := ss.Stream.Desc.Medias[i] + for i, inMedia := range ss.InDesc.Medias { + origMedia := ss.Stream.OrigDesc.Medias[i] ssm := &subStreamMedia{ - curMedia: curMedia, - streamMedia: ss.Stream.medias[media], + inMedia: inMedia, + streamMedia: ss.Stream.medias[origMedia], useRTPPackets: ss.UseRTPPackets, } err := ssm.initialize() if err != nil { return err } - ss.medias[curMedia] = ssm + ss.medias[inMedia] = ssm } if ss.Stream.AlwaysAvailable { @@ -163,7 +163,7 @@ func (ss *SubStream) Initialize() error { } // WriteUnit writes a Unit. -func (ss *SubStream) WriteUnit(medi *description.Media, forma format.Format, u *unit.Unit) { +func (ss *SubStream) WriteUnit(inMedia *description.Media, inFormat format.Format, u *unit.Unit) { ss.Stream.mutex.RLock() defer ss.Stream.mutex.RUnlock() @@ -171,8 +171,8 @@ func (ss *SubStream) WriteUnit(medi *description.Media, forma format.Format, u * return } - ssm := ss.medias[medi] - ssf := ssm.formats[forma] + ssm := ss.medias[inMedia] + ssf := ssm.formats[inFormat] ssf.writeUnit(u) } diff --git a/internal/stream/sub_stream_format.go b/internal/stream/sub_stream_format.go index a3641feb..8aa315d4 100644 --- a/internal/stream/sub_stream_format.go +++ b/internal/stream/sub_stream_format.go @@ -11,19 +11,17 @@ import ( ) type subStreamFormat struct { - curFormat format.Format + inFormat format.Format streamFormat *streamFormat useRTPPackets bool - rtpDecoder rtpDecoder - tempRTPEncoder rtpEncoder - tempRTPTimeOffset uint32 + rtpDecoder rtpDecoder } func (ssf *subStreamFormat) initialize() error { if ssf.useRTPPackets { var err error - ssf.rtpDecoder, err = newRTPDecoder(ssf.curFormat) + ssf.rtpDecoder, err = newRTPDecoder(ssf.inFormat) if err != nil { return err } @@ -31,12 +29,16 @@ func (ssf *subStreamFormat) initialize() error { if ssf.streamFormat.rtpEncoder == nil && (!ssf.useRTPPackets || ssf.streamFormat.alwaysAvailable) { var err error - ssf.tempRTPEncoder, err = newRTPEncoder(ssf.curFormat, ssf.streamFormat.rtpMaxPayloadSize, nil, nil) + ssf.streamFormat.rtpEncoder, err = newRTPEncoder( + ssf.streamFormat.outFormat, + ssf.streamFormat.rtpMaxPayloadSize, + nil, + nil) if err != nil { return err } - ssf.tempRTPTimeOffset, err = randUint32() + ssf.streamFormat.rtpTimeOffset, err = randUint32() if err != nil { return err } @@ -46,45 +48,32 @@ func (ssf *subStreamFormat) initialize() error { } func (ssf *subStreamFormat) initialize2(firstTimeReceived bool, lastPTS time.Duration, lastSystemTime time.Time) { - if ssf.tempRTPEncoder != nil { - if ssf.streamFormat.rtpEncoder == nil { - ssf.streamFormat.rtpEncoder = ssf.tempRTPEncoder - ssf.streamFormat.rtpTimeOffset = ssf.tempRTPTimeOffset - } - - ssf.tempRTPEncoder = nil - ssf.tempRTPTimeOffset = 0 - } - if ssf.streamFormat.alwaysAvailable { if firstTimeReceived { ptsOffsetGo := lastPTS + time.Since(lastSystemTime) ssf.streamFormat.ptsOffset = multiplyAndDivide(int64(ptsOffsetGo), - int64(ssf.streamFormat.format.ClockRate()), int64(time.Second)) + int64(ssf.streamFormat.outFormat.ClockRate()), int64(time.Second)) } - switch curFormat := ssf.curFormat.(type) { + // transfer parameters from inFormat to outFormat by writing them in the stream + switch inFormat := ssf.inFormat.(type) { case *format.H265: - vps, sps, pps := curFormat.SafeParams() - - if vps != nil && sps != nil && pps != nil { + if inFormat.VPS != nil && inFormat.SPS != nil && inFormat.PPS != nil { ssf.writeUnit(&unit.Unit{ PTS: 0, NTP: time.Time{}, RTPPackets: nil, - Payload: unit.PayloadH265([][]byte{vps, sps, pps}), + Payload: unit.PayloadH265([][]byte{inFormat.VPS, inFormat.SPS, inFormat.PPS}), }) } case *format.H264: - sps, pps := curFormat.SafeParams() - - if sps != nil && pps != nil { + if inFormat.SPS != nil && inFormat.PPS != nil { ssf.writeUnit(&unit.Unit{ PTS: 0, NTP: time.Time{}, RTPPackets: nil, - Payload: unit.PayloadH264([][]byte{sps, pps}), + Payload: unit.PayloadH264([][]byte{inFormat.SPS, inFormat.PPS}), }) } } @@ -105,7 +94,7 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { ssf.streamFormat.updateLastTime( multiplyAndDivide2(time.Duration(u.PTS), - time.Second, time.Duration(ssf.streamFormat.format.ClockRate()))) + time.Second, time.Duration(ssf.streamFormat.outFormat.ClockRate()))) } if ssf.streamFormat.replaceNTP { @@ -125,7 +114,7 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { for _, pkt := range u.RTPPackets { if len(pkt.Payload) > ssf.streamFormat.rtpMaxPayloadSize { var err error - ssf.streamFormat.rtpEncoder, err = newRTPEncoder(ssf.streamFormat.format, ssf.streamFormat.rtpMaxPayloadSize, + ssf.streamFormat.rtpEncoder, err = newRTPEncoder(ssf.streamFormat.outFormat, ssf.streamFormat.rtpMaxPayloadSize, new(pkt.SSRC), new(pkt.SequenceNumber)) if err != nil { if _, ok := errors.AsType[rtpEncoderNotAvailableError](err); ok { @@ -151,9 +140,9 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { } if !u.NilPayload() { - ssf.streamFormat.formatUpdater(ssf.streamFormat.format, u.Payload) + ssf.streamFormat.formatUpdater(ssf.streamFormat.outFormat, u.Payload, ssf.streamFormat.updateOutDesc) - u.Payload = ssf.streamFormat.unitRemuxer(ssf.streamFormat.format, u.Payload) + u.Payload = ssf.streamFormat.unitRemuxer(ssf.streamFormat.outFormat, u.Payload) if ssf.streamFormat.rtpEncoder != nil && !u.NilPayload() { var err error @@ -171,7 +160,7 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { size := unitSize(u) ssf.streamFormat.inboundBytes.Add(size) - ssf.streamFormat.writeRTSP(ssf.streamFormat.media, u.RTPPackets, u.NTP) + ssf.streamFormat.writeRTSP(u.RTPPackets, u.NTP) for sr, onData := range ssf.streamFormat.onDatas { csr := sr diff --git a/internal/stream/sub_stream_media.go b/internal/stream/sub_stream_media.go index e767a26f..a46fd48c 100644 --- a/internal/stream/sub_stream_media.go +++ b/internal/stream/sub_stream_media.go @@ -6,7 +6,7 @@ import ( ) type subStreamMedia struct { - curMedia *description.Media + inMedia *description.Media streamMedia *streamMedia useRTPPackets bool @@ -16,19 +16,19 @@ type subStreamMedia struct { func (ssm *subStreamMedia) initialize() error { ssm.formats = make(map[format.Format]*subStreamFormat) - for i, curFormat := range ssm.curMedia.Formats { - forma := ssm.streamMedia.media.Formats[i] + for i, inFormat := range ssm.inMedia.Formats { + origFormat := ssm.streamMedia.origMedia.Formats[i] ssf := &subStreamFormat{ - curFormat: curFormat, - streamFormat: ssm.streamMedia.formats[forma], + inFormat: inFormat, + streamFormat: ssm.streamMedia.formats[origFormat], useRTPPackets: ssm.useRTPPackets, } err := ssf.initialize() if err != nil { return err } - ssm.formats[curFormat] = ssf + ssm.formats[inFormat] = ssf } return nil diff --git a/internal/test/static_source_parent.go b/internal/test/static_source_parent.go index 2d8bd047..b724b567 100644 --- a/internal/test/static_source_parent.go +++ b/internal/test/static_source_parent.go @@ -30,7 +30,7 @@ func (p *StaticSourceParent) Close() { // SetReady implements parent. func (p *StaticSourceParent) SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes { p.stream = &stream.Stream{ - Desc: req.Desc, + OrigDesc: req.Desc, WriteQueueSize: 512, RTPMaxPayloadSize: 1450, ReplaceNTP: req.ReplaceNTP,