pimhwang
/
mediamtx
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

record: normalize decoding and encoding of segment paths (#2775)

pull/2777/head
Alessandro Ros 2 years ago committed by GitHub
parent
5da2ded815
commit
7c8e593b0b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 199 additions and 210 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 21
      internal/core/core.go
  2. 14
      internal/core/path.go
  3. 8
      internal/record/agent.go
  4. 28
      internal/record/agent_instance.go
  5. 30
      internal/record/agent_test.go
  6. 66
      internal/record/cleaner.go
  7. 15
      internal/record/cleaner_test.go
  8. 28
      internal/record/format_fmp4.go
  9. 10
      internal/record/format_fmp4_part.go
  10. 8
      internal/record/format_fmp4_segment.go
  11. 2
      internal/record/format_fmp4_track.go
  12. 24
      internal/record/format_mpegts.go
  13. 16
      internal/record/format_mpegts_segment.go
  14. 51
      internal/record/record_path_test.go
  15. 37
      internal/record/segment_path.go
  16. 51
      internal/record/segment_path_test.go

21
internal/core/core.go
Unescape View File

@ -40,9 +40,9 @@ func gatherCleanerEntries(paths map[string]*conf.Path) []record.CleanerEntry {
for _, pa := range paths { for _, pa := range paths {
if pa.Record && pa.RecordDeleteAfter != 0 { if pa.Record && pa.RecordDeleteAfter != 0 {
entry := record.CleanerEntry{ entry := record.CleanerEntry{
RecordPath: pa.RecordPath, SegmentPathFormat: pa.RecordPath,
RecordFormat: pa.RecordFormat, Format: pa.RecordFormat,
RecordDeleteAfter: time.Duration(pa.RecordDeleteAfter), DeleteAfter: time.Duration(pa.RecordDeleteAfter),
} }
out[entry] = struct{}{} out[entry] = struct{}{}
} }
@ -57,10 +57,10 @@ func gatherCleanerEntries(paths map[string]*conf.Path) []record.CleanerEntry {
} }
sort.Slice(out2, func(i, j int) bool { sort.Slice(out2, func(i, j int) bool {
if out2[i].RecordPath != out2[j].RecordPath { if out2[i].SegmentPathFormat != out2[j].SegmentPathFormat {
return out2[i].RecordPath < out2[j].RecordPath return out2[i].SegmentPathFormat < out2[j].SegmentPathFormat
} }
return out2[i].RecordDeleteAfter < out2[j].RecordDeleteAfter return out2[i].DeleteAfter < out2[j].DeleteAfter
}) })
return out2 return out2
@ -295,10 +295,11 @@ func (p *Core) createResources(initial bool) error {
cleanerEntries := gatherCleanerEntries(p.conf.Paths) cleanerEntries := gatherCleanerEntries(p.conf.Paths)
if len(cleanerEntries) != 0 && if len(cleanerEntries) != 0 &&
p.recordCleaner == nil { p.recordCleaner == nil {
p.recordCleaner = record.NewCleaner( p.recordCleaner = &record.Cleaner{
cleanerEntries, Entries: cleanerEntries,
p, Parent: p,
) }
p.recordCleaner.Initialize()
} }
if p.pathManager == nil { if p.pathManager == nil {

14
internal/core/path.go
Unescape View File

@ -897,13 +897,13 @@ func (pa *path) setNotReady() {
func (pa *path) startRecording() { func (pa *path) startRecording() {
pa.recordAgent = &record.Agent{ pa.recordAgent = &record.Agent{
WriteQueueSize: pa.writeQueueSize, WriteQueueSize: pa.writeQueueSize,
RecordPath: pa.conf.RecordPath, SegmentPathFormat: pa.conf.RecordPath,
Format: pa.conf.RecordFormat, Format: pa.conf.RecordFormat,
PartDuration: time.Duration(pa.conf.RecordPartDuration), PartDuration: time.Duration(pa.conf.RecordPartDuration),
SegmentDuration: time.Duration(pa.conf.RecordSegmentDuration), SegmentDuration: time.Duration(pa.conf.RecordSegmentDuration),
PathName: pa.name, PathName: pa.name,
Stream: pa.stream, Stream: pa.stream,
OnSegmentCreate: func(segmentPath string) { OnSegmentCreate: func(segmentPath string) {
if pa.conf.RunOnRecordSegmentCreate != "" { if pa.conf.RunOnRecordSegmentCreate != "" {
env := pa.externalCmdEnv() env := pa.externalCmdEnv()

8
internal/record/agent.go
Unescape View File

@ -8,10 +8,10 @@ import (
"github.com/bluenviron/mediamtx/internal/stream" "github.com/bluenviron/mediamtx/internal/stream"
) )
// Agent is a record agent. // Agent writes recordings to disk.
type Agent struct { type Agent struct {
WriteQueueSize int WriteQueueSize int
RecordPath string SegmentPathFormat string
Format conf.RecordFormat Format conf.RecordFormat
PartDuration time.Duration PartDuration time.Duration
SegmentDuration time.Duration SegmentDuration time.Duration
@ -47,7 +47,7 @@ func (w *Agent) Initialize() {
w.done = make(chan struct{}) w.done = make(chan struct{})
w.currentInstance = &agentInstance{ w.currentInstance = &agentInstance{
wrapper: w, agent: w,
} }
w.currentInstance.initialize() w.currentInstance.initialize()
@ -85,7 +85,7 @@ func (w *Agent) run() {
} }
w.currentInstance = &agentInstance{ w.currentInstance = &agentInstance{
wrapper: w, agent: w,
} }
w.currentInstance.initialize() w.currentInstance.initialize()
} }

28
internal/record/agent_instance.go
Unescape View File

@ -20,33 +20,35 @@ type sample struct {
} }
type agentInstance struct { type agentInstance struct {
wrapper *Agent agent *Agent
resolvedPath string segmentPathFormat string
writer *asyncwriter.Writer writer *asyncwriter.Writer
format format format format
terminate chan struct{} terminate chan struct{}
done chan struct{} done chan struct{}
} }
func (a *agentInstance) initialize() { func (a *agentInstance) initialize() {
a.resolvedPath = strings.ReplaceAll(a.wrapper.RecordPath, "%path", a.wrapper.PathName) a.segmentPathFormat = a.agent.SegmentPathFormat
switch a.wrapper.Format { a.segmentPathFormat = strings.ReplaceAll(a.segmentPathFormat, "%path", a.agent.PathName)
switch a.agent.Format {
case conf.RecordFormatMPEGTS: case conf.RecordFormatMPEGTS:
a.resolvedPath += ".ts" a.segmentPathFormat += ".ts"
default: default:
a.resolvedPath += ".mp4" a.segmentPathFormat += ".mp4"
} }
a.terminate = make(chan struct{}) a.terminate = make(chan struct{})
a.done = make(chan struct{}) a.done = make(chan struct{})
a.writer = asyncwriter.New(a.wrapper.WriteQueueSize, a.wrapper) a.writer = asyncwriter.New(a.agent.WriteQueueSize, a.agent)
switch a.wrapper.Format { switch a.agent.Format {
case conf.RecordFormatMPEGTS: case conf.RecordFormatMPEGTS:
a.format = &formatMPEGTS{ a.format = &formatMPEGTS{
a: a, a: a,
@ -75,11 +77,11 @@ func (a *agentInstance) run() {
select { select {
case err := <-a.writer.Error(): case err := <-a.writer.Error():
a.wrapper.Log(logger.Error, err.Error()) a.agent.Log(logger.Error, err.Error())
a.wrapper.Stream.RemoveReader(a.writer) a.agent.Stream.RemoveReader(a.writer)
case <-a.terminate: case <-a.terminate:
a.wrapper.Stream.RemoveReader(a.writer) a.agent.Stream.RemoveReader(a.writer)
a.writer.Stop() a.writer.Stop()
} }

30
internal/record/agent_test.go
Unescape View File

@ -156,13 +156,13 @@ func TestAgent(t *testing.T) {
} }
w := &Agent{ w := &Agent{
WriteQueueSize: 1024, WriteQueueSize: 1024,
RecordPath: recordPath, SegmentPathFormat: recordPath,
Format: f, Format: f,
PartDuration: 100 * time.Millisecond, PartDuration: 100 * time.Millisecond,
SegmentDuration: 1 * time.Second, SegmentDuration: 1 * time.Second,
PathName: "mypath", PathName: "mypath",
Stream: stream, Stream: stream,
OnSegmentCreate: func(fpath string) { OnSegmentCreate: func(fpath string) {
segCreated <- struct{}{} segCreated <- struct{}{}
}, },
@ -266,14 +266,14 @@ func TestAgentFMP4NegativeDTS(t *testing.T) {
recordPath := filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f") recordPath := filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f")
w := &Agent{ w := &Agent{
WriteQueueSize: 1024, WriteQueueSize: 1024,
RecordPath: recordPath, SegmentPathFormat: recordPath,
Format: conf.RecordFormatFMP4, Format: conf.RecordFormatFMP4,
PartDuration: 100 * time.Millisecond, PartDuration: 100 * time.Millisecond,
SegmentDuration: 1 * time.Second, SegmentDuration: 1 * time.Second,
PathName: "mypath", PathName: "mypath",
Stream: stream, Stream: stream,
Parent: &nilLogger{}, Parent: &nilLogger{},
} }
w.Initialize() w.Initialize()

66
internal/record/cleaner.go
Unescape View File

@ -41,39 +41,28 @@ func commonPath(v string) string {
// CleanerEntry is a cleaner entry. // CleanerEntry is a cleaner entry.
type CleanerEntry struct { type CleanerEntry struct {
RecordPath string SegmentPathFormat string
RecordFormat conf.RecordFormat Format conf.RecordFormat
RecordDeleteAfter time.Duration DeleteAfter time.Duration
} }
// Cleaner removes expired recording segments from disk. // Cleaner removes expired recording segments from disk.
type Cleaner struct { type Cleaner struct {
Entries []CleanerEntry
Parent logger.Writer
ctx context.Context ctx context.Context
ctxCancel func() ctxCancel func()
entries []CleanerEntry
parent logger.Writer
done chan struct{} done chan struct{}
} }
// NewCleaner allocates a Cleaner. // Initialize initializes a Cleaner.
func NewCleaner( func (c *Cleaner) Initialize() {
entries []CleanerEntry, c.ctx, c.ctxCancel = context.WithCancel(context.Background())
parent logger.Writer, c.done = make(chan struct{})
) *Cleaner {
ctx, ctxCancel := context.WithCancel(context.Background())
c := &Cleaner{
ctx: ctx,
ctxCancel: ctxCancel,
entries: entries,
parent: parent,
done: make(chan struct{}),
}
go c.run() go c.run()
return c
} }
// Close closes the Cleaner. // Close closes the Cleaner.
@ -84,16 +73,16 @@ func (c *Cleaner) Close() {
// Log is the main logging function. // Log is the main logging function.
func (c *Cleaner) Log(level logger.Level, format string, args ...interface{}) { func (c *Cleaner) Log(level logger.Level, format string, args ...interface{}) {
c.parent.Log(level, "[record cleaner]"+format, args...) c.Parent.Log(level, "[record cleaner]"+format, args...)
} }
func (c *Cleaner) run() { func (c *Cleaner) run() {
defer close(c.done) defer close(c.done)
interval := 30 * 60 * time.Second interval := 30 * 60 * time.Second
for _, e := range c.entries { for _, e := range c.Entries {
if interval > (e.RecordDeleteAfter / 2) { if interval > (e.DeleteAfter / 2) {
interval = e.RecordDeleteAfter / 2 interval = e.DeleteAfter / 2
} }
} }
@ -111,27 +100,27 @@ func (c *Cleaner) run() {
} }
func (c *Cleaner) doRun() { func (c *Cleaner) doRun() {
for _, e := range c.entries { for _, e := range c.Entries {
c.doRunEntry(&e) //nolint:errcheck c.doRunEntry(&e) //nolint:errcheck
} }
} }
func (c *Cleaner) doRunEntry(e *CleanerEntry) error { func (c *Cleaner) doRunEntry(e *CleanerEntry) error {
recordPath := e.RecordPath segmentPathFormat := e.SegmentPathFormat
// we have to convert to absolute paths switch e.Format {
// otherwise, commonPath and fpath inside Walk() won't have common elements
recordPath, _ = filepath.Abs(recordPath)
switch e.RecordFormat {
case conf.RecordFormatMPEGTS: case conf.RecordFormatMPEGTS:
recordPath += ".ts" segmentPathFormat += ".ts"
default: default:
recordPath += ".mp4" segmentPathFormat += ".mp4"
} }
commonPath := commonPath(recordPath) // we have to convert to absolute paths
// otherwise, commonPath and fpath inside Walk() won't have common elements
segmentPathFormat, _ = filepath.Abs(segmentPathFormat)
commonPath := commonPath(segmentPathFormat)
now := timeNow() now := timeNow()
filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error { //nolint:errcheck filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error { //nolint:errcheck
@ -140,9 +129,10 @@ func (c *Cleaner) doRunEntry(e *CleanerEntry) error {
} }
if !info.IsDir() { if !info.IsDir() {
params := decodeRecordPath(recordPath, fpath) var pa segmentPath
if params != nil { ok := pa.decode(segmentPathFormat, fpath)
if now.Sub(params.time) > e.RecordDeleteAfter { if ok {
if now.Sub(pa.time) > e.DeleteAfter {
c.Log(logger.Debug, "removing %s", fpath) c.Log(logger.Debug, "removing %s", fpath)
os.Remove(fpath) os.Remove(fpath)
} }

15
internal/record/cleaner_test.go
Unescape View File

@ -31,14 +31,15 @@ func TestCleaner(t *testing.T) {
err = os.WriteFile(filepath.Join(dir, "_-+*?^$()[]{}|_mypath", "2009-05-20_22-15-25-000427.mp4"), []byte{1}, 0o644) err = os.WriteFile(filepath.Join(dir, "_-+*?^$()[]{}|_mypath", "2009-05-20_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err) require.NoError(t, err)
c := NewCleaner( c := &Cleaner{
[]CleanerEntry{{ Entries: []CleanerEntry{{
RecordPath: recordPath, SegmentPathFormat: recordPath,
RecordFormat: conf.RecordFormatFMP4, Format: conf.RecordFormatFMP4,
RecordDeleteAfter: 10 * time.Second, DeleteAfter: 10 * time.Second,
}}, }},
nilLogger{}, Parent: nilLogger{},
) }
c.Initialize()
defer c.Close() defer c.Close()
time.Sleep(500 * time.Millisecond) time.Sleep(500 * time.Millisecond)

28
internal/record/format_fmp4.go
Unescape View File

@ -132,7 +132,7 @@ func (f *formatFMP4) initialize() {
} }
} }
for _, media := range f.a.wrapper.Stream.Desc().Medias { for _, media := range f.a.agent.Stream.Desc().Medias {
for _, forma := range media.Formats { for _, forma := range media.Formats {
switch forma := forma.(type) { switch forma := forma.(type) {
case *rtspformat.AV1: case *rtspformat.AV1:
@ -145,7 +145,7 @@ func (f *formatFMP4) initialize() {
firstReceived := false firstReceived := false
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.AV1) tunit := u.(*unit.AV1)
if tunit.TU == nil { if tunit.TU == nil {
return nil return nil
@ -202,7 +202,7 @@ func (f *formatFMP4) initialize() {
firstReceived := false firstReceived := false
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.VP9) tunit := u.(*unit.VP9)
if tunit.Frame == nil { if tunit.Frame == nil {
return nil return nil
@ -299,7 +299,7 @@ func (f *formatFMP4) initialize() {
var dtsExtractor *h265.DTSExtractor var dtsExtractor *h265.DTSExtractor
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.H265) tunit := u.(*unit.H265)
if tunit.AU == nil { if tunit.AU == nil {
return nil return nil
@ -384,7 +384,7 @@ func (f *formatFMP4) initialize() {
var dtsExtractor *h264.DTSExtractor var dtsExtractor *h264.DTSExtractor
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.H264) tunit := u.(*unit.H264)
if tunit.AU == nil { if tunit.AU == nil {
return nil return nil
@ -460,7 +460,7 @@ func (f *formatFMP4) initialize() {
firstReceived := false firstReceived := false
var lastPTS time.Duration var lastPTS time.Duration
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.MPEG4Video) tunit := u.(*unit.MPEG4Video)
if tunit.Frame == nil { if tunit.Frame == nil {
return nil return nil
@ -512,7 +512,7 @@ func (f *formatFMP4) initialize() {
firstReceived := false firstReceived := false
var lastPTS time.Duration var lastPTS time.Duration
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.MPEG1Video) tunit := u.(*unit.MPEG1Video)
if tunit.Frame == nil { if tunit.Frame == nil {
return nil return nil
@ -560,7 +560,7 @@ func (f *formatFMP4) initialize() {
parsed := false parsed := false
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.MJPEG) tunit := u.(*unit.MJPEG)
if tunit.Frame == nil { if tunit.Frame == nil {
return nil return nil
@ -596,7 +596,7 @@ func (f *formatFMP4) initialize() {
} }
track := addTrack(codec) track := addTrack(codec)
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.Opus) tunit := u.(*unit.Opus)
if tunit.Packets == nil { if tunit.Packets == nil {
return nil return nil
@ -629,7 +629,7 @@ func (f *formatFMP4) initialize() {
sampleRate := time.Duration(forma.ClockRate()) sampleRate := time.Duration(forma.ClockRate())
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.MPEG4Audio) tunit := u.(*unit.MPEG4Audio)
if tunit.AUs == nil { if tunit.AUs == nil {
return nil return nil
@ -662,7 +662,7 @@ func (f *formatFMP4) initialize() {
parsed := false parsed := false
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.MPEG1Audio) tunit := u.(*unit.MPEG1Audio)
if tunit.Frames == nil { if tunit.Frames == nil {
return nil return nil
@ -716,7 +716,7 @@ func (f *formatFMP4) initialize() {
parsed := false parsed := false
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.AC3) tunit := u.(*unit.AC3)
if tunit.Frames == nil { if tunit.Frames == nil {
return nil return nil
@ -782,7 +782,7 @@ func (f *formatFMP4) initialize() {
} }
track := addTrack(codec) track := addTrack(codec)
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.LPCM) tunit := u.(*unit.LPCM)
if tunit.Samples == nil { if tunit.Samples == nil {
return nil return nil
@ -799,7 +799,7 @@ func (f *formatFMP4) initialize() {
} }
} }
f.a.wrapper.Log(logger.Info, "recording %d %s", f.a.agent.Log(logger.Info, "recording %d %s",
len(f.tracks), len(f.tracks),
func() string { func() string {
if len(f.tracks) == 1 { if len(f.tracks) == 1 {

10
internal/record/format_fmp4_part.go
Unescape View File

@ -65,20 +65,20 @@ func newFormatFMP4Part(
func (p *formatFMP4Part) close() error { func (p *formatFMP4Part) close() error {
if p.s.fi == nil { if p.s.fi == nil {
p.s.fpath = encodeRecordPath(&recordPathParams{time: p.created}, p.s.f.a.resolvedPath) p.s.path = segmentPath{time: p.created}.encode(p.s.f.a.segmentPathFormat)
p.s.f.a.wrapper.Log(logger.Debug, "creating segment %s", p.s.fpath) p.s.f.a.agent.Log(logger.Debug, "creating segment %s", p.s.path)
err := os.MkdirAll(filepath.Dir(p.s.fpath), 0o755) err := os.MkdirAll(filepath.Dir(p.s.path), 0o755)
if err != nil { if err != nil {
return err return err
} }
fi, err := os.Create(p.s.fpath) fi, err := os.Create(p.s.path)
if err != nil { if err != nil {
return err return err
} }
p.s.f.a.wrapper.OnSegmentCreate(p.s.fpath) p.s.f.a.agent.OnSegmentCreate(p.s.path)
err = writeInit(fi, p.s.f.tracks) err = writeInit(fi, p.s.f.tracks)
if err != nil { if err != nil {

8
internal/record/format_fmp4_segment.go
Unescape View File

@ -37,7 +37,7 @@ type formatFMP4Segment struct {
f *formatFMP4 f *formatFMP4
startDTS time.Duration startDTS time.Duration
fpath string path string
fi *os.File fi *os.File
curPart *formatFMP4Part curPart *formatFMP4Part
} }
@ -60,14 +60,14 @@ func (s *formatFMP4Segment) close() error {
} }
if s.fi != nil { if s.fi != nil {
s.f.a.wrapper.Log(logger.Debug, "closing segment %s", s.fpath) s.f.a.agent.Log(logger.Debug, "closing segment %s", s.path)
err2 := s.fi.Close() err2 := s.fi.Close()
if err == nil { if err == nil {
err = err2 err = err2
} }
if err2 == nil { if err2 == nil {
s.f.a.wrapper.OnSegmentComplete(s.fpath) s.f.a.agent.OnSegmentComplete(s.path)
} }
} }
@ -78,7 +78,7 @@ func (s *formatFMP4Segment) record(track *formatFMP4Track, sample *sample) error
if s.curPart == nil { if s.curPart == nil {
s.curPart = newFormatFMP4Part(s, s.f.nextSequenceNumber, sample.dts) s.curPart = newFormatFMP4Part(s, s.f.nextSequenceNumber, sample.dts)
s.f.nextSequenceNumber++ s.f.nextSequenceNumber++
} else if s.curPart.duration() >= s.f.a.wrapper.PartDuration { } else if s.curPart.duration() >= s.f.a.agent.PartDuration {
err := s.curPart.close() err := s.curPart.close()
s.curPart = nil s.curPart = nil

2
internal/record/format_fmp4_track.go
Unescape View File

@ -47,7 +47,7 @@ func (t *formatFMP4Track) record(sample *sample) error {
if (!t.f.hasVideo || t.initTrack.Codec.IsVideo()) && if (!t.f.hasVideo || t.initTrack.Codec.IsVideo()) &&
!t.nextSample.IsNonSyncSample && !t.nextSample.IsNonSyncSample &&
(t.nextSample.dts-t.f.currentSegment.startDTS) >= t.f.a.wrapper.SegmentDuration { (t.nextSample.dts-t.f.currentSegment.startDTS) >= t.f.a.agent.SegmentDuration {
err := t.f.currentSegment.close() err := t.f.currentSegment.close()
if err != nil { if err != nil {
return err return err

24
internal/record/format_mpegts.go
Unescape View File

@ -59,7 +59,7 @@ func (f *formatMPEGTS) initialize() {
return track return track
} }
for _, media := range f.a.wrapper.Stream.Desc().Medias { for _, media := range f.a.agent.Stream.Desc().Medias {
for _, forma := range media.Formats { for _, forma := range media.Formats {
switch forma := forma.(type) { switch forma := forma.(type) {
case *rtspformat.H265: case *rtspformat.H265:
@ -67,7 +67,7 @@ func (f *formatMPEGTS) initialize() {
var dtsExtractor *h265.DTSExtractor var dtsExtractor *h265.DTSExtractor
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.H265) tunit := u.(*unit.H265)
if tunit.AU == nil { if tunit.AU == nil {
return nil return nil
@ -95,7 +95,7 @@ func (f *formatMPEGTS) initialize() {
var dtsExtractor *h264.DTSExtractor var dtsExtractor *h264.DTSExtractor
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.H264) tunit := u.(*unit.H264)
if tunit.AU == nil { if tunit.AU == nil {
return nil return nil
@ -124,7 +124,7 @@ func (f *formatMPEGTS) initialize() {
firstReceived := false firstReceived := false
var lastPTS time.Duration var lastPTS time.Duration
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.MPEG4Video) tunit := u.(*unit.MPEG4Video)
if tunit.Frame == nil { if tunit.Frame == nil {
return nil return nil
@ -154,7 +154,7 @@ func (f *formatMPEGTS) initialize() {
firstReceived := false firstReceived := false
var lastPTS time.Duration var lastPTS time.Duration
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.MPEG1Video) tunit := u.(*unit.MPEG1Video)
if tunit.Frame == nil { if tunit.Frame == nil {
return nil return nil
@ -188,7 +188,7 @@ func (f *formatMPEGTS) initialize() {
}(), }(),
}) })
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.Opus) tunit := u.(*unit.Opus)
if tunit.Packets == nil { if tunit.Packets == nil {
return nil return nil
@ -207,7 +207,7 @@ func (f *formatMPEGTS) initialize() {
Config: *forma.GetConfig(), Config: *forma.GetConfig(),
}) })
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.MPEG4Audio) tunit := u.(*unit.MPEG4Audio)
if tunit.AUs == nil { if tunit.AUs == nil {
return nil return nil
@ -224,7 +224,7 @@ func (f *formatMPEGTS) initialize() {
case *rtspformat.MPEG1Audio: case *rtspformat.MPEG1Audio:
track := addTrack(&mpegts.CodecMPEG1Audio{}) track := addTrack(&mpegts.CodecMPEG1Audio{})
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.MPEG1Audio) tunit := u.(*unit.MPEG1Audio)
if tunit.Frames == nil { if tunit.Frames == nil {
return nil return nil
@ -243,7 +243,7 @@ func (f *formatMPEGTS) initialize() {
sampleRate := time.Duration(forma.SampleRate) sampleRate := time.Duration(forma.SampleRate)
f.a.wrapper.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error { f.a.agent.Stream.AddReader(f.a.writer, media, forma, func(u unit.Unit) error {
tunit := u.(*unit.AC3) tunit := u.(*unit.AC3)
if tunit.Frames == nil { if tunit.Frames == nil {
return nil return nil
@ -269,7 +269,7 @@ func (f *formatMPEGTS) initialize() {
f.bw = bufio.NewWriterSize(f.dw, mpegtsMaxBufferSize) f.bw = bufio.NewWriterSize(f.dw, mpegtsMaxBufferSize)
f.mw = mpegts.NewWriter(f.bw, tracks) f.mw = mpegts.NewWriter(f.bw, tracks)
f.a.wrapper.Log(logger.Info, "recording %d %s", f.a.agent.Log(logger.Info, "recording %d %s",
len(tracks), len(tracks),
func() string { func() string {
if len(tracks) == 1 { if len(tracks) == 1 {
@ -292,7 +292,7 @@ func (f *formatMPEGTS) setupSegment(dts time.Duration, isVideo bool, randomAcces
case (!f.hasVideo || isVideo) && case (!f.hasVideo || isVideo) &&
randomAccess && randomAccess &&
(dts-f.currentSegment.startDTS) >= f.a.wrapper.SegmentDuration: (dts-f.currentSegment.startDTS) >= f.a.agent.SegmentDuration:
err := f.currentSegment.close() err := f.currentSegment.close()
if err != nil { if err != nil {
return err return err
@ -300,7 +300,7 @@ func (f *formatMPEGTS) setupSegment(dts time.Duration, isVideo bool, randomAcces
f.currentSegment = newFormatMPEGTSSegment(f, dts) f.currentSegment = newFormatMPEGTSSegment(f, dts)
case (dts - f.currentSegment.lastFlush) >= f.a.wrapper.PartDuration: case (dts - f.currentSegment.lastFlush) >= f.a.agent.PartDuration:
err := f.bw.Flush() err := f.bw.Flush()
if err != nil { if err != nil {
return err return err

16
internal/record/format_mpegts_segment.go
Unescape View File

@ -14,7 +14,7 @@ type formatMPEGTSSegment struct {
lastFlush time.Duration lastFlush time.Duration
created time.Time created time.Time
fpath string path string
fi *os.File fi *os.File
} }
@ -35,14 +35,14 @@ func (s *formatMPEGTSSegment) close() error {
err := s.f.bw.Flush() err := s.f.bw.Flush()
if s.fi != nil { if s.fi != nil {
s.f.a.wrapper.Log(logger.Debug, "closing segment %s", s.fpath) s.f.a.agent.Log(logger.Debug, "closing segment %s", s.path)
err2 := s.fi.Close() err2 := s.fi.Close()
if err == nil { if err == nil {
err = err2 err = err2
} }
if err2 == nil { if err2 == nil {
s.f.a.wrapper.OnSegmentComplete(s.fpath) s.f.a.agent.OnSegmentComplete(s.path)
} }
} }
@ -51,20 +51,20 @@ func (s *formatMPEGTSSegment) close() error {
func (s *formatMPEGTSSegment) Write(p []byte) (int, error) { func (s *formatMPEGTSSegment) Write(p []byte) (int, error) {
if s.fi == nil { if s.fi == nil {
s.fpath = encodeRecordPath(&recordPathParams{time: s.created}, s.f.a.resolvedPath) s.path = segmentPath{time: s.created}.encode(s.f.a.segmentPathFormat)
s.f.a.wrapper.Log(logger.Debug, "creating segment %s", s.fpath) s.f.a.agent.Log(logger.Debug, "creating segment %s", s.path)
err := os.MkdirAll(filepath.Dir(s.fpath), 0o755) err := os.MkdirAll(filepath.Dir(s.path), 0o755)
if err != nil { if err != nil {
return 0, err return 0, err
} }
fi, err := os.Create(s.fpath) fi, err := os.Create(s.path)
if err != nil { if err != nil {
return 0, err return 0, err
} }
s.f.a.wrapper.OnSegmentCreate(s.fpath) s.f.a.agent.OnSegmentCreate(s.path)
s.fi = fi s.fi = fi
} }

51
internal/record/record_path_test.go
Unescape View File

@ -1,51 +0,0 @@
package record
import (
"strings"
"testing"
"time"
"github.com/stretchr/testify/require"
)
var recordPathCases = []struct {
name string
format string
dec *recordPathParams
enc string
}{
{
"standard",
"%path/%Y-%m-%d_%H-%M-%S-%f.mp4",
&recordPathParams{
path: "mypath",
time: time.Date(2008, 11, 0o7, 11, 22, 4, 123456000, time.Local),
},
"mypath/2008-11-07_11-22-04-123456.mp4",
},
{
"unix seconds",
"%path/%s.mp4",
&recordPathParams{
path: "mypath",
time: time.Date(2021, 12, 2, 12, 15, 23, 0, time.UTC).Local(),
},
"mypath/1638447323.mp4",
},
}
func TestRecordPathDecode(t *testing.T) {
for _, ca := range recordPathCases {
t.Run(ca.name, func(t *testing.T) {
require.Equal(t, ca.dec, decodeRecordPath(ca.format, ca.enc))
})
}
}
func TestRecordPathEncode(t *testing.T) {
for _, ca := range recordPathCases {
t.Run(ca.name, func(t *testing.T) {
require.Equal(t, ca.enc, strings.ReplaceAll(encodeRecordPath(ca.dec, ca.format), "%path", "mypath"))
})
}
}

37
internal/record/record_path.go → internal/record/segment_path.go
Unescape View File

@ -21,12 +21,11 @@ func leadingZeros(v int, size int) string {
return out2 + out return out2 + out
} }
type recordPathParams struct { type segmentPath struct {
path string
time time.Time time time.Time
} }
func decodeRecordPath(format string, v string) *recordPathParams { func (p *segmentPath) decode(format string, v string) bool {
re := format re := format
for _, ch := range []uint8{ for _, ch := range []uint8{
@ -90,7 +89,7 @@ func decodeRecordPath(format string, v string) *recordPathParams {
matches := r.FindStringSubmatch(v) matches := r.FindStringSubmatch(v)
if matches == nil { if matches == nil {
return nil return false
} }
values := make(map[string]string) values := make(map[string]string)
@ -143,27 +142,23 @@ func decodeRecordPath(format string, v string) *recordPathParams {
} }
} }
var t time.Time
if unixSec > 0 { if unixSec > 0 {
t = time.Unix(unixSec, 0) p.time = time.Unix(unixSec, 0)
} else { } else {
t = time.Date(year, month, day, hour, minute, second, micros*1000, time.Local) p.time = time.Date(year, month, day, hour, minute, second, micros*1000, time.Local)
} }
return &recordPathParams{ return true
path: values["%path"],
time: t,
}
} }
func encodeRecordPath(params *recordPathParams, v string) string { func (p segmentPath) encode(format string) string {
v = strings.ReplaceAll(v, "%Y", strconv.FormatInt(int64(params.time.Year()), 10)) format = strings.ReplaceAll(format, "%Y", strconv.FormatInt(int64(p.time.Year()), 10))
v = strings.ReplaceAll(v, "%m", leadingZeros(int(params.time.Month()), 2)) format = strings.ReplaceAll(format, "%m", leadingZeros(int(p.time.Month()), 2))
v = strings.ReplaceAll(v, "%d", leadingZeros(params.time.Day(), 2)) format = strings.ReplaceAll(format, "%d", leadingZeros(p.time.Day(), 2))
v = strings.ReplaceAll(v, "%H", leadingZeros(params.time.Hour(), 2)) format = strings.ReplaceAll(format, "%H", leadingZeros(p.time.Hour(), 2))
v = strings.ReplaceAll(v, "%M", leadingZeros(params.time.Minute(), 2)) format = strings.ReplaceAll(format, "%M", leadingZeros(p.time.Minute(), 2))
v = strings.ReplaceAll(v, "%S", leadingZeros(params.time.Second(), 2)) format = strings.ReplaceAll(format, "%S", leadingZeros(p.time.Second(), 2))
v = strings.ReplaceAll(v, "%f", leadingZeros(params.time.Nanosecond()/1000, 6)) format = strings.ReplaceAll(format, "%f", leadingZeros(p.time.Nanosecond()/1000, 6))
v = strings.ReplaceAll(v, "%s", strconv.FormatInt(params.time.Unix(), 10)) format = strings.ReplaceAll(format, "%s", strconv.FormatInt(p.time.Unix(), 10))
return v return format
} }

51
internal/record/segment_path_test.go
Unescape View File

@ -0,0 +1,51 @@
package record
import (
"testing"
"time"
"github.com/stretchr/testify/require"
)
var segmentPathCases = []struct {
name string
format string
dec segmentPath
enc string
}{
{
"standard",
"%path/%Y-%m-%d_%H-%M-%S-%f.mp4",
segmentPath{
time: time.Date(2008, 11, 0o7, 11, 22, 4, 123456000, time.Local),
},
"%path/2008-11-07_11-22-04-123456.mp4",
},
{
"unix seconds",
"%path/%s.mp4",
segmentPath{
time: time.Date(2021, 12, 2, 12, 15, 23, 0, time.UTC).Local(),
},
"%path/1638447323.mp4",
},
}
func TestSegmentPathDecode(t *testing.T) {
for _, ca := range segmentPathCases {
t.Run(ca.name, func(t *testing.T) {
var dec segmentPath
ok := dec.decode(ca.format, ca.enc)
require.Equal(t, true, ok)
require.Equal(t, ca.dec, dec)
})
}
}
func TestSegmentPathEncode(t *testing.T) {
for _, ca := range segmentPathCases {
t.Run(ca.name, func(t *testing.T) {
require.Equal(t, ca.enc, ca.dec.encode(ca.format))
})
}
}
Write Preview
Loading…
Cancel
Save