OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
mediamtx/internal/stream/stream_test.go
T
bluenviron-botGitHubaler9
 3e09466c18 bump golangci-lint to 2.13.1 (#6129) 
* bump golangci-lint to 2.13.1

* additional changes

---------

Co-authored-by: aler9 <46489434+aler9@users.noreply.github.com>
2026-08-22 16:57:28 +00:00

907 lines
21 KiB
Go
Raw Permalink Blame History

package stream
import (
"os"
"sync"
"sync/atomic"
"testing"
"github.com/bluenviron/gortsplib/v5/pkg/description"
"github.com/bluenviron/gortsplib/v5/pkg/format"
"github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg4audio"
"github.com/bluenviron/mediacommon/v2/pkg/formats/mp4/codecs"
"github.com/bluenviron/mediacommon/v2/pkg/formats/pmp4"
"github.com/pion/rtp"
"github.com/stretchr/testify/require"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/unit"
)
type nilLogger struct{}
func (nilLogger) Log(logger.Level, string, ...any) {
}
func TestStream(t *testing.T) {
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{PacketizationMode: 1}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
},
}}
strm := &Stream{
OrigDesc: desc,
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
subStream := &SubStream{
Stream: strm,
UseRTPPackets: false,
}
err = subStream.Initialize()
require.NoError(t, err)
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)
subStream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.Unit{
PTS: 30000 * 2,
Payload: unit.PayloadH264{
{5, 2}, // IDR
},
})
<-recv
require.Equal(t, uint64(14), strm.InboundBytes())
require.Equal(t, uint64(14), strm.OutboundBytes())
}
func TestStreamSkipOutboundBytes(t *testing.T) {
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{PacketizationMode: 1}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP8{}},
},
}}
strm := &Stream{
OrigDesc: desc,
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
subStream := &SubStream{
Stream: strm,
UseRTPPackets: false,
}
err = subStream.Initialize()
require.NoError(t, err)
r := &Reader{
SkipOutboundBytes: 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)
subStream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.Unit{
PTS: 30000 * 2,
Payload: unit.PayloadH264{
{5, 2}, // IDR
},
})
<-recv
require.Equal(t, uint64(14), strm.InboundBytes())
require.Equal(t, uint64(0), strm.OutboundBytes())
}
func TestStreamResizeOversizedRTPPackets(t *testing.T) {
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{
PacketizationMode: 1,
SPS: []byte{ // 1920x1080 baseline
0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20,
},
PPS: []byte{0x08, 0x06, 0x07, 0x08},
}},
},
}}
strm := &Stream{
OrigDesc: desc,
WriteQueueSize: 512,
RTPMaxPayloadSize: 400,
Parent: &nilLogger{},
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
subStream := &SubStream{
Stream: strm,
UseRTPPackets: true,
}
err = subStream.Initialize()
require.NoError(t, err)
r := &Reader{}
recv := make(chan *unit.Unit)
n := 0
r.OnData(desc.Medias[0], desc.Medias[0].Formats[0], func(u *unit.Unit) error {
switch n {
case 0:
case 1:
recv <- u
default:
t.Error("should not happen")
}
n++
return nil
})
strm.AddReader(r)
defer strm.RemoveReader(r)
subStream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.Unit{
PTS: 90000,
RTPPackets: []*rtp.Packet{
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 122,
Timestamp: 45343,
SSRC: 563423,
},
Payload: []byte{1, 2, 3, 4},
},
},
})
oversizedPayload := make([]byte, 1000)
for i := range oversizedPayload {
oversizedPayload[i] = byte(i % 256)
}
subStream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.Unit{
PTS: 90000,
RTPPackets: []*rtp.Packet{
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
},
Payload: oversizedPayload,
},
},
})
received := <-recv
require.Equal(t, 3, len(received.RTPPackets))
for i, pkt := range received.RTPPackets {
require.Equal(t, 123+uint16(i), pkt.SequenceNumber)
require.Equal(t, uint32(45343), pkt.Timestamp)
}
totalPayloadSize := 0
for _, pkt := range received.RTPPackets {
require.LessOrEqual(t, len(pkt.Payload), 400)
totalPayloadSize += len(pkt.Payload)
}
require.Equal(t, 1005, totalPayloadSize)
}
func TestStreamUpgradeH264PacketizationMode(t *testing.T) {
forma := &format.H264{
PacketizationMode: 0,
SPS: []byte{0x67, 0x42, 0xc0, 0x28},
PPS: []byte{0x08, 0x06},
}
strm := &Stream{
OrigDesc: &description.Session{Medias: []*description.Media{{Formats: []format.Format{forma}}}},
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Parent: &nilLogger{},
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
outDesc := strm.OutDescCopy()
require.Equal(t, &format.H264{
PacketizationMode: 1,
SPS: []byte{0x67, 0x42, 0xc0, 0x28},
PPS: []byte{0x08, 0x06},
}, outDesc.Medias[0].Formats[0])
}
func TestStreamAlwaysAvailableErrors(t *testing.T) {
for _, ca := range []struct {
name string
tracks []conf.AlwaysAvailableTrack
desc *description.Session
err string
}{
{
"wrong tracks",
[]conf.AlwaysAvailableTrack{
{Codec: "H264"},
{Codec: "H265"},
},
&description.Session{
Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{}},
},
},
},
"wants to publish [H264], but stream expects [H264 H265]",
},
{
"wrong mpeg-4 audio config",
[]conf.AlwaysAvailableTrack{
{Codec: "MPEG4Audio", SampleRate: 44100, ChannelCount: 2},
},
&description.Session{
Medias: []*description.Media{
{
Type: description.MediaTypeAudio,
Formats: []format.Format{&format.MPEG4Audio{
Config: &mpeg4audio.AudioSpecificConfig{
Type: 2,
SampleRate: 48000,
ChannelConfig: 1,
},
}},
},
},
},
"MPEG-4 audio configuration does not match, is type=2, sampleRate=48000, " +
"channelCount=1, but stream expects type=2, sampleRate=44100, channelCount=2",
},
{
"wrong g711 config",
[]conf.AlwaysAvailableTrack{
{Codec: "G711", MULaw: true, SampleRate: 8000, ChannelCount: 2},
},
&description.Session{
Medias: []*description.Media{
{
Type: description.MediaTypeAudio,
Formats: []format.Format{&format.G711{
MULaw: false,
SampleRate: 8000,
ChannelCount: 2,
}},
},
},
},
"G711 configuration does not match, is MULaw=false, sampleRate=8000, " +
"channelCount=2, but stream expects MULaw=true, sampleRate=8000, channelCount=2",
},
{
"wrong lpcm config",
[]conf.AlwaysAvailableTrack{
{Codec: "LPCM", SampleRate: 44100, ChannelCount: 2},
},
&description.Session{
Medias: []*description.Media{
{
Type: description.MediaTypeAudio,
Formats: []format.Format{&format.LPCM{
BitDepth: 16,
SampleRate: 48000,
ChannelCount: 2,
}},
},
},
},
"LPCM configuration does not match, is bitDepth=16, sampleRate=48000, " +
"channelCount=2, but stream expects bitDepth=16, sampleRate=44100, channelCount=2",
},
} {
t.Run(ca.name, func(t *testing.T) {
strm := &Stream{
AlwaysAvailable: true,
AlwaysAvailableTracks: ca.tracks,
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
ReplaceNTP: true,
Parent: &nilLogger{},
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
subStream := &SubStream{
Stream: strm,
InDesc: ca.desc,
UseRTPPackets: false,
}
err = subStream.Initialize()
require.EqualError(t, err, ca.err)
})
}
}
func TestStreamAlwaysAvailable(t *testing.T) {
for _, ca := range []string{"default", "file"} {
t.Run(ca, func(t *testing.T) {
strm := &Stream{
AlwaysAvailable: true,
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
ReplaceNTP: true,
Parent: &nilLogger{},
}
if ca == "default" {
strm.AlwaysAvailableTracks = []conf.AlwaysAvailableTrack{
{Codec: conf.CodecAV1},
{Codec: conf.CodecVP9},
{Codec: conf.CodecH265},
{Codec: conf.CodecH264},
{Codec: conf.CodecOpus},
{Codec: conf.CodecMPEG4Audio, SampleRate: 44100, ChannelCount: 2},
{Codec: conf.CodecLPCM, SampleRate: 48000, ChannelCount: 2},
}
} else {
tmpf, err := os.CreateTemp(t.TempDir(), "rtsp-")
require.NoError(t, err)
pmp4 := &pmp4.Presentation{
Tracks: []*pmp4.Track{
{
ID: 1,
TimeScale: 90000,
Codec: &codecs.AV1{
SequenceHeader: []byte{8, 0, 0, 0, 66, 167, 191, 228, 96, 13, 0, 64},
},
Samples: []*pmp4.Sample{
{
Duration: 90000,
PayloadSize: 13,
GetPayload: func() ([]byte, error) {
return []byte{0xa, 0xb, 0x0, 0x0, 0x0, 0x42, 0xa7, 0xbf, 0xe4, 0x60, 0xd, 0x0, 0x40}, nil
},
},
},
},
{
ID: 2,
TimeScale: 90000,
Codec: &codecs.VP9{
Width: 1280,
Height: 720,
Profile: 1,
BitDepth: 8,
ChromaSubsampling: 1,
ColorRange: false,
},
Samples: []*pmp4.Sample{
{
Duration: 90000,
PayloadSize: 4,
GetPayload: func() ([]byte, error) {
return []byte{1, 2, 3, 4}, nil
},
},
},
},
{
ID: 3,
TimeScale: 90000,
Codec: &codecs.H265{
VPS: []byte{
0x40, 0x01, 0x0c, 0x01, 0xff, 0xff, 0x02, 0x20,
0x00, 0x00, 0x03, 0x00, 0xb0, 0x00, 0x00, 0x03,
0x00, 0x00, 0x03, 0x00, 0x7b, 0x18, 0xb0, 0x24,
},
SPS: []byte{
0x42, 0x01, 0x01, 0x02, 0x20, 0x00, 0x00, 0x03,
0x00, 0xb0, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03,
0x00, 0x7b, 0xa0, 0x07, 0x82, 0x00, 0x88, 0x7d,
0xb6, 0x71, 0x8b, 0x92, 0x44, 0x80, 0x53, 0x88,
0x88, 0x92, 0xcf, 0x24, 0xa6, 0x92, 0x72, 0xc9,
0x12, 0x49, 0x22, 0xdc, 0x91, 0xaa, 0x48, 0xfc,
0xa2, 0x23, 0xff, 0x00, 0x01, 0x00, 0x01, 0x6a,
0x02, 0x02, 0x02, 0x01,
},
PPS: []byte{
0x44, 0x01, 0xc0, 0x25, 0x2f, 0x05, 0x32, 0x40,
},
},
Samples: []*pmp4.Sample{
{
Duration: 90000,
PayloadSize: 8,
GetPayload: func() ([]byte, error) {
return []byte{0x0, 0x0, 0x0, 0x4, 0x1, 0x2, 0x3, 0x4}, nil
},
},
},
},
{
ID: 4,
TimeScale: 90000,
Codec: &codecs.H264{
SPS: []byte{ // 1920x1080 baseline
0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20,
},
PPS: []byte{0x08, 0x06, 0x07, 0x08},
},
Samples: []*pmp4.Sample{
{
Duration: 90000,
PayloadSize: 8,
GetPayload: func() ([]byte, error) {
return []byte{0x0, 0x0, 0x0, 0x4, 0x1, 0x2, 0x3, 0x4}, nil
},
},
},
},
{
ID: 5,
TimeScale: 48000,
Codec: &codecs.Opus{
ChannelCount: 2,
},
Samples: []*pmp4.Sample{
{
Duration: 48000,
PayloadSize: 2,
GetPayload: func() ([]byte, error) {
return []byte{0x1, 0x2}, nil
},
},
},
},
{
ID: 6,
TimeScale: 44100,
Codec: &codecs.MPEG4Audio{
Config: mpeg4audio.AudioSpecificConfig{
Type: 2,
SampleRate: 44100,
ChannelConfig: 2,
ChannelCount: 2, //nolint:staticcheck
},
},
Samples: []*pmp4.Sample{
{
Duration: 44100,
PayloadSize: 4,
GetPayload: func() ([]byte, error) {
return []byte{0x12, 0x10, 0x0, 0x0}, nil
},
},
},
},
{
ID: 7,
TimeScale: 48000,
Codec: &codecs.LPCM{
BitDepth: 16,
SampleRate: 48000,
ChannelCount: 2,
},
Samples: []*pmp4.Sample{
{
Duration: 48000,
PayloadSize: 4,
GetPayload: func() ([]byte, error) {
return []byte{0x12, 0x10, 0x0, 0x0}, nil
},
},
},
},
},
}
err = pmp4.Marshal(tmpf)
require.NoError(t, err)
tmpf.Close()
strm.AlwaysAvailableFile = tmpf.Name()
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
r := &Reader{
Parent: &nilLogger{},
}
var wg sync.WaitGroup
n := 0
var phase2 atomic.Bool
wg.Add(1)
var lastPTSAV1 int64
var soAV1a sync.Once
var soAV1b sync.Once
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
if !phase2.Load() {
soAV1a.Do(func() {
require.NotEmpty(t, u.Payload)
wg.Done()
})
} else {
soAV1b.Do(func() {
require.Equal(t, unit.PayloadAV1{{1, 2, 3, 4}}, u.Payload)
wg.Done()
})
}
return nil
})
n++
wg.Add(1)
var lastPTSVP9 int64
var soVP9a sync.Once
var soVP9b sync.Once
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
if !phase2.Load() {
soVP9a.Do(func() {
require.NotEmpty(t, u.Payload)
wg.Done()
})
} else {
soVP9b.Do(func() {
require.Equal(t, unit.PayloadVP9{1, 2, 3, 4}, u.Payload)
wg.Done()
})
}
return nil
})
n++
wg.Add(1)
var lastPTSH265 int64
var soH265a sync.Once
var soH265b sync.Once
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
if !phase2.Load() {
soH265a.Do(func() {
require.NotEmpty(t, u.Payload)
wg.Done()
})
} else {
soH265b.Do(func() {
require.Equal(t, unit.PayloadH265{{1, 2, 3, 4}}, u.Payload)
wg.Done()
})
}
return nil
})
n++
wg.Add(1)
var lastPTSH264 int64
var soH264a sync.Once
var soH264b sync.Once
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
if !phase2.Load() {
soH264a.Do(func() {
require.NotEmpty(t, u.Payload)
wg.Done()
})
} else {
soH264b.Do(func() {
require.Equal(t, unit.PayloadH264{{1, 2, 3, 4}}, u.Payload)
wg.Done()
})
}
return nil
})
n++
wg.Add(1)
var lastPTSOpus int64
var soOpusa sync.Once
var soOpusb sync.Once
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
if !phase2.Load() {
soOpusa.Do(func() {
require.NotEmpty(t, u.Payload)
wg.Done()
})
} else {
soOpusb.Do(func() {
require.Equal(t, unit.PayloadOpus{{1, 2}}, u.Payload)
wg.Done()
})
}
return nil
})
n++
wg.Add(1)
var lastPTSMPEG4Audio int64
var soMPEG4Audioa sync.Once
var soMPEG4Audiob sync.Once
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
if !phase2.Load() {
soMPEG4Audioa.Do(func() {
require.NotEmpty(t, u.Payload)
wg.Done()
})
} else {
soMPEG4Audiob.Do(func() {
require.Equal(t, unit.PayloadMPEG4Audio{{1, 2, 3, 4}}, u.Payload)
wg.Done()
})
}
return nil
})
n++
wg.Add(1)
var lastPTSLPCM int64
var soLPCMa sync.Once
var soLPCMb sync.Once
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
if !phase2.Load() {
soLPCMa.Do(func() {
require.NotEmpty(t, u.Payload)
wg.Done()
})
} else {
soLPCMb.Do(func() {
require.Equal(t, unit.PayloadLPCM{1, 2, 3, 4}, u.Payload)
wg.Done()
})
}
return nil
})
strm.AddReader(r)
defer strm.RemoveReader(r)
wg.Wait()
subStream := &SubStream{
Stream: strm,
InDesc: &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.AV1{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.VP9{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H265{}},
},
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{}},
},
{
Type: description.MediaTypeAudio,
Formats: []format.Format{&format.Opus{}},
},
{
Type: description.MediaTypeAudio,
Formats: []format.Format{&format.MPEG4Audio{
Config: &mpeg4audio.AudioSpecificConfig{
Type: 2,
SampleRate: 44100,
ChannelConfig: 2,
ChannelCount: 2, //nolint:staticcheck
},
}},
},
{
Type: description.MediaTypeAudio,
Formats: []format.Format{&format.LPCM{
BitDepth: 16,
SampleRate: 48000,
ChannelCount: 2,
}},
},
}},
UseRTPPackets: false,
}
err = subStream.Initialize()
require.NoError(t, err)
wg.Add(7)
phase2.Store(true)
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.InDesc.Medias[1], subStream.InDesc.Medias[1].Formats[0], &unit.Unit{
PTS: 0,
Payload: unit.PayloadVP9{1, 2, 3, 4},
})
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.InDesc.Medias[3], subStream.InDesc.Medias[3].Formats[0], &unit.Unit{
PTS: 0,
Payload: unit.PayloadH264{{1, 2, 3, 4}},
})
subStream.WriteUnit(subStream.InDesc.Medias[4], subStream.InDesc.Medias[4].Formats[0], &unit.Unit{
PTS: 0,
Payload: unit.PayloadOpus{{1, 2}},
})
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.InDesc.Medias[6], subStream.InDesc.Medias[6].Formats[0], &unit.Unit{
PTS: 0,
Payload: unit.PayloadLPCM{1, 2, 3, 4},
})
wg.Wait()
})
}
}
func TestStreamAlwaysAvailableConcurrentReaders(t *testing.T) {
strm := &Stream{
AlwaysAvailable: true,
AlwaysAvailableTracks: []conf.AlwaysAvailableTrack{
{Codec: conf.CodecH264},
},
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
ReplaceNTP: true,
Parent: &nilLogger{},
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
origMedia := strm.OrigDesc.Medias[0]
origFormat := origMedia.Formats[0]
start := make(chan struct{})
ready := make(chan struct{}, 1)
done := make(chan struct{})
var readersWG sync.WaitGroup
for range 4 {
readersWG.Go(func() {
<-start
for {
select {
case <-done:
return
default:
}
r := &Reader{Parent: &nilLogger{}}
r.OnData(origMedia, origFormat, func(_ *unit.Unit) error {
return nil
})
strm.AddReader(r)
select {
case ready <- struct{}{}:
default:
}
strm.RemoveReader(r)
}
})
}
close(start)
<-ready
for range 256 {
subStream := &SubStream{
Stream: strm,
InDesc: &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []format.Format{&format.H264{
PacketizationMode: 1,
SPS: offlineH264SPS,
PPS: offlineH264PPS,
}},
},
}},
UseRTPPackets: false,
}
err = subStream.Initialize()
require.NoError(t, err)
}
close(done)
readersWG.Wait()
}
Reference in New Issue View Git Blame Copy Permalink