OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

rtsp: use ServerStream.ReloadDesc to propagate description changes to RTSP servers (#5856)

Browse Source
This commit is contained in:
Alessandro Ros
2026-06-10 23:46:02 +02:00
committed by GitHub
parent 01602555d7
commit 7e576d077f
42 changed files with 596 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/2-features/06-authentication.md
+2 -1
View File
@@ -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.
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
internal/core/path.go
+16 -7
View File
@@ -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)
internal/defs/api_path_track.go
+9 -10
View File
@@ -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))
}
internal/defs/api_path_track_codec_props.go
+26 -28
View File
@@ -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:
internal/protocols/hls/from_stream.go
+52 -21
View File
@@ -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))
internal/protocols/hls/from_stream_test.go
+4 -4
View File
@@ -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))
internal/protocols/hls/to_stream_test.go
+1 -1
View File
@@ -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,
internal/protocols/rtmp/from_stream.go
+78 -74
View File
@@ -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++
}
internal/protocols/rtmp/from_stream_test.go
+6 -6
View File
@@ -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)
internal/protocols/webrtc/from_stream_test.go
+8 -8
View File
@@ -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{{
internal/recorder/format_fmp4.go
+76 -63
View File
@@ -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))
internal/recorder/format_mpegts.go
+2 -2
View File
@@ -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))
internal/recorder/recorder_test.go
+7 -7
View File
@@ -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,
internal/servers/hls/muxer_instance.go
+5 -1
View File
@@ -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
}
internal/servers/hls/server_test.go
+4 -4
View File
@@ -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,
internal/servers/hls/session.go
+3 -2
View File
@@ -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 {
internal/servers/moq/server_test.go
+2 -3
View File
@@ -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)
}
internal/servers/moq/session.go
+1 -1
View File
@@ -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})
internal/servers/rtmp/conn.go
+7 -1
View File
@@ -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
}
internal/servers/rtmp/server_test.go
+4 -4
View File
@@ -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,
internal/servers/rtsp/conn.go
+8 -2
View File
@@ -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{
internal/servers/rtsp/server_test.go
+8 -8
View File
@@ -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,
internal/servers/rtsp/session.go
+19 -10
View File
@@ -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{
internal/servers/srt/conn.go
+1 -1
View File
@@ -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
}
internal/servers/srt/server_test.go
+4 -4
View File
@@ -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,
internal/servers/webrtc/server_test.go
+7 -7
View File
@@ -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:
internal/servers/webrtc/session.go
+1 -1
View File
@@ -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
}
internal/staticsources/rpicamera/source.go
+2 -2
View File
@@ -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]
internal/stream/format_updater.go
+22 -13
View File
@@ -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())) {
})
}
}
internal/stream/offline_sub_stream.go
+3 -3
View File
@@ -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,
internal/stream/reader.go
+4 -4
View File
@@ -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.
internal/stream/stream.go
+77 -96
View File
@@ -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
}
}
}
internal/stream/stream_alwaysavailable_test.go
+16 -16
View File
@@ -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},
})
internal/stream/stream_format.go
+8 -7
View File
@@ -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(),
}
}
internal/stream/stream_media.go
+21 -7
View File
@@ -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)
}
internal/stream/stream_standard_test.go
+38 -48
View File
@@ -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{},
internal/stream/sub_stream.go
+13 -13
View File
@@ -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)
}
internal/stream/sub_stream_format.go
+21 -32
View File
@@ -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
internal/stream/sub_stream_media.go
+6 -6
View File
@@ -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
internal/test/static_source_parent.go
+1 -1
View File
@@ -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,