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

do not include recorder and HLS muxer in sent bytes (#4380) (#5039)

Browse Source 
in API (/paths/list, /paths/get) and metrics (paths_bytes_sent), the
amount of sent bytes was increased even in case of writes to the
recorder and HLS muxer, which are not generating network traffic. This
fixes the issue.
This commit is contained in:
Alessandro Ros
2025-09-29 09:28:04 +02:00
committed by GitHub
parent ada39d22cd
commit cd80814009
28 changed files with 615 additions and 656 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/protocols/hls/from_stream.go
+25 -30
View File
@@ -4,6 +4,7 @@ package hls
import (
"errors"
"fmt"
"slices"
"github.com/bluenviron/gohlslib/v2"
"github.com/bluenviron/gohlslib/v2/pkg/codecs"
@@ -20,24 +21,22 @@ var ErrNoSupportedCodecs = errors.New(
"the stream doesn't contain any supported codec, which are currently AV1, VP9, H265, H264, Opus, MPEG-4 Audio")
func setupVideoTrack(
strea *stream.Stream,
reader stream.Reader,
desc *description.Session,
r *stream.Reader,
muxer *gohlslib.Muxer,
setuppedFormats map[format.Format]struct{},
) {
addTrack := func(
media *description.Media,
forma format.Format,
track *gohlslib.Track,
readFunc stream.ReadFunc,
onData stream.OnDataFunc,
) {
muxer.Tracks = append(muxer.Tracks, track)
setuppedFormats[forma] = struct{}{}
strea.AddReader(reader, media, forma, readFunc)
r.OnData(media, forma, onData)
}
var videoFormatAV1 *format.AV1
videoMedia := strea.Desc.FindFormat(&videoFormatAV1)
videoMedia := desc.FindFormat(&videoFormatAV1)
if videoFormatAV1 != nil {
track := &gohlslib.Track{
@@ -72,7 +71,7 @@ func setupVideoTrack(
}
var videoFormatVP9 *format.VP9
videoMedia = strea.Desc.FindFormat(&videoFormatVP9)
videoMedia = desc.FindFormat(&videoFormatVP9)
if videoFormatVP9 != nil {
track := &gohlslib.Track{
@@ -107,7 +106,7 @@ func setupVideoTrack(
}
var videoFormatH265 *format.H265
videoMedia = strea.Desc.FindFormat(&videoFormatH265)
videoMedia = desc.FindFormat(&videoFormatH265)
if videoFormatH265 != nil {
vps, sps, pps := videoFormatH265.SafeParams()
@@ -147,7 +146,7 @@ func setupVideoTrack(
}
var videoFormatH264 *format.H264
videoMedia = strea.Desc.FindFormat(&videoFormatH264)
videoMedia = desc.FindFormat(&videoFormatH264)
if videoFormatH264 != nil {
sps, pps := videoFormatH264.SafeParams()
@@ -187,23 +186,21 @@ func setupVideoTrack(
}
func setupAudioTracks(
strea *stream.Stream,
reader stream.Reader,
desc *description.Session,
r *stream.Reader,
muxer *gohlslib.Muxer,
setuppedFormats map[format.Format]struct{},
) {
addTrack := func(
medi *description.Media,
forma format.Format,
track *gohlslib.Track,
readFunc stream.ReadFunc,
onData stream.OnDataFunc,
) {
muxer.Tracks = append(muxer.Tracks, track)
setuppedFormats[forma] = struct{}{}
strea.AddReader(reader, medi, forma, readFunc)
r.OnData(medi, forma, onData)
}
for _, media := range strea.Desc.Medias {
for _, media := range desc.Medias {
for _, forma := range media.Formats {
switch forma := forma.(type) {
case *format.Opus:
@@ -305,35 +302,33 @@ func setupAudioTracks(
// FromStream maps a MediaMTX stream to a HLS muxer.
func FromStream(
stream *stream.Stream,
reader stream.Reader,
desc *description.Session,
r *stream.Reader,
muxer *gohlslib.Muxer,
) error {
setuppedFormats := make(map[format.Format]struct{})
setupVideoTrack(
stream,
reader,
desc,
r,
muxer,
setuppedFormats,
)
setupAudioTracks(
stream,
reader,
desc,
r,
muxer,
setuppedFormats,
)
if len(muxer.Tracks) == 0 {
return ErrNoSupportedCodecs
}
setuppedFormats := r.Formats()
n := 1
for _, media := range stream.Desc.Medias {
for _, media := range desc.Medias {
for _, forma := range media.Formats {
if _, ok := setuppedFormats[forma]; !ok {
reader.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
if !slices.Contains(setuppedFormats, forma) {
r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
}
n++
}
internal/protocols/hls/from_stream_test.go
+37 -50
View File
@@ -14,71 +14,58 @@ import (
)
func TestFromStreamNoSupportedCodecs(t *testing.T) {
strm := &stream.Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: &description.Session{Medias: []*description.Media{{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
}}},
GenerateRTPPackets: true,
Parent: test.NilLogger,
}
err := strm.Initialize()
require.NoError(t, err)
desc := &description.Session{Medias: []*description.Media{{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
}}}
l := test.Logger(func(logger.Level, string, ...interface{}) {
t.Error("should not happen")
})
r := &stream.Reader{
Parent: test.Logger(func(logger.Level, string, ...interface{}) {
t.Error("should not happen")
}),
}
m := &gohlslib.Muxer{}
err = FromStream(strm, l, m)
err := FromStream(desc, r, m)
require.Equal(t, ErrNoSupportedCodecs, err)
}
func TestFromStreamSkipUnsupportedTracks(t *testing.T) {
strm := &stream.Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP9{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
},
{
Type: description.MediaTypeAudio,
Formats: []format.Format{&format.MPEG1Audio{}},
},
}},
GenerateRTPPackets: true,
Parent: test.NilLogger,
}
err := strm.Initialize()
require.NoError(t, err)
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP9{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
},
{
Type: description.MediaTypeAudio,
Formats: []format.Format{&format.MPEG1Audio{}},
},
}}
m := &gohlslib.Muxer{}
n := 0
l := test.Logger(func(l logger.Level, format string, args ...interface{}) {
require.Equal(t, logger.Warn, l)
switch n {
case 0:
require.Equal(t, "skipping track 2 (VP8)", fmt.Sprintf(format, args...))
case 1:
require.Equal(t, "skipping track 3 (MPEG-1/2 Audio)", fmt.Sprintf(format, args...))
}
n++
})
r := &stream.Reader{
Parent: test.Logger(func(l logger.Level, format string, args ...interface{}) {
require.Equal(t, logger.Warn, l)
switch n {
case 0:
require.Equal(t, "skipping track 2 (VP8)", fmt.Sprintf(format, args...))
case 1:
require.Equal(t, "skipping track 3 (MPEG-1/2 Audio)", fmt.Sprintf(format, args...))
}
n++
}),
}
err = FromStream(strm, l, m)
err := FromStream(desc, r, m)
require.NoError(t, err)
defer strm.RemoveReader(l)
require.Equal(t, 2, n)
}
internal/protocols/hls/to_stream_test.go
+4 -5
View File
@@ -86,7 +86,7 @@ func TestToStream(t *testing.T) {
var strm *stream.Stream
done := make(chan struct{})
reader := test.NilLogger
r := &stream.Reader{Parent: test.NilLogger}
var c *gohlslib.Client
c = &gohlslib.Client{
@@ -112,8 +112,7 @@ func TestToStream(t *testing.T) {
err2 = strm.Initialize()
require.NoError(t, err2)
strm.AddReader(
reader,
r.OnData(
medias[0],
medias[0].Formats[0],
func(u unit.Unit) error {
@@ -122,7 +121,7 @@ func TestToStream(t *testing.T) {
return nil
})
strm.StartReader(reader)
strm.AddReader(r)
return nil
},
@@ -133,6 +132,6 @@ func TestToStream(t *testing.T) {
<-done
strm.RemoveReader(reader)
strm.RemoveReader(r)
strm.Close()
}
internal/protocols/mpegts/from_stream.go
+12 -10
View File
@@ -3,6 +3,7 @@ package mpegts
import (
"bufio"
"fmt"
"slices"
"time"
"github.com/bluenviron/gortsplib/v5/pkg/description"
@@ -27,28 +28,26 @@ func multiplyAndDivide(v, m, d int64) int64 {
// FromStream maps a MediaMTX stream to a MPEG-TS writer.
func FromStream(
strea *stream.Stream,
reader stream.Reader,
desc *description.Session,
r *stream.Reader,
bw *bufio.Writer,
sconn srt.Conn,
writeTimeout time.Duration,
) error {
var w *mcmpegts.Writer
var tracks []*mcmpegts.Track
setuppedFormats := make(map[format.Format]struct{})
addTrack := func(
media *description.Media,
forma format.Format,
track *mcmpegts.Track,
readFunc stream.ReadFunc,
onData stream.OnDataFunc,
) {
tracks = append(tracks, track)
setuppedFormats[forma] = struct{}{}
strea.AddReader(reader, media, forma, readFunc)
r.OnData(media, forma, onData)
}
for _, media := range strea.Desc.Medias {
for _, media := range desc.Medias {
for _, forma := range media.Formats {
clockRate := forma.ClockRate()
@@ -233,6 +232,7 @@ func FromStream(
}
return bw.Flush()
})
case *format.KLV:
track := &mcmpegts.Track{
Codec: &mcmpegts.CodecKLV{
@@ -414,11 +414,13 @@ func FromStream(
return errNoSupportedCodecs
}
setuppedFormats := r.Formats()
n := 1
for _, medi := range strea.Desc.Medias {
for _, medi := range desc.Medias {
for _, forma := range medi.Formats {
if _, ok := setuppedFormats[forma]; !ok {
reader.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
if !slices.Contains(setuppedFormats, forma) {
r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
}
n++
}
internal/protocols/mpegts/from_stream_test.go
+30 -43
View File
@@ -13,60 +13,47 @@ import (
)
func TestFromStreamNoSupportedCodecs(t *testing.T) {
strm := &stream.Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: &description.Session{Medias: []*description.Media{{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
}}},
GenerateRTPPackets: true,
Parent: test.NilLogger,
desc := &description.Session{Medias: []*description.Media{{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
}}}
r := &stream.Reader{
Parent: test.Logger(func(logger.Level, string, ...interface{}) {
t.Error("should not happen")
}),
}
err := strm.Initialize()
require.NoError(t, err)
l := test.Logger(func(logger.Level, string, ...interface{}) {
t.Error("should not happen")
})
err = FromStream(strm, l, nil, nil, 0)
err := FromStream(desc, r, nil, nil, 0)
require.Equal(t, errNoSupportedCodecs, err)
}
func TestFromStreamSkipUnsupportedTracks(t *testing.T) {
strm := &stream.Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H265{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
},
}},
GenerateRTPPackets: true,
Parent: test.NilLogger,
}
err := strm.Initialize()
require.NoError(t, err)
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H265{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
},
}}
n := 0
l := test.Logger(func(l logger.Level, format string, args ...interface{}) {
require.Equal(t, logger.Warn, l)
if n == 0 {
require.Equal(t, "skipping track 2 (VP8)", fmt.Sprintf(format, args...))
}
n++
})
r := &stream.Reader{
Parent: test.Logger(func(l logger.Level, format string, args ...interface{}) {
require.Equal(t, logger.Warn, l)
if n == 0 {
require.Equal(t, "skipping track 2 (VP8)", fmt.Sprintf(format, args...))
}
n++
}),
}
err = FromStream(strm, l, nil, nil, 0)
err := FromStream(desc, r, nil, nil, 0)
require.NoError(t, err)
defer strm.RemoveReader(l)
require.Equal(t, 1, n)
}
internal/protocols/rtmp/from_stream.go
+17 -27
View File
@@ -9,6 +9,7 @@ import (
"github.com/bluenviron/gortmplib"
"github.com/bluenviron/gortmplib/pkg/message"
"github.com/bluenviron/gortsplib/v5/pkg/description"
"github.com/bluenviron/gortsplib/v5/pkg/format"
"github.com/bluenviron/mediacommon/v2/pkg/codecs/ac3"
"github.com/bluenviron/mediacommon/v2/pkg/codecs/h264"
@@ -37,8 +38,8 @@ func timestampToDuration(t int64, clockRate int) time.Duration {
// FromStream maps a MediaMTX stream to a RTMP stream.
func FromStream(
str *stream.Stream,
reader stream.Reader,
desc *description.Session,
r *stream.Reader,
conn *gortmplib.ServerConn,
nconn net.Conn,
writeTimeout time.Duration,
@@ -46,13 +47,12 @@ func FromStream(
var tracks []format.Format
var w *gortmplib.Writer
for _, media := range str.Desc.Medias {
for _, media := range desc.Medias {
for _, forma := range media.Formats {
switch forma := forma.(type) {
case *format.AV1:
if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCAV1))) {
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -74,8 +74,7 @@ func FromStream(
case *format.VP9:
if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCVP9))) {
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -99,8 +98,7 @@ func FromStream(
if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCHEVC))) {
var videoDTSExtractor *h265.DTSExtractor
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -137,8 +135,7 @@ func FromStream(
case *format.H264:
var videoDTSExtractor *h264.DTSExtractor
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -191,8 +188,7 @@ func FromStream(
case *format.Opus:
if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCOpus))) {
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -225,8 +221,7 @@ func FromStream(
}
case *format.MPEG4Audio:
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -257,8 +252,7 @@ func FromStream(
case *format.MPEG4AudioLATM:
if !forma.CPresent {
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -289,8 +283,7 @@ func FromStream(
case *format.MPEG1Audio:
// TODO: check sample rate and layer,
// unfortunately they are not available at this stage.
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -325,8 +318,7 @@ func FromStream(
case *format.AC3:
if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCAC3))) {
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -353,8 +345,7 @@ func FromStream(
case *format.G711:
if forma.SampleRate == 8000 {
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -381,8 +372,7 @@ func FromStream(
forma.SampleRate == 11025 ||
forma.SampleRate == 22050 ||
forma.SampleRate == 44100) {
str.AddReader(
reader,
r.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -420,10 +410,10 @@ func FromStream(
}
n := 1
for _, media := range str.Desc.Medias {
for _, media := range desc.Medias {
for _, forma := range media.Formats {
if !slices.Contains(tracks, forma) {
reader.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
}
n++
}
internal/protocols/rtmp/from_stream_test.go
+34 -47
View File
@@ -409,13 +409,13 @@ func TestFromStream(t *testing.T) {
err = conn.Accept()
require.NoError(t, err)
reader := test.NilLogger
r := &stream.Reader{Parent: test.NilLogger}
err = FromStream(strm, reader, conn, nconn, 10*time.Second)
err = FromStream(strm.Desc, r, conn, nconn, 10*time.Second)
require.NoError(t, err)
defer strm.RemoveReader(reader)
strm.StartReader(reader)
strm.AddReader(r)
defer strm.RemoveReader(r)
switch ca {
case "h264 + aac":
@@ -675,56 +675,44 @@ func TestFromStream(t *testing.T) {
}
func TestFromStreamNoSupportedCodecs(t *testing.T) {
strm := &stream.Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: &description.Session{Medias: []*description.Media{{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
}}},
GenerateRTPPackets: true,
Parent: test.NilLogger,
desc := &description.Session{Medias: []*description.Media{{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
}}}
r := &stream.Reader{
Parent: test.Logger(func(logger.Level, string, ...interface{}) {
t.Error("should not happen")
}),
}
err := strm.Initialize()
require.NoError(t, err)
l := test.Logger(func(logger.Level, string, ...interface{}) {
t.Error("should not happen")
})
err = FromStream(strm, l, nil, nil, 0)
err := FromStream(desc, r, nil, nil, 0)
require.Equal(t, errNoSupportedCodecsFrom, err)
}
func TestFromStreamSkipUnsupportedTracks(t *testing.T) {
strm := &stream.Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{}},
},
}},
GenerateRTPPackets: true,
Parent: test.NilLogger,
}
err := strm.Initialize()
require.NoError(t, err)
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{}},
},
}}
n := 0
l := test.Logger(func(l logger.Level, format string, args ...interface{}) {
require.Equal(t, logger.Warn, l)
if n == 0 {
require.Equal(t, "skipping track 1 (VP8)", fmt.Sprintf(format, args...))
}
n++
})
r := &stream.Reader{
Parent: test.Logger(func(l logger.Level, format string, args ...interface{}) {
require.Equal(t, logger.Warn, l)
if n == 0 {
require.Equal(t, "skipping track 1 (VP8)", fmt.Sprintf(format, args...))
}
n++
}),
}
ln, err := net.Listen("tcp", "127.0.0.1:9121")
require.NoError(t, err)
@@ -754,9 +742,8 @@ func TestFromStreamSkipUnsupportedTracks(t *testing.T) {
err = conn.Accept()
require.NoError(t, err)
err = FromStream(strm, l, conn, nil, 0)
err = FromStream(desc, r, conn, nil, 0)
require.NoError(t, err)
defer strm.RemoveReader(l)
require.Equal(t, 1, n)
}
internal/protocols/webrtc/from_stream.go
+30 -39
View File
@@ -6,6 +6,7 @@ import (
"fmt"
"time"
"github.com/bluenviron/gortsplib/v5/pkg/description"
"github.com/bluenviron/gortsplib/v5/pkg/format"
"github.com/bluenviron/gortsplib/v5/pkg/format/rtpav1"
"github.com/bluenviron/gortsplib/v5/pkg/format/rtph264"
@@ -52,12 +53,12 @@ func timestampToDuration(t int64, clockRate int) time.Duration {
}
func setupVideoTrack(
stream *stream.Stream,
reader stream.Reader,
desc *description.Session,
r *stream.Reader,
pc *PeerConnection,
) (format.Format, error) {
var av1Format *format.AV1
media := stream.Desc.FindFormat(&av1Format)
media := desc.FindFormat(&av1Format)
if av1Format != nil { //nolint:dupl
track := &OutgoingTrack{
@@ -77,8 +78,7 @@ func setupVideoTrack(
return nil, err
}
stream.AddReader(
reader,
r.OnData(
media,
av1Format,
func(u unit.Unit) error {
@@ -106,7 +106,7 @@ func setupVideoTrack(
}
var vp9Format *format.VP9
media = stream.Desc.FindFormat(&vp9Format)
media = desc.FindFormat(&vp9Format)
if vp9Format != nil {
track := &OutgoingTrack{
@@ -128,8 +128,7 @@ func setupVideoTrack(
return nil, err
}
stream.AddReader(
reader,
r.OnData(
media,
vp9Format,
func(u unit.Unit) error {
@@ -157,7 +156,7 @@ func setupVideoTrack(
}
var vp8Format *format.VP8
media = stream.Desc.FindFormat(&vp8Format)
media = desc.FindFormat(&vp8Format)
if vp8Format != nil { //nolint:dupl
track := &OutgoingTrack{
@@ -177,8 +176,7 @@ func setupVideoTrack(
return nil, err
}
stream.AddReader(
reader,
r.OnData(
media,
vp8Format,
func(u unit.Unit) error {
@@ -206,7 +204,7 @@ func setupVideoTrack(
}
var h265Format *format.H265
media = stream.Desc.FindFormat(&h265Format)
media = desc.FindFormat(&h265Format)
if h265Format != nil { //nolint:dupl
track := &OutgoingTrack{
@@ -230,8 +228,7 @@ func setupVideoTrack(
firstReceived := false
var lastPTS int64
stream.AddReader(
reader,
r.OnData(
media,
h265Format,
func(u unit.Unit) error {
@@ -266,7 +263,7 @@ func setupVideoTrack(
}
var h264Format *format.H264
media = stream.Desc.FindFormat(&h264Format)
media = desc.FindFormat(&h264Format)
if h264Format != nil { //nolint:dupl
track := &OutgoingTrack{
@@ -290,8 +287,7 @@ func setupVideoTrack(
firstReceived := false
var lastPTS int64
stream.AddReader(
reader,
r.OnData(
media,
h264Format,
func(u unit.Unit) error {
@@ -329,12 +325,12 @@ func setupVideoTrack(
}
func setupAudioTrack(
stream *stream.Stream,
reader stream.Reader,
desc *description.Session,
r *stream.Reader,
pc *PeerConnection,
) (format.Format, error) {
var opusFormat *format.Opus
media := stream.Desc.FindFormat(&opusFormat)
media := desc.FindFormat(&opusFormat)
if opusFormat != nil {
var caps webrtc.RTPCodecCapability
@@ -371,8 +367,7 @@ func setupAudioTrack(
}
pc.OutgoingTracks = append(pc.OutgoingTracks, track)
stream.AddReader(
reader,
r.OnData(
media,
opusFormat,
func(u unit.Unit) error {
@@ -388,7 +383,7 @@ func setupAudioTrack(
}
var g722Format *format.G722
media = stream.Desc.FindFormat(&g722Format)
media = desc.FindFormat(&g722Format)
if g722Format != nil {
track := &OutgoingTrack{
@@ -399,8 +394,7 @@ func setupAudioTrack(
}
pc.OutgoingTracks = append(pc.OutgoingTracks, track)
stream.AddReader(
reader,
r.OnData(
media,
g722Format,
func(u unit.Unit) error {
@@ -416,7 +410,7 @@ func setupAudioTrack(
}
var g711Format *format.G711
media = stream.Desc.FindFormat(&g711Format)
media = desc.FindFormat(&g711Format)
if g711Format != nil {
// These are the sample rates and channels supported by Chrome.
@@ -479,8 +473,7 @@ func setupAudioTrack(
return nil, err
}
stream.AddReader(
reader,
r.OnData(
media,
g711Format,
func(u unit.Unit) error {
@@ -513,8 +506,7 @@ func setupAudioTrack(
return nil, err
}
stream.AddReader(
reader,
r.OnData(
media,
g711Format,
func(u unit.Unit) error {
@@ -559,7 +551,7 @@ func setupAudioTrack(
}
var lpcmFormat *format.LPCM
media = stream.Desc.FindFormat(&lpcmFormat)
media = desc.FindFormat(&lpcmFormat)
if lpcmFormat != nil {
if lpcmFormat.BitDepth != 16 {
@@ -602,8 +594,7 @@ func setupAudioTrack(
return nil, err
}
stream.AddReader(
reader,
r.OnData(
media,
lpcmFormat,
func(u unit.Unit) error {
@@ -640,16 +631,16 @@ func setupAudioTrack(
// FromStream maps a MediaMTX stream to a WebRTC connection
func FromStream(
stream *stream.Stream,
reader stream.Reader,
desc *description.Session,
r *stream.Reader,
pc *PeerConnection,
) error {
videoFormat, err := setupVideoTrack(stream, reader, pc)
videoFormat, err := setupVideoTrack(desc, r, pc)
if err != nil {
return err
}
audioFormat, err := setupAudioTrack(stream, reader, pc)
audioFormat, err := setupAudioTrack(desc, r, pc)
if err != nil {
return err
}
@@ -659,10 +650,10 @@ func FromStream(
}
n := 1
for _, media := range stream.Desc.Medias {
for _, media := range desc.Medias {
for _, forma := range media.Formats {
if forma != videoFormat && forma != audioFormat {
reader.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
}
n++
}
internal/protocols/webrtc/from_stream_test.go
+36 -58
View File
@@ -13,62 +13,49 @@ import (
)
func TestFromStreamNoSupportedCodecs(t *testing.T) {
strm := &stream.Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: &description.Session{Medias: []*description.Media{{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.MJPEG{}},
}}},
GenerateRTPPackets: true,
Parent: test.NilLogger,
desc := &description.Session{Medias: []*description.Media{{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.MJPEG{}},
}}}
r := &stream.Reader{
Parent: test.Logger(func(logger.Level, string, ...interface{}) {
t.Error("should not happen")
}),
}
err := strm.Initialize()
require.NoError(t, err)
l := test.Logger(func(logger.Level, string, ...interface{}) {
t.Error("should not happen")
})
err = FromStream(strm, l, nil)
err := FromStream(desc, r, nil)
require.Equal(t, errNoSupportedCodecsFrom, err)
}
func TestFromStreamSkipUnsupportedTracks(t *testing.T) {
strm := &stream.Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.MJPEG{}},
},
}},
GenerateRTPPackets: true,
Parent: test.NilLogger,
}
err := strm.Initialize()
require.NoError(t, err)
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.MJPEG{}},
},
}}
n := 0
l := test.Logger(func(l logger.Level, format string, args ...interface{}) {
require.Equal(t, logger.Warn, l)
if n == 0 {
require.Equal(t, "skipping track 2 (M-JPEG)", fmt.Sprintf(format, args...))
}
n++
})
r := &stream.Reader{
Parent: test.Logger(func(l logger.Level, format string, args ...interface{}) {
require.Equal(t, logger.Warn, l)
if n == 0 {
require.Equal(t, "skipping track 2 (M-JPEG)", fmt.Sprintf(format, args...))
}
n++
}),
}
pc := &PeerConnection{}
err = FromStream(strm, l, pc)
err := FromStream(desc, r, pc)
require.NoError(t, err)
defer strm.RemoveReader(l)
require.Equal(t, 1, n)
}
@@ -76,26 +63,17 @@ func TestFromStreamSkipUnsupportedTracks(t *testing.T) {
func TestFromStream(t *testing.T) {
for _, ca := range toFromStreamCases {
t.Run(ca.name, func(t *testing.T) {
strm := &stream.Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: &description.Session{
Medias: []*description.Media{{
Formats: []format.Format{ca.in},
}},
},
GenerateRTPPackets: false,
Parent: test.NilLogger,
desc := &description.Session{
Medias: []*description.Media{{
Formats: []format.Format{ca.in},
}},
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
pc := &PeerConnection{}
r := &stream.Reader{Parent: test.NilLogger}
err = FromStream(strm, nil, pc)
err := FromStream(desc, r, pc)
require.NoError(t, err)
defer strm.RemoveReader(nil)
require.Equal(t, ca.webrtcCaps, pc.OutgoingTracks[0].Caps)
})
internal/recorder/format_fmp4.go
+19 -34
View File
@@ -3,6 +3,7 @@ package recorder
import (
"bytes"
"fmt"
"slices"
"time"
rtspformat "github.com/bluenviron/gortsplib/v5/pkg/format"
@@ -108,8 +109,6 @@ type formatFMP4 struct {
func (f *formatFMP4) initialize() bool {
nextID := 1
var setuppedFormats []rtspformat.Format
setuppedFormatsMap := make(map[rtspformat.Format]struct{})
addTrack := func(format rtspformat.Format, codec mp4.Codec) *formatFMP4Track {
initTrack := &fmp4.InitTrack{
@@ -125,8 +124,6 @@ func (f *formatFMP4) initialize() bool {
}
f.tracks = append(f.tracks, track)
setuppedFormats = append(setuppedFormats, format)
setuppedFormatsMap[format] = struct{}{}
return track
}
@@ -143,8 +140,7 @@ func (f *formatFMP4) initialize() bool {
firstReceived := false
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -206,8 +202,7 @@ func (f *formatFMP4) initialize() bool {
firstReceived := false
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -296,8 +291,7 @@ func (f *formatFMP4) initialize() bool {
var dtsExtractor *h265.DTSExtractor
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -384,8 +378,7 @@ func (f *formatFMP4) initialize() bool {
var dtsExtractor *h264.DTSExtractor
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -465,8 +458,7 @@ func (f *formatFMP4) initialize() bool {
firstReceived := false
var lastPTS int64
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -519,8 +511,7 @@ func (f *formatFMP4) initialize() bool {
firstReceived := false
var lastPTS int64
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -573,8 +564,7 @@ func (f *formatFMP4) initialize() bool {
parsed := false
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -610,8 +600,7 @@ func (f *formatFMP4) initialize() bool {
}
track := addTrack(forma, codec)
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -647,8 +636,7 @@ func (f *formatFMP4) initialize() bool {
}
track := addTrack(forma, codec)
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -683,8 +671,7 @@ func (f *formatFMP4) initialize() bool {
}
track := addTrack(forma, codec)
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -720,8 +707,7 @@ func (f *formatFMP4) initialize() bool {
parsed := false
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -780,8 +766,7 @@ func (f *formatFMP4) initialize() bool {
parsed := false
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -846,8 +831,7 @@ func (f *formatFMP4) initialize() bool {
}
track := addTrack(forma, codec)
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -886,8 +870,7 @@ func (f *formatFMP4) initialize() bool {
}
track := addTrack(forma, codec)
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -909,15 +892,17 @@ func (f *formatFMP4) initialize() bool {
}
}
if len(setuppedFormats) == 0 {
if len(f.tracks) == 0 {
f.ri.Log(logger.Warn, "no supported tracks found, skipping recording")
return false
}
setuppedFormats := f.ri.reader.Formats()
n := 1
for _, medi := range f.ri.stream.Desc.Medias {
for _, forma := range medi.Formats {
if _, ok := setuppedFormatsMap[forma]; !ok {
if !slices.Contains(setuppedFormats, forma) {
f.ri.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
}
n++
internal/recorder/format_mpegts.go
+26 -38
View File
@@ -5,6 +5,7 @@ import (
"bytes"
"fmt"
"io"
"slices"
"time"
rtspformat "github.com/bluenviron/gortsplib/v5/pkg/format"
@@ -64,17 +65,12 @@ type formatMPEGTS struct {
func (f *formatMPEGTS) initialize() bool {
var tracks []*mpegts.Track
var setuppedFormats []rtspformat.Format
setuppedFormatsMap := make(map[rtspformat.Format]struct{})
addTrack := func(format rtspformat.Format, codec mpegts.Codec) *mpegts.Track {
addTrack := func(codec mpegts.Codec) *mpegts.Track {
track := &mpegts.Track{
Codec: codec,
}
tracks = append(tracks, track)
setuppedFormats = append(setuppedFormats, format)
setuppedFormatsMap[format] = struct{}{}
return track
}
@@ -84,12 +80,11 @@ func (f *formatMPEGTS) initialize() bool {
switch forma := forma.(type) {
case *rtspformat.H265: //nolint:dupl
track := addTrack(forma, &mpegts.CodecH265{})
track := addTrack(&mpegts.CodecH265{})
var dtsExtractor *h265.DTSExtractor
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -130,12 +125,11 @@ func (f *formatMPEGTS) initialize() bool {
})
case *rtspformat.H264: //nolint:dupl
track := addTrack(forma, &mpegts.CodecH264{})
track := addTrack(&mpegts.CodecH264{})
var dtsExtractor *h264.DTSExtractor
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -176,13 +170,12 @@ func (f *formatMPEGTS) initialize() bool {
})
case *rtspformat.MPEG4Video:
track := addTrack(forma, &mpegts.CodecMPEG4Video{})
track := addTrack(&mpegts.CodecMPEG4Video{})
firstReceived := false
var lastPTS int64
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -216,13 +209,12 @@ func (f *formatMPEGTS) initialize() bool {
})
case *rtspformat.MPEG1Video:
track := addTrack(forma, &mpegts.CodecMPEG1Video{})
track := addTrack(&mpegts.CodecMPEG1Video{})
firstReceived := false
var lastPTS int64
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -256,12 +248,11 @@ func (f *formatMPEGTS) initialize() bool {
})
case *rtspformat.Opus:
track := addTrack(forma, &mpegts.CodecOpus{
track := addTrack(&mpegts.CodecOpus{
ChannelCount: forma.ChannelCount,
})
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -286,12 +277,11 @@ func (f *formatMPEGTS) initialize() bool {
})
case *rtspformat.KLV:
track := addTrack(forma, &mpegts.CodecKLV{
track := addTrack(&mpegts.CodecKLV{
Synchronous: true,
})
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -313,12 +303,11 @@ func (f *formatMPEGTS) initialize() bool {
})
case *rtspformat.MPEG4Audio:
track := addTrack(forma, &mpegts.CodecMPEG4Audio{
track := addTrack(&mpegts.CodecMPEG4Audio{
Config: *forma.Config,
})
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -344,12 +333,11 @@ func (f *formatMPEGTS) initialize() bool {
case *rtspformat.MPEG4AudioLATM:
if !forma.CPresent {
track := addTrack(forma, &mpegts.CodecMPEG4Audio{
track := addTrack(&mpegts.CodecMPEG4Audio{
Config: *forma.StreamMuxConfig.Programs[0].Layers[0].AudioSpecificConfig,
})
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -382,10 +370,9 @@ func (f *formatMPEGTS) initialize() bool {
}
case *rtspformat.MPEG1Audio:
track := addTrack(forma, &mpegts.CodecMPEG1Audio{})
track := addTrack(&mpegts.CodecMPEG1Audio{})
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -410,10 +397,9 @@ func (f *formatMPEGTS) initialize() bool {
})
case *rtspformat.AC3:
track := addTrack(forma, &mpegts.CodecAC3{})
track := addTrack(&mpegts.CodecAC3{})
f.ri.stream.AddReader(
f.ri,
f.ri.reader.OnData(
media,
forma,
func(u unit.Unit) error {
@@ -449,15 +435,17 @@ func (f *formatMPEGTS) initialize() bool {
}
}
if len(setuppedFormats) == 0 {
if len(tracks) == 0 {
f.ri.Log(logger.Warn, "no supported tracks found, skipping recording")
return false
}
setuppedFormats := f.ri.reader.Formats()
n := 1
for _, medi := range f.ri.stream.Desc.Medias {
for _, forma := range medi.Formats {
if _, ok := setuppedFormatsMap[forma]; !ok {
if !slices.Contains(setuppedFormats, forma) {
f.ri.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
}
n++
internal/recorder/recorder_instance.go
+8 -4
View File
@@ -33,6 +33,7 @@ type recorderInstance struct {
pathFormat2 string
format2 format
skip bool
reader *stream.Reader
terminate chan struct{}
done chan struct{}
@@ -45,11 +46,14 @@ func (ri *recorderInstance) Log(level logger.Level, format string, args ...inter
func (ri *recorderInstance) initialize() {
ri.pathFormat2 = ri.pathFormat
ri.pathFormat2 = recordstore.PathAddExtension(
strings.ReplaceAll(ri.pathFormat2, "%path", ri.pathName),
ri.format,
)
ri.reader = &stream.Reader{
SkipBytesSent: true,
Parent: ri,
}
ri.terminate = make(chan struct{})
ri.done = make(chan struct{})
@@ -71,7 +75,7 @@ func (ri *recorderInstance) initialize() {
}
if !ri.skip {
ri.stream.StartReader(ri)
ri.stream.AddReader(ri.reader)
}
go ri.run()
@@ -87,13 +91,13 @@ func (ri *recorderInstance) run() {
if !ri.skip {
select {
case err := <-ri.stream.ReaderError(ri):
case err := <-ri.reader.Error():
ri.Log(logger.Error, err.Error())
case <-ri.terminate:
}
ri.stream.RemoveReader(ri)
ri.stream.RemoveReader(ri.reader)
} else {
<-ri.terminate
}
internal/servers/hls/muxer_instance.go
+11 -6
View File
@@ -27,6 +27,7 @@ type muxerInstance struct {
parent logger.Writer
hmuxer *gohlslib.Muxer
reader *stream.Reader
}
func (mi *muxerInstance) initialize() error {
@@ -48,21 +49,25 @@ func (mi *muxerInstance) initialize() error {
},
}
err := hls.FromStream(mi.stream, mi, mi.hmuxer)
mi.reader = &stream.Reader{
SkipBytesSent: true,
Parent: mi,
}
err := hls.FromStream(mi.stream.Desc, mi.reader, mi.hmuxer)
if err != nil {
return err
}
err = mi.hmuxer.Start()
if err != nil {
mi.stream.RemoveReader(mi)
return err
}
mi.Log(logger.Info, "is converting into HLS, %s",
defs.FormatsInfo(mi.stream.ReaderFormats(mi)))
defs.FormatsInfo(mi.reader.Formats()))
mi.stream.StartReader(mi)
mi.stream.AddReader(mi.reader)
return nil
}
@@ -73,7 +78,7 @@ func (mi *muxerInstance) Log(level logger.Level, format string, args ...interfac
}
func (mi *muxerInstance) close() {
mi.stream.RemoveReader(mi)
mi.stream.RemoveReader(mi.reader)
mi.hmuxer.Close()
if mi.hmuxer.Directory != "" {
os.Remove(mi.hmuxer.Directory)
@@ -81,7 +86,7 @@ func (mi *muxerInstance) close() {
}
func (mi *muxerInstance) errorChan() chan error {
return mi.stream.ReaderError(mi)
return mi.reader.Error()
}
func (mi *muxerInstance) handleRequest(ctx *gin.Context) {
internal/servers/hls/server_test.go
+1 -1
View File
@@ -324,7 +324,7 @@ func TestServerRead(t *testing.T) {
s.PathReady(&dummyPath{})
strm.WaitRunningReader()
time.Sleep(500 * time.Millisecond)
for i := range 4 {
strm.WriteUnit(test.MediaH264, test.FormatH264, &unit.H264{
internal/servers/rtmp/conn.go
+8 -6
View File
@@ -151,7 +151,7 @@ func (c *conn) runRead() error {
pathName := strings.TrimLeft(c.rconn.URL.Path, "/")
query := c.rconn.URL.Query()
path, stream, err := c.pathManager.AddReader(defs.PathAddReaderReq{
path, strm, err := c.pathManager.AddReader(defs.PathAddReaderReq{
Author: c,
AccessRequest: defs.PathAccessRequest{
Name: pathName,
@@ -183,13 +183,15 @@ func (c *conn) runRead() error {
c.query = c.rconn.URL.RawQuery
c.mutex.Unlock()
err = rtmp.FromStream(stream, c, c.rconn, c.nconn, time.Duration(c.writeTimeout))
r := &stream.Reader{Parent: c}
err = rtmp.FromStream(strm.Desc, r, c.rconn, c.nconn, time.Duration(c.writeTimeout))
if err != nil {
return err
}
c.Log(logger.Info, "is reading from path '%s', %s",
path.Name(), defs.FormatsInfo(stream.ReaderFormats(c)))
path.Name(), defs.FormatsInfo(r.Formats()))
onUnreadHook := hooks.OnRead(hooks.OnReadParams{
Logger: c,
@@ -204,14 +206,14 @@ func (c *conn) runRead() error {
// disable read deadline
c.nconn.SetReadDeadline(time.Time{})
stream.StartReader(c)
defer stream.RemoveReader(c)
strm.AddReader(r)
defer strm.RemoveReader(r)
select {
case <-c.ctx.Done():
return fmt.Errorf("terminated")
case err = <-stream.ReaderError(c):
case err = <-r.Error():
return err
}
}
internal/servers/rtmp/server_test.go
+5 -6
View File
@@ -145,10 +145,9 @@ func TestServerPublish(t *testing.T) {
recv := make(chan struct{})
reader := test.NilLogger
r := &stream.Reader{Parent: test.NilLogger}
strm.AddReader(
reader,
r.OnData(
strm.Desc.Medias[0],
strm.Desc.Medias[0].Formats[0],
func(u unit.Unit) error {
@@ -161,8 +160,8 @@ func TestServerPublish(t *testing.T) {
return nil
})
strm.StartReader(reader)
defer strm.RemoveReader(reader)
strm.AddReader(r)
defer strm.RemoveReader(r)
err = w.WriteH264(
test.FormatH264,
@@ -259,7 +258,7 @@ func TestServerRead(t *testing.T) {
defer conn.Close()
go func() {
strm.WaitRunningReader()
time.Sleep(500 * time.Millisecond)
strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
Base: unit.Base{
internal/servers/rtsp/server_test.go
+4 -5
View File
@@ -133,12 +133,11 @@ func TestServerPublish(t *testing.T) {
<-streamCreated
reader := test.NilLogger
r := &stream.Reader{Parent: test.NilLogger}
recv := make(chan struct{})
strm.AddReader(
reader,
r.OnData(
strm.Desc.Medias[0],
strm.Desc.Medias[0].Formats[0],
func(u unit.Unit) error {
@@ -151,8 +150,8 @@ func TestServerPublish(t *testing.T) {
return nil
})
strm.StartReader(reader)
defer strm.RemoveReader(reader)
strm.AddReader(r)
defer strm.RemoveReader(r)
err = source.WritePacketRTP(media0, &rtp.Packet{
Header: rtp.Header{
internal/servers/srt/conn.go
+8 -6
View File
@@ -255,7 +255,7 @@ func (c *conn) runPublishReader(sconn srt.Conn, streamID *streamID, pathConf *co
}
func (c *conn) runRead(streamID *streamID) error {
path, stream, err := c.pathManager.AddReader(defs.PathAddReaderReq{
path, strm, err := c.pathManager.AddReader(defs.PathAddReaderReq{
Author: c,
AccessRequest: defs.PathAccessRequest{
Name: streamID.path,
@@ -297,7 +297,9 @@ func (c *conn) runRead(streamID *streamID) error {
bw := bufio.NewWriterSize(sconn, srtMaxPayloadSize(c.udpMaxPayloadSize))
err = mpegts.FromStream(stream, c, bw, sconn, time.Duration(c.writeTimeout))
r := &stream.Reader{Parent: c}
err = mpegts.FromStream(strm.Desc, r, bw, sconn, time.Duration(c.writeTimeout))
if err != nil {
return err
}
@@ -310,7 +312,7 @@ func (c *conn) runRead(streamID *streamID) error {
c.mutex.Unlock()
c.Log(logger.Info, "is reading from path '%s', %s",
path.Name(), defs.FormatsInfo(stream.ReaderFormats(c)))
path.Name(), defs.FormatsInfo(r.Formats()))
onUnreadHook := hooks.OnRead(hooks.OnReadParams{
Logger: c,
@@ -325,14 +327,14 @@ func (c *conn) runRead(streamID *streamID) error {
// disable read deadline
sconn.SetReadDeadline(time.Time{})
stream.StartReader(c)
defer stream.RemoveReader(c)
strm.AddReader(r)
defer strm.RemoveReader(r)
select {
case <-c.ctx.Done():
return fmt.Errorf("terminated")
case err = <-stream.ReaderError(c):
case err = <-r.Error():
return err
}
}
internal/servers/srt/server_test.go
+5 -6
View File
@@ -125,12 +125,11 @@ func TestServerPublish(t *testing.T) {
<-streamCreated
reader := test.NilLogger
r := &stream.Reader{Parent: test.NilLogger}
recv := make(chan struct{})
strm.AddReader(
reader,
r.OnData(
strm.Desc.Medias[0],
strm.Desc.Medias[0].Formats[0],
func(u unit.Unit) error {
@@ -143,8 +142,8 @@ func TestServerPublish(t *testing.T) {
return nil
})
strm.StartReader(reader)
defer strm.RemoveReader(reader)
strm.AddReader(r)
defer strm.RemoveReader(r)
err = w.WriteH264(track, 0, 0, [][]byte{
{5, 2},
@@ -214,7 +213,7 @@ func TestServerRead(t *testing.T) {
require.NoError(t, err)
defer reader.Close()
strm.WaitRunningReader()
time.Sleep(500 * time.Millisecond)
strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
Base: unit.Base{
internal/servers/webrtc/server_test.go
+5 -6
View File
@@ -288,12 +288,11 @@ func TestServerPublish(t *testing.T) {
<-streamCreated
reader := test.NilLogger
r := &stream.Reader{Parent: test.NilLogger}
recv := make(chan struct{})
strm.AddReader(
reader,
r.OnData(
strm.Desc.Medias[0],
strm.Desc.Medias[0].Formats[0],
func(u unit.Unit) error {
@@ -311,8 +310,8 @@ func TestServerPublish(t *testing.T) {
return nil
})
strm.StartReader(reader)
defer strm.RemoveReader(reader)
strm.AddReader(r)
defer strm.RemoveReader(r)
err = track.WriteRTP(&rtp.Packet{
Header: rtp.Header{
@@ -547,7 +546,7 @@ func TestServerRead(t *testing.T) {
go func() {
defer close(writerDone)
strm.WaitRunningReader()
time.Sleep(500 * time.Millisecond)
r := reflect.New(reflect.TypeOf(ca.unit).Elem())
r.Elem().Set(reflect.ValueOf(ca.unit).Elem())
internal/servers/webrtc/session.go
+8 -9
View File
@@ -281,7 +281,7 @@ func (s *session) runRead() (int, error) {
IP: net.ParseIP(ip),
}
path, stream, err := s.pathManager.AddReader(defs.PathAddReaderReq{
path, strm, err := s.pathManager.AddReader(defs.PathAddReaderReq{
Author: s,
AccessRequest: req,
})
@@ -316,14 +316,15 @@ func (s *session) runRead() (int, error) {
Log: s,
}
err = webrtc.FromStream(stream, s, pc)
r := &stream.Reader{Parent: s}
err = webrtc.FromStream(strm.Desc, r, pc)
if err != nil {
return http.StatusBadRequest, err
}
err = pc.Start()
if err != nil {
stream.RemoveReader(s)
return http.StatusBadRequest, err
}
@@ -346,7 +347,6 @@ func (s *session) runRead() (int, error) {
answer, err := pc.CreateFullAnswer(offer)
if err != nil {
stream.RemoveReader(s)
return http.StatusBadRequest, err
}
@@ -356,7 +356,6 @@ func (s *session) runRead() (int, error) {
err = pc.WaitUntilConnected()
if err != nil {
stream.RemoveReader(s)
return 0, err
}
@@ -365,7 +364,7 @@ func (s *session) runRead() (int, error) {
s.mutex.Unlock()
s.Log(logger.Info, "is reading from path '%s', %s",
path.Name(), defs.FormatsInfo(stream.ReaderFormats(s)))
path.Name(), defs.FormatsInfo(r.Formats()))
onUnreadHook := hooks.OnRead(hooks.OnReadParams{
Logger: s,
@@ -377,14 +376,14 @@ func (s *session) runRead() (int, error) {
})
defer onUnreadHook()
stream.StartReader(s)
defer stream.RemoveReader(s)
strm.AddReader(r)
defer strm.RemoveReader(r)
select {
case <-pc.Failed():
return 0, fmt.Errorf("peer connection closed")
case err = <-stream.ReaderError(s):
case err = <-r.Error():
return 0, err
case <-s.ctx.Done():
internal/staticsources/rpicamera/source.go
+6 -5
View File
@@ -282,8 +282,9 @@ func (s *Source) runSecondary(params defs.StaticSourceRunParams) error {
return res.Err
}
origStream.AddReader(
s,
rdr := &stream.Reader{Parent: s}
rdr.OnData(
origStream.Desc.Medias[1],
origStream.Desc.Medias[1].Formats[0],
func(u unit.Unit) error {
@@ -299,11 +300,11 @@ func (s *Source) runSecondary(params defs.StaticSourceRunParams) error {
return nil
})
origStream.StartReader(s)
defer origStream.RemoveReader(s)
origStream.AddReader(rdr)
defer origStream.RemoveReader(rdr)
select {
case err = <-origStream.ReaderError(s):
case err = <-rdr.Error():
return err
case <-r.ctx.Done():
internal/stream/reader.go
+113
View File
@@ -0,0 +1,113 @@
package stream
import (
"fmt"
"github.com/bluenviron/gortsplib/v5/pkg/description"
"github.com/bluenviron/gortsplib/v5/pkg/format"
"github.com/bluenviron/gortsplib/v5/pkg/ringbuffer"
"github.com/bluenviron/mediamtx/internal/counterdumper"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/unit"
)
// OnDataFunc is the callback passed to OnData().
type OnDataFunc func(unit.Unit) error
// Reader is a stream reader.
type Reader struct {
SkipBytesSent bool
Parent logger.Writer
onDatas map[*description.Media]map[format.Format]OnDataFunc
queueSize int
buffer *ringbuffer.RingBuffer
discardedFrames *counterdumper.CounterDumper
// out
err chan error
}
// 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) {
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)
}
r.onDatas[medi][forma] = cb
}
// Formats returns all formats for which the reader has registered a OnData callback.
func (r *Reader) Formats() []format.Format {
var out []format.Format
for _, formats := range r.onDatas {
for forma := range formats {
out = append(out, forma)
}
}
return out
}
// error returns whenever there's an error.
// It can be called only after stream.AddReader().
func (r *Reader) Error() chan error {
return r.err
}
func (r *Reader) start() {
buffer, _ := ringbuffer.New(uint64(r.queueSize))
r.buffer = buffer
r.err = make(chan error)
r.discardedFrames = &counterdumper.CounterDumper{
OnReport: func(val uint64) {
r.Parent.Log(logger.Warn, "reader is too slow, discarding %d %s",
val,
func() string {
if val == 1 {
return "frame"
}
return "frames"
}())
},
}
r.discardedFrames.Start()
go r.run()
}
func (r *Reader) stop() {
r.buffer.Close()
r.discardedFrames.Stop()
<-r.err
}
func (r *Reader) run() {
r.err <- r.runInner()
close(r.err)
}
func (r *Reader) runInner() error {
for {
cb, ok := r.buffer.Pull()
if !ok {
return fmt.Errorf("terminated")
}
err := cb.(func() error)()
if err != nil {
return err
}
}
}
func (r *Reader) push(cb func() error) {
ok := r.buffer.Push(cb)
if !ok {
r.discardedFrames.Increase()
}
}
internal/stream/stream.go
+31 -95
View File
@@ -16,14 +16,6 @@ import (
"github.com/bluenviron/mediamtx/internal/unit"
)
// Reader is a stream reader.
type Reader interface {
logger.Writer
}
// ReadFunc is the callback passed to AddReader().
type ReadFunc func(unit.Unit) error
// Stream is a media stream.
// It stores tracks, readers and allows to write data to readers, converting it when needed.
type Stream struct {
@@ -35,23 +27,20 @@ type Stream struct {
bytesReceived *uint64
bytesSent *uint64
streamMedias map[*description.Media]*streamMedia
medias map[*description.Media]*streamMedia
mutex sync.RWMutex
rtspStream *gortsplib.ServerStream
rtspsStream *gortsplib.ServerStream
streamReaders map[Reader]*streamReader
readers map[*Reader]struct{}
processingErrors *counterdumper.CounterDumper
readerRunning chan struct{}
}
// Initialize initializes a Stream.
func (s *Stream) Initialize() error {
s.bytesReceived = new(uint64)
s.bytesSent = new(uint64)
s.streamMedias = make(map[*description.Media]*streamMedia)
s.streamReaders = make(map[Reader]*streamReader)
s.readerRunning = make(chan struct{})
s.medias = make(map[*description.Media]*streamMedia)
s.readers = make(map[*Reader]struct{})
s.processingErrors = &counterdumper.CounterDumper{
OnReport: func(val uint64) {
@@ -68,14 +57,14 @@ func (s *Stream) Initialize() error {
s.processingErrors.Start()
for _, media := range s.Desc.Medias {
s.streamMedias[media] = &streamMedia{
s.medias[media] = &streamMedia{
rtpMaxPayloadSize: s.RTPMaxPayloadSize,
media: media,
generateRTPPackets: s.GenerateRTPPackets,
processingErrors: s.processingErrors,
parent: s.Parent,
}
err := s.streamMedias[media].initialize()
err := s.medias[media].initialize()
if err != nil {
return err
}
@@ -103,10 +92,11 @@ func (s *Stream) BytesReceived() uint64 {
// BytesSent returns sent bytes.
func (s *Stream) BytesSent() uint64 {
bytesSent := atomic.LoadUint64(s.bytesSent)
s.mutex.RLock()
defer s.mutex.RUnlock()
bytesSent := atomic.LoadUint64(s.bytesSent)
if s.rtspStream != nil {
stats := s.rtspStream.Stats()
bytesSent += stats.BytesSent
@@ -115,6 +105,7 @@ func (s *Stream) BytesSent() uint64 {
stats := s.rtspsStream.Stats()
bytesSent += stats.BytesSent
}
return bytesSent
}
@@ -156,103 +147,48 @@ func (s *Stream) RTSPSStream(server *gortsplib.Server) *gortsplib.ServerStream {
// AddReader adds a reader.
// Used by all protocols except RTSP.
func (s *Stream) AddReader(reader Reader, medi *description.Media, forma format.Format, cb ReadFunc) {
func (s *Stream) AddReader(r *Reader) {
s.mutex.Lock()
defer s.mutex.Unlock()
sr, ok := s.streamReaders[reader]
if !ok {
sr = &streamReader{
queueSize: s.WriteQueueSize,
parent: reader,
}
sr.initialize()
s.readers[r] = struct{}{}
s.streamReaders[reader] = sr
for medi, formats := range r.onDatas {
sm := s.medias[medi]
for forma, onData := range formats {
sf := sm.formats[forma]
sf.onDatas[r] = onData
}
}
sm := s.streamMedias[medi]
sf := sm.formats[forma]
sf.addReader(sr, cb)
r.queueSize = s.WriteQueueSize
r.start()
}
// RemoveReader removes a reader.
// Used by all protocols except RTSP.
func (s *Stream) RemoveReader(reader Reader) {
func (s *Stream) RemoveReader(r *Reader) {
s.mutex.Lock()
defer s.mutex.Unlock()
sr := s.streamReaders[reader]
r.stop()
for _, sm := range s.streamMedias {
for _, sf := range sm.formats {
sf.removeReader(sr)
for medi, formats := range r.onDatas {
sm := s.medias[medi]
for forma := range formats {
sf := sm.formats[forma]
delete(sf.onDatas, r)
}
}
delete(s.streamReaders, reader)
sr.stop()
}
// StartReader starts a reader.
// Used by all protocols except RTSP.
func (s *Stream) StartReader(reader Reader) {
s.mutex.Lock()
defer s.mutex.Unlock()
sr := s.streamReaders[reader]
sr.start()
for _, sm := range s.streamMedias {
for _, sf := range sm.formats {
sf.startReader(sr)
}
}
select {
case <-s.readerRunning:
default:
close(s.readerRunning)
}
}
// ReaderError returns whenever there's an error.
func (s *Stream) ReaderError(reader Reader) chan error {
sr := s.streamReaders[reader]
return sr.error()
}
// ReaderFormats returns all formats that a reader is reading.
func (s *Stream) ReaderFormats(reader Reader) []format.Format {
s.mutex.RLock()
defer s.mutex.RUnlock()
sr := s.streamReaders[reader]
var formats []format.Format
for _, sm := range s.streamMedias {
for forma, sf := range sm.formats {
if _, ok := sf.pausedReaders[sr]; ok {
formats = append(formats, forma)
} else if _, ok = sf.runningReaders[sr]; ok {
formats = append(formats, forma)
}
}
}
return formats
}
// WaitRunningReader waits for a running reader.
func (s *Stream) WaitRunningReader() {
<-s.readerRunning
delete(s.readers, r)
}
// WriteUnit writes a Unit.
func (s *Stream) WriteUnit(medi *description.Media, forma format.Format, u unit.Unit) {
sm := s.streamMedias[medi]
sm := s.medias[medi]
sf := sm.formats[forma]
s.mutex.RLock()
@@ -269,7 +205,7 @@ func (s *Stream) WriteRTPPacket(
ntp time.Time,
pts int64,
) {
sm := s.streamMedias[medi]
sm := s.medias[medi]
sf := sm.formats[forma]
s.mutex.RLock()
internal/stream/stream_format.go
+11 -26
View File
@@ -29,14 +29,12 @@ type streamFormat struct {
processingErrors *counterdumper.CounterDumper
parent logger.Writer
proc codecprocessor.Processor
pausedReaders map[*streamReader]ReadFunc
runningReaders map[*streamReader]ReadFunc
proc codecprocessor.Processor
onDatas map[*Reader]OnDataFunc
}
func (sf *streamFormat) initialize() error {
sf.pausedReaders = make(map[*streamReader]ReadFunc)
sf.runningReaders = make(map[*streamReader]ReadFunc)
sf.onDatas = make(map[*Reader]OnDataFunc)
var err error
sf.proc, err = codecprocessor.New(sf.rtpMaxPayloadSize, sf.format, sf.generateRTPPackets, sf.parent)
@@ -47,22 +45,6 @@ func (sf *streamFormat) initialize() error {
return nil
}
func (sf *streamFormat) addReader(sr *streamReader, cb ReadFunc) {
sf.pausedReaders[sr] = cb
}
func (sf *streamFormat) removeReader(sr *streamReader) {
delete(sf.pausedReaders, sr)
delete(sf.runningReaders, sr)
}
func (sf *streamFormat) startReader(sr *streamReader) {
if cb, ok := sf.pausedReaders[sr]; ok {
delete(sf.pausedReaders, sr)
sf.runningReaders[sr] = cb
}
}
func (sf *streamFormat) writeUnit(s *Stream, medi *description.Media, u unit.Unit) {
err := sf.proc.ProcessUnit(u)
if err != nil {
@@ -80,7 +62,7 @@ func (sf *streamFormat) writeRTPPacket(
ntp time.Time,
pts int64,
) {
hasNonRTSPReaders := len(sf.pausedReaders) > 0 || len(sf.runningReaders) > 0
hasNonRTSPReaders := len(sf.onDatas) > 0
u, err := sf.proc.ProcessRTPPacket(pkt, ntp, pts, hasNonRTSPReaders)
if err != nil {
@@ -108,11 +90,14 @@ func (sf *streamFormat) writeUnitInner(s *Stream, medi *description.Media, u uni
}
}
for sr, cb := range sf.runningReaders {
ccb := cb
for sr, onData := range sf.onDatas {
csr := sr
cOnData := onData
sr.push(func() error {
atomic.AddUint64(s.bytesSent, size)
return ccb(u)
if !csr.SkipBytesSent {
atomic.AddUint64(s.bytesSent, size)
}
return cOnData(u)
})
}
}
internal/stream/stream_reader.go
-86
View File
@@ -1,86 +0,0 @@
package stream
import (
"fmt"
"github.com/bluenviron/gortsplib/v5/pkg/ringbuffer"
"github.com/bluenviron/mediamtx/internal/counterdumper"
"github.com/bluenviron/mediamtx/internal/logger"
)
type streamReader struct {
queueSize int
parent logger.Writer
buffer *ringbuffer.RingBuffer
started bool
discardedFrames *counterdumper.CounterDumper
// out
err chan error
}
func (w *streamReader) initialize() {
buffer, _ := ringbuffer.New(uint64(w.queueSize))
w.buffer = buffer
w.err = make(chan error)
}
func (w *streamReader) start() {
w.started = true
w.discardedFrames = &counterdumper.CounterDumper{
OnReport: func(val uint64) {
w.parent.Log(logger.Warn, "reader is too slow, discarding %d %s",
val,
func() string {
if val == 1 {
return "frame"
}
return "frames"
}())
},
}
w.discardedFrames.Start()
go w.run()
}
func (w *streamReader) stop() {
w.buffer.Close()
if w.started {
w.discardedFrames.Stop()
<-w.err
}
}
func (w *streamReader) error() chan error {
return w.err
}
func (w *streamReader) run() {
w.err <- w.runInner()
close(w.err)
}
func (w *streamReader) runInner() error {
for {
cb, ok := w.buffer.Pull()
if !ok {
return fmt.Errorf("terminated")
}
err := cb.(func() error)()
if err != nil {
return err
}
}
}
func (w *streamReader) push(cb func() error) {
ok := w.buffer.Push(cb)
if !ok {
w.discardedFrames.Increase()
}
}
internal/stream/stream_test.go
+110
View File
@@ -0,0 +1,110 @@
package stream
import (
"testing"
"github.com/bluenviron/gortsplib/v5/pkg/description"
"github.com/bluenviron/gortsplib/v5/pkg/format"
"github.com/bluenviron/mediamtx/internal/unit"
"github.com/stretchr/testify/require"
)
func TestStream(t *testing.T) {
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
},
}}
strm := &Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: desc,
GenerateRTPPackets: true,
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
r := &Reader{}
recv := make(chan struct{})
r.OnData(desc.Medias[0], desc.Medias[0].Formats[0], func(_ unit.Unit) error {
close(recv)
return nil
})
strm.AddReader(r)
defer strm.RemoveReader(r)
strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
Base: unit.Base{
PTS: 30000 * 2,
},
AU: [][]byte{
{5, 2}, // IDR
},
})
<-recv
require.Equal(t, uint64(14), strm.BytesReceived())
require.Equal(t, uint64(14), strm.BytesSent())
}
func TestStreamSkipBytesSent(t *testing.T) {
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
},
}}
strm := &Stream{
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Desc: desc,
GenerateRTPPackets: true,
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
r := &Reader{
SkipBytesSent: true,
}
recv := make(chan struct{})
r.OnData(desc.Medias[0], desc.Medias[0].Formats[0], func(_ unit.Unit) error {
close(recv)
return nil
})
strm.AddReader(r)
defer strm.RemoveReader(r)
strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
Base: unit.Base{
PTS: 30000 * 2,
},
AU: [][]byte{
{5, 2}, // IDR
},
})
<-recv
require.Equal(t, uint64(14), strm.BytesReceived())
require.Equal(t, uint64(0), strm.BytesSent())
}
internal/test/static_source_parent.go
+11 -8
View File
@@ -10,7 +10,7 @@ import (
// StaticSourceParent is a dummy static source parent.
type StaticSourceParent struct {
stream *stream.Stream
reader stream.Reader
reader *stream.Reader
Unit chan unit.Unit
}
@@ -41,15 +41,18 @@ func (p *StaticSourceParent) SetReady(req defs.PathSourceStaticSetReadyReq) defs
panic(err)
}
p.reader = NilLogger
p.reader = &stream.Reader{Parent: NilLogger}
p.stream.AddReader(p.reader, req.Desc.Medias[0], req.Desc.Medias[0].Formats[0], func(u unit.Unit) error {
p.Unit <- u
close(p.Unit)
return nil
})
p.reader.OnData(
req.Desc.Medias[0],
req.Desc.Medias[0].Formats[0],
func(u unit.Unit) error {
p.Unit <- u
close(p.Unit)
return nil
})
p.stream.StartReader(p.reader)
p.stream.AddReader(p.reader)
return defs.PathSourceStaticSetReadyRes{Stream: p.stream}
}