zishuo/rtsp-simple-server
1
0
Fork 0
mirror of https://github.com/aler9/rtsp-simple-server synced 2025-09-26 19:51:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

record: fix loss of audio samples during segment switch (#4556)

Browse Source
This commit is contained in:
Alessandro Ros
2025-05-25 21:29:33 +02:00
committed by GitHub
parent 01e6fedc36
commit fff3b9b315
13 changed files with 273 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
internal/playback/segment_fmp4.go
View File

@@ -14,7 +14,7 @@ import (
const (
sampleFlagIsNonSyncSample = 1 << 16
concatenationTolerance = 500 * time.Millisecond
concatenationTolerance = 1 * time.Second
)
var errTerminated = errors.New("terminated")

4
internal/protocols/rtsp/to_stream.go
View File

@@ -32,7 +32,7 @@ func ToStream(
source rtspSource,
medias []*description.Media,
pathConf *conf.Path,
stream *stream.Stream,
strm *stream.Stream,
log logger.Writer,
) {
for _, medi := range medias {
@@ -82,7 +82,7 @@ func ToStream(
return
}
stream.WriteRTPPacket(cmedi, cforma, pkt, ntp, pts)
strm.WriteRTPPacket(cmedi, cforma, pkt, ntp, pts)
})
}
}

30
internal/recorder/format_fmp4.go
View File

@@ -136,7 +136,7 @@ func (f *formatFMP4) initialize() bool {
}
}
for _, media := range f.ri.rec.Stream.Desc.Medias {
for _, media := range f.ri.stream.Desc.Medias {
for _, forma := range media.Formats {
clockRate := forma.ClockRate()
@@ -149,7 +149,7 @@ func (f *formatFMP4) initialize() bool {
firstReceived := false
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -210,7 +210,7 @@ func (f *formatFMP4) initialize() bool {
firstReceived := false
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -295,7 +295,7 @@ func (f *formatFMP4) initialize() bool {
var dtsExtractor *h265.DTSExtractor
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -378,7 +378,7 @@ func (f *formatFMP4) initialize() bool {
var dtsExtractor *h264.DTSExtractor
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -453,7 +453,7 @@ func (f *formatFMP4) initialize() bool {
firstReceived := false
var lastPTS int64
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -506,7 +506,7 @@ func (f *formatFMP4) initialize() bool {
firstReceived := false
var lastPTS int64
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -559,7 +559,7 @@ func (f *formatFMP4) initialize() bool {
parsed := false
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -595,7 +595,7 @@ func (f *formatFMP4) initialize() bool {
}
track := addTrack(forma, codec)
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -633,7 +633,7 @@ func (f *formatFMP4) initialize() bool {
}
track := addTrack(forma, codec)
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -671,7 +671,7 @@ func (f *formatFMP4) initialize() bool {
parsed := false
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -730,7 +730,7 @@ func (f *formatFMP4) initialize() bool {
parsed := false
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -795,7 +795,7 @@ func (f *formatFMP4) initialize() bool {
}
track := addTrack(forma, codec)
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -835,7 +835,7 @@ func (f *formatFMP4) initialize() bool {
}
track := addTrack(forma, codec)
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -863,7 +863,7 @@ func (f *formatFMP4) initialize() bool {
}
n := 1
for _, medi := range f.ri.rec.Stream.Desc.Medias {
for _, medi := range f.ri.stream.Desc.Medias {
for _, forma := range medi.Formats {
if _, ok := setuppedFormatsMap[forma]; !ok {
f.ri.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())

4
internal/recorder/format_fmp4_part.go
View File

@@ -55,7 +55,7 @@ func (p *formatFMP4Part) initialize() {
func (p *formatFMP4Part) close() error {
if p.s.fi == nil {
p.s.path = recordstore.Path{Start: p.s.startNTP}.Encode(p.s.f.ri.pathFormat)
p.s.path = recordstore.Path{Start: p.s.startNTP}.Encode(p.s.f.ri.pathFormat2)
p.s.f.ri.Log(logger.Debug, "creating segment %s", p.s.path)
err := os.MkdirAll(filepath.Dir(p.s.path), 0o755)
@@ -68,7 +68,7 @@ func (p *formatFMP4Part) close() error {
return err
}
p.s.f.ri.rec.OnSegmentCreate(p.s.path)
p.s.f.ri.onSegmentCreate(p.s.path)
err = writeInit(fi, p.s.f.tracks)
if err != nil {

6
internal/recorder/format_fmp4_segment.go
View File

@@ -136,7 +136,7 @@ func (s *formatFMP4Segment) close() error {
}
if err2 == nil {
s.f.ri.rec.OnSegmentComplete(s.path, duration)
s.f.ri.onSegmentComplete(s.path, duration)
}
}
@@ -150,11 +150,11 @@ func (s *formatFMP4Segment) write(track *formatFMP4Track, sample *sample, dtsDur
s.curPart = &formatFMP4Part{
s: s,
sequenceNumber: s.f.nextSequenceNumber,
startDTS: dtsDuration,
startDTS: s.startDTS,
}
s.curPart.initialize()
s.f.nextSequenceNumber++
} else if s.curPart.duration() >= s.f.ri.rec.PartDuration {
} else if s.curPart.duration() >= s.f.ri.partDuration {
err := s.curPart.close()
s.curPart = nil

44
internal/recorder/format_fmp4_track.go
View File

@@ -1,9 +1,34 @@
package recorder
import (
"time"
"github.com/bluenviron/mediacommon/v2/pkg/formats/fmp4"
"github.com/bluenviron/mediamtx/internal/logger"
)
const (
// this corresponds to concatenationTolerance
maxBasetime = 1 * time.Second
)
func findOldestNextSample(tracks []*formatFMP4Track) (*sample, time.Duration) {
var oldestSample *sample
var oldestDTS time.Duration
for _, track := range tracks {
if track.nextSample != nil {
normalizedDTS := timestampToDuration(track.nextSample.dts, int(track.initTrack.TimeScale))
if oldestSample == nil || normalizedDTS < oldestDTS {
oldestSample = track.nextSample
oldestDTS = normalizedDTS
}
}
}
return oldestSample, oldestDTS
}
type formatFMP4Track struct {
f *formatFMP4
initTrack *fmp4.InitTrack
@@ -32,8 +57,8 @@ func (t *formatFMP4Track) write(sample *sample) error {
startNTP: sample.ntp,
}
t.f.currentSegment.initialize()
// BaseTime is negative, this is not supported by fMP4. Reject the sample silently.
} else if (dtsDuration - t.f.currentSegment.startDTS) < 0 {
} else if (dtsDuration - t.f.currentSegment.startDTS) < 0 { // BaseTime is negative, this is not supported by fMP4
t.f.ri.Log(logger.Warn, "sample of track %d received too late, discarding", t.initTrack.ID)
return nil
}
@@ -46,17 +71,26 @@ func (t *formatFMP4Track) write(sample *sample) error {
if (!t.f.hasVideo || t.initTrack.Codec.IsVideo()) &&
!t.nextSample.IsNonSyncSample &&
(nextDTSDuration-t.f.currentSegment.startDTS) >= t.f.ri.rec.SegmentDuration {
(nextDTSDuration-t.f.currentSegment.startDTS) >= t.f.ri.segmentDuration {
t.f.currentSegment.lastDTS = nextDTSDuration
err := t.f.currentSegment.close()
if err != nil {
return err
}
// start next segment from the oldest next sample, in order to avoid the "negative basetime" issue
oldestSample, oldestDTS := findOldestNextSample(t.f.tracks)
// prevent going too back in time
if (nextDTSDuration - oldestDTS) > maxBasetime {
oldestSample = t.nextSample
oldestDTS = nextDTSDuration
}
t.f.currentSegment = &formatFMP4Segment{
f: t.f,
startDTS: nextDTSDuration,
startNTP: t.nextSample.ntp,
startDTS: oldestDTS,
startNTP: oldestSample.ntp,
}
t.f.currentSegment.initialize()
}

24
internal/recorder/format_mpegts.go
View File

@@ -77,7 +77,7 @@ func (f *formatMPEGTS) initialize() bool {
return track
}
for _, media := range f.ri.rec.Stream.Desc.Medias {
for _, media := range f.ri.stream.Desc.Medias {
for _, forma := range media.Formats {
clockRate := forma.ClockRate()
@@ -87,7 +87,7 @@ func (f *formatMPEGTS) initialize() bool {
var dtsExtractor *h265.DTSExtractor
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -132,7 +132,7 @@ func (f *formatMPEGTS) initialize() bool {
var dtsExtractor *h264.DTSExtractor
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -178,7 +178,7 @@ func (f *formatMPEGTS) initialize() bool {
firstReceived := false
var lastPTS int64
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -217,7 +217,7 @@ func (f *formatMPEGTS) initialize() bool {
firstReceived := false
var lastPTS int64
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -255,7 +255,7 @@ func (f *formatMPEGTS) initialize() bool {
ChannelCount: forma.ChannelCount,
})
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -288,7 +288,7 @@ func (f *formatMPEGTS) initialize() bool {
Config: *co,
})
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -316,7 +316,7 @@ func (f *formatMPEGTS) initialize() bool {
case *rtspformat.MPEG1Audio:
track := addTrack(forma, &mpegts.CodecMPEG1Audio{})
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -343,7 +343,7 @@ func (f *formatMPEGTS) initialize() bool {
case *rtspformat.AC3:
track := addTrack(forma, &mpegts.CodecAC3{})
f.ri.rec.Stream.AddReader(
f.ri.stream.AddReader(
f.ri,
media,
forma,
@@ -385,7 +385,7 @@ func (f *formatMPEGTS) initialize() bool {
}
n := 1
for _, medi := range f.ri.rec.Stream.Desc.Medias {
for _, medi := range f.ri.stream.Desc.Medias {
for _, forma := range medi.Formats {
if _, ok := setuppedFormatsMap[forma]; !ok {
f.ri.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec())
@@ -436,7 +436,7 @@ func (f *formatMPEGTS) write(
f.currentSegment.initialize()
case (!f.hasVideo || isVideo) &&
randomAccess &&
(dtsDuration-f.currentSegment.startDTS) >= f.ri.rec.SegmentDuration:
(dtsDuration-f.currentSegment.startDTS) >= f.ri.segmentDuration:
f.currentSegment.lastDTS = dtsDuration
err := f.currentSegment.close()
if err != nil {
@@ -450,7 +450,7 @@ func (f *formatMPEGTS) write(
}
f.currentSegment.initialize()
case (dtsDuration - f.currentSegment.lastFlush) >= f.ri.rec.PartDuration:
case (dtsDuration - f.currentSegment.lastFlush) >= f.ri.partDuration:
err := f.bw.Flush()
if err != nil {
return err

6
internal/recorder/format_mpegts_segment.go
View File

@@ -38,7 +38,7 @@ func (s *formatMPEGTSSegment) close() error {
if err2 == nil {
duration := s.lastDTS - s.startDTS
s.f.ri.rec.OnSegmentComplete(s.path, duration)
s.f.ri.onSegmentComplete(s.path, duration)
}
}
@@ -47,7 +47,7 @@ func (s *formatMPEGTSSegment) close() error {
func (s *formatMPEGTSSegment) Write(p []byte) (int, error) {
if s.fi == nil {
s.path = recordstore.Path{Start: s.startNTP}.Encode(s.f.ri.pathFormat)
s.path = recordstore.Path{Start: s.startNTP}.Encode(s.f.ri.pathFormat2)
s.f.ri.Log(logger.Debug, "creating segment %s", s.path)
err := os.MkdirAll(filepath.Dir(s.path), 0o755)
@@ -60,7 +60,7 @@ func (s *formatMPEGTSSegment) Write(p []byte) (int, error) {
return 0, err
}
s.f.ri.rec.OnSegmentCreate(s.path)
s.f.ri.onSegmentCreate(s.path)
s.fi = fi
}

20
internal/recorder/recorder.go
View File

@@ -53,7 +53,15 @@ func (r *Recorder) Initialize() {
r.done = make(chan struct{})
r.currentInstance = &recorderInstance{
rec: r,
pathFormat: r.PathFormat,
format: r.Format,
partDuration: r.PartDuration,
segmentDuration: r.SegmentDuration,
pathName: r.PathName,
stream: r.Stream,
onSegmentCreate: r.OnSegmentCreate,
onSegmentComplete: r.OnSegmentComplete,
parent: r,
}
r.currentInstance.initialize()
@@ -91,7 +99,15 @@ func (r *Recorder) run() {
}
r.currentInstance = &recorderInstance{
rec: r,
pathFormat: r.PathFormat,
format: r.Format,
partDuration: r.PartDuration,
segmentDuration: r.SegmentDuration,
pathName: r.PathName,
stream: r.Stream,
onSegmentCreate: r.OnSegmentCreate,
onSegmentComplete: r.OnSegmentComplete,
parent: r,
}
r.currentInstance.initialize()
}

45
internal/recorder/recorder_instance.go
View File

@@ -9,6 +9,7 @@ import (
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/recordstore"
"github.com/bluenviron/mediamtx/internal/stream"
)
type sample struct {
@@ -18,11 +19,19 @@ type sample struct {
}
type recorderInstance struct {
rec *Recorder
pathFormat string
format conf.RecordFormat
partDuration time.Duration
segmentDuration time.Duration
pathName string
stream *stream.Stream
onSegmentCreate OnSegmentCreateFunc
onSegmentComplete OnSegmentCompleteFunc
parent logger.Writer
pathFormat string
format format
skip bool
pathFormat2 string
format2 format
skip bool
terminate chan struct{}
done chan struct{}
@@ -30,38 +39,38 @@ type recorderInstance struct {
// Log implements logger.Writer.
func (ri *recorderInstance) Log(level logger.Level, format string, args ...interface{}) {
ri.rec.Log(level, format, args...)
ri.parent.Log(level, format, args...)
}
func (ri *recorderInstance) initialize() {
ri.pathFormat = ri.rec.PathFormat
ri.pathFormat2 = ri.pathFormat
ri.pathFormat = recordstore.PathAddExtension(
strings.ReplaceAll(ri.pathFormat, "%path", ri.rec.PathName),
ri.rec.Format,
ri.pathFormat2 = recordstore.PathAddExtension(
strings.ReplaceAll(ri.pathFormat2, "%path", ri.pathName),
ri.format,
)
ri.terminate = make(chan struct{})
ri.done = make(chan struct{})
switch ri.rec.Format {
switch ri.format {
case conf.RecordFormatMPEGTS:
ri.format = &formatMPEGTS{
ri.format2 = &formatMPEGTS{
ri: ri,
}
ok := ri.format.initialize()
ok := ri.format2.initialize()
ri.skip = !ok
default:
ri.format = &formatFMP4{
ri.format2 = &formatFMP4{
ri: ri,
}
ok := ri.format.initialize()
ok := ri.format2.initialize()
ri.skip = !ok
}
if !ri.skip {
ri.rec.Stream.StartReader(ri)
ri.stream.StartReader(ri)
}
go ri.run()
@@ -77,16 +86,16 @@ func (ri *recorderInstance) run() {
if !ri.skip {
select {
case err := <-ri.rec.Stream.ReaderError(ri):
case err := <-ri.stream.ReaderError(ri):
ri.Log(logger.Error, err.Error())
case <-ri.terminate:
}
ri.rec.Stream.RemoveReader(ri)
ri.stream.RemoveReader(ri)
} else {
<-ri.terminate
}
ri.format.close()
ri.format2.close()
}

131
internal/recorder/recorder_test.go
View File

@@ -71,12 +71,12 @@ func TestRecorder(t *testing.T) {
},
}}
writeToStream := func(stream *stream.Stream, startDTS int64, startNTP time.Time) {
writeToStream := func(strm *stream.Stream, startDTS int64, startNTP time.Time) {
for i := 0; i < 2; i++ {
pts := startDTS + int64(i)*100*90000/1000
ntp := startNTP.Add(time.Duration(i*60) * time.Second)
stream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
Base: unit.Base{
PTS: pts,
NTP: ntp,
@@ -88,7 +88,7 @@ func TestRecorder(t *testing.T) {
},
})
stream.WriteUnit(desc.Medias[1], desc.Medias[1].Formats[0], &unit.H265{
strm.WriteUnit(desc.Medias[1], desc.Medias[1].Formats[0], &unit.H265{
Base: unit.Base{
PTS: pts,
},
@@ -100,21 +100,21 @@ func TestRecorder(t *testing.T) {
},
})
stream.WriteUnit(desc.Medias[2], desc.Medias[2].Formats[0], &unit.MPEG4Audio{
strm.WriteUnit(desc.Medias[2], desc.Medias[2].Formats[0], &unit.MPEG4Audio{
Base: unit.Base{
PTS: pts * int64(desc.Medias[2].Formats[0].ClockRate()) / 90000,
},
AUs: [][]byte{{1, 2, 3, 4}},
})
stream.WriteUnit(desc.Medias[3], desc.Medias[3].Formats[0], &unit.G711{
strm.WriteUnit(desc.Medias[3], desc.Medias[3].Formats[0], &unit.G711{
Base: unit.Base{
PTS: pts * int64(desc.Medias[3].Formats[0].ClockRate()) / 90000,
},
Samples: []byte{1, 2, 3, 4},
})
stream.WriteUnit(desc.Medias[4], desc.Medias[4].Formats[0], &unit.LPCM{
strm.WriteUnit(desc.Medias[4], desc.Medias[4].Formats[0], &unit.LPCM{
Base: unit.Base{
PTS: pts * int64(desc.Medias[4].Formats[0].ClockRate()) / 90000,
},
@@ -410,7 +410,7 @@ func TestRecorderFMP4NegativeDTS(t *testing.T) {
}
}
require.Equal(t, true, found)
require.True(t, found)
}
func TestRecorderSkipTracksPartial(t *testing.T) {
@@ -538,3 +538,120 @@ func TestRecorderSkipTracksFull(t *testing.T) {
})
}
}
func TestRecorderFMP4SegmentSwitch(t *testing.T) {
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
Formats: []rtspformat.Format{test.FormatH264},
},
{
Type: description.MediaTypeAudio,
Formats: []rtspformat.Format{test.FormatMPEG4Audio},
},
}}
strm := &stream.Stream{
WriteQueueSize: 512,
UDPMaxPayloadSize: 1472,
Desc: desc,
GenerateRTPPackets: true,
Parent: test.NilLogger,
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
dir, err := os.MkdirTemp("", "mediamtx-agent")
require.NoError(t, err)
defer os.RemoveAll(dir)
n := 0
w := &Recorder{
PathFormat: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
Format: conf.RecordFormatFMP4,
PartDuration: 100 * time.Millisecond,
SegmentDuration: 1 * time.Second,
PathName: "mypath",
Stream: strm,
Parent: test.NilLogger,
OnSegmentCreate: func(segPath string) {
switch n {
case 0:
require.Equal(t, filepath.Join(dir, "mypath", "2008-05-20_22-15-25-000000.mp4"), segPath)
case 1:
require.Equal(t, filepath.Join(dir, "mypath", "2008-05-20_22-15-25-700000.mp4"), segPath) // +0.7s
}
n++
},
}
w.Initialize()
pts := 50 * time.Second
ntp := time.Date(2008, 5, 20, 22, 15, 25, 0, time.UTC)
strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
Base: unit.Base{
PTS: int64(pts) * 90000 / int64(time.Second),
NTP: ntp,
},
AU: [][]byte{
{5}, // IDR
},
})
pts += 700 * time.Millisecond
ntp = ntp.Add(700 * time.Millisecond)
strm.WriteUnit(desc.Medias[1], desc.Medias[1].Formats[0], &unit.MPEG4Audio{ // segment switch should happen here
Base: unit.Base{
PTS: int64(pts) * 44100 / int64(time.Second),
NTP: ntp,
},
AUs: [][]byte{{1, 2}},
})
pts += 400 * time.Millisecond
ntp = ntp.Add(400 * time.Millisecond)
strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
Base: unit.Base{
PTS: int64(pts) * 90000 / int64(time.Second),
NTP: ntp,
},
AU: [][]byte{
{5}, // IDR
},
})
pts += 100 * time.Millisecond
ntp = ntp.Add(100 * time.Millisecond)
strm.WriteUnit(desc.Medias[1], desc.Medias[1].Formats[0], &unit.MPEG4Audio{
Base: unit.Base{
PTS: int64(pts) * 44100 / int64(time.Second),
NTP: ntp,
},
AUs: [][]byte{{3, 4}},
})
pts += 400 * time.Millisecond
ntp = ntp.Add(400 * time.Millisecond)
strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
Base: unit.Base{
PTS: int64(pts) * 90000 / int64(time.Second),
NTP: ntp,
},
AU: [][]byte{
{5}, // IDR
},
})
time.Sleep(100 * time.Millisecond)
w.Close()
require.Equal(t, 2, n)
}

12
internal/staticsources/rpicamera/source.go
View File

@@ -148,10 +148,10 @@ func (s *Source) runPrimary(params defs.StaticSourceRunParams) error {
medias = append(medias, mediaSecondary)
}
var stream *stream.Stream
var strm *stream.Stream
initializeStream := func() {
if stream == nil {
if strm == nil {
res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{
Desc: &description.Session{Medias: medias},
GenerateRTPPackets: false,
@@ -160,7 +160,7 @@ func (s *Source) runPrimary(params defs.StaticSourceRunParams) error {
panic("should not happen")
}
stream = res.Stream
strm = res.Stream
}
}
@@ -184,7 +184,7 @@ func (s *Source) runPrimary(params defs.StaticSourceRunParams) error {
for _, pkt := range pkts {
pkt.Timestamp = uint32(pts)
stream.WriteRTPPacket(medi, medi.Formats[0], pkt, ntp, pts)
strm.WriteRTPPacket(medi, medi.Formats[0], pkt, ntp, pts)
}
}
@@ -211,13 +211,13 @@ func (s *Source) runPrimary(params defs.StaticSourceRunParams) error {
for _, pkt := range pkts {
pkt.Timestamp = uint32(pts)
pkt.PayloadType = 96
stream.WriteRTPPacket(mediaSecondary, mediaSecondary.Formats[0], pkt, ntp, pts)
strm.WriteRTPPacket(mediaSecondary, mediaSecondary.Formats[0], pkt, ntp, pts)
}
}
}
defer func() {
if stream != nil {
if strm != nil {
s.Parent.SetNotReady(defs.PathSourceStaticSetNotReadyReq{})
}
}()

42
internal/staticsources/rtsp/source_test.go
View File

@@ -44,7 +44,7 @@ func TestSource(t *testing.T) {
"tls",
} {
t.Run(source, func(t *testing.T) {
var stream *gortsplib.ServerStream
var strm *gortsplib.ServerStream
nonce, err := auth.GenerateNonce()
require.NoError(t, err)
@@ -67,17 +67,17 @@ func TestSource(t *testing.T) {
return &base.Response{
StatusCode: base.StatusOK,
}, stream, nil
}, strm, nil
},
onSetup: func(_ *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, stream, nil
}, strm, nil
},
onPlay: func(_ *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) {
go func() {
time.Sleep(100 * time.Millisecond)
err2 := stream.WritePacketRTP(media0, &rtp.Packet{
err2 := strm.WritePacketRTP(media0, &rtp.Packet{
Header: rtp.Header{
Version: 0x02,
PayloadType: 96,
@@ -126,13 +126,13 @@ func TestSource(t *testing.T) {
require.NoError(t, err)
defer s.Close()
stream = &gortsplib.ServerStream{
strm = &gortsplib.ServerStream{
Server: &s,
Desc: &description.Session{Medias: []*description.Media{media0}},
}
err = stream.Initialize()
err = strm.Initialize()
require.NoError(t, err)
defer stream.Close()
defer strm.Close()
var te *test.SourceTester
@@ -179,7 +179,7 @@ func TestSource(t *testing.T) {
}
func TestSourceNoPassword(t *testing.T) {
var stream *gortsplib.ServerStream
var strm *gortsplib.ServerStream
nonce, err := auth.GenerateNonce()
require.NoError(t, err)
@@ -201,12 +201,12 @@ func TestSourceNoPassword(t *testing.T) {
return &base.Response{
StatusCode: base.StatusOK,
}, stream, nil
}, strm, nil
},
onSetup: func(_ *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error) {
go func() {
time.Sleep(100 * time.Millisecond)
err2 := stream.WritePacketRTP(media0, &rtp.Packet{
err2 := strm.WritePacketRTP(media0, &rtp.Packet{
Header: rtp.Header{
Version: 0x02,
PayloadType: 96,
@@ -222,7 +222,7 @@ func TestSourceNoPassword(t *testing.T) {
return &base.Response{
StatusCode: base.StatusOK,
}, stream, nil
}, strm, nil
},
onPlay: func(_ *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) {
return &base.Response{
@@ -237,13 +237,13 @@ func TestSourceNoPassword(t *testing.T) {
require.NoError(t, err)
defer s.Close()
stream = &gortsplib.ServerStream{
strm = &gortsplib.ServerStream{
Server: &s,
Desc: &description.Session{Medias: []*description.Media{media0}},
}
err = stream.Initialize()
err = strm.Initialize()
require.NoError(t, err)
defer stream.Close()
defer strm.Close()
var sp conf.RTSPTransport
sp.UnmarshalJSON([]byte(`"tcp"`)) //nolint:errcheck
@@ -270,7 +270,7 @@ func TestSourceNoPassword(t *testing.T) {
func TestSourceRange(t *testing.T) {
for _, ca := range []string{"clock", "npt", "smpte"} {
t.Run(ca, func(t *testing.T) {
var stream *gortsplib.ServerStream
var strm *gortsplib.ServerStream
media0 := test.UniqueMediaH264()
@@ -279,12 +279,12 @@ func TestSourceRange(t *testing.T) {
onDescribe: func(_ *gortsplib.ServerHandlerOnDescribeCtx) (*base.Response, *gortsplib.ServerStream, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, stream, nil
}, strm, nil
},
onSetup: func(_ *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, stream, nil
}, strm, nil
},
onPlay: func(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) {
switch ca {
@@ -300,7 +300,7 @@ func TestSourceRange(t *testing.T) {
go func() {
time.Sleep(100 * time.Millisecond)
err := stream.WritePacketRTP(media0, &rtp.Packet{
err := strm.WritePacketRTP(media0, &rtp.Packet{
Header: rtp.Header{
Version: 0x02,
PayloadType: 96,
@@ -326,13 +326,13 @@ func TestSourceRange(t *testing.T) {
require.NoError(t, err)
defer s.Close()
stream = &gortsplib.ServerStream{
strm = &gortsplib.ServerStream{
Server: &s,
Desc: &description.Session{Medias: []*description.Media{media0}},
}
err = stream.Initialize()
err = strm.Initialize()
require.NoError(t, err)
defer stream.Close()
defer strm.Close()
cnf := &conf.Path{}