restart recordings in case of errors (#2439)

internal/core/path.go

@@ -917,15 +917,15 @@ func (pa *path) setNotReady() {
 }
 
 func (pa *path) startRecording() {
-	pa.recordAgent = record.NewAgent(
-		pa.writeQueueSize,
-		pa.conf.RecordPath,
-		pa.conf.RecordFormat,
-		time.Duration(pa.conf.RecordPartDuration),
-		time.Duration(pa.conf.RecordSegmentDuration),
-		pa.name,
-		pa.stream,
-		func(segmentPath string) {
+	pa.recordAgent = &record.Agent{
+		WriteQueueSize:  pa.writeQueueSize,
+		RecordPath:      pa.conf.RecordPath,
+		Format:          pa.conf.RecordFormat,
+		PartDuration:    time.Duration(pa.conf.RecordPartDuration),
+		SegmentDuration: time.Duration(pa.conf.RecordSegmentDuration),
+		PathName:        pa.name,
+		Stream:          pa.stream,
+		OnSegmentCreate: func(segmentPath string) {
 			if pa.conf.RunOnRecordSegmentCreate != "" {
 				env := pa.externalCmdEnv()
 				env["MTX_SEGMENT_PATH"] = segmentPath
@@ -939,7 +939,7 @@ func (pa *path) startRecording() {
 					nil)
 			}
 		},
-		func(segmentPath string) {
+		OnSegmentComplete: func(segmentPath string) {
 			if pa.conf.RunOnRecordSegmentComplete != "" {
 				env := pa.externalCmdEnv()
 				env["MTX_SEGMENT_PATH"] = segmentPath
@@ -953,8 +953,9 @@ func (pa *path) startRecording() {
 					nil)
 			}
 		},
-		pa,
-	)
+		Parent: pa,
+	}
+	pa.recordAgent.Initialize()
 }
 
 func (pa *path) executeRemoveReader(r reader) {

internal/httpserv/wrapped_server_test.go

@@ -14,7 +14,6 @@ import (
 type testLogger struct{}
 
 func (testLogger) Log(_ logger.Level, _ string, _ ...interface{}) {
-	// fmt.Printf(format, args...)
 }
 
 func TestFilterEmptyPath(t *testing.T) {

internal/record/agent.go

@@ -1,124 +1,84 @@
 package record
 
 import (
-	"context"
-	"strings"
 	"time"
 
-	"github.com/bluenviron/mediacommon/pkg/formats/fmp4"
-
-	"github.com/bluenviron/mediamtx/internal/asyncwriter"
 	"github.com/bluenviron/mediamtx/internal/conf"
 	"github.com/bluenviron/mediamtx/internal/logger"
 	"github.com/bluenviron/mediamtx/internal/stream"
 )
 
-// OnSegmentFunc is the prototype of the function passed as runOnSegmentStart / runOnSegmentComplete
-type OnSegmentFunc = func(string)
-
-type sample struct {
-	*fmp4.PartSample
-	dts time.Duration
-}
-
-// Agent saves streams on disk.
+// Agent is a record agent.
 type Agent struct {
-	path              string
-	partDuration      time.Duration
-	segmentDuration   time.Duration
-	stream            *stream.Stream
-	onSegmentCreate   OnSegmentFunc
-	onSegmentComplete OnSegmentFunc
-	parent            logger.Writer
-
-	ctx       context.Context
-	ctxCancel func()
-	writer    *asyncwriter.Writer
-	format    recFormat
-
-	done chan struct{}
+	WriteQueueSize    int
+	RecordPath        string
+	Format            conf.RecordFormat
+	PartDuration      time.Duration
+	SegmentDuration   time.Duration
+	PathName          string
+	Stream            *stream.Stream
+	OnSegmentCreate   OnSegmentFunc
+	OnSegmentComplete OnSegmentFunc
+	Parent            logger.Writer
+
+	restartPause time.Duration
+
+	currentInstance *agentInstance
+
+	terminate chan struct{}
+	done      chan struct{}
 }
 
-// NewAgent allocates an Agent.
-func NewAgent(
-	writeQueueSize int,
-	path string,
-	format conf.RecordFormat,
-	partDuration time.Duration,
-	segmentDuration time.Duration,
-	pathName string,
-	stream *stream.Stream,
-	onSegmentCreate OnSegmentFunc,
-	onSegmentComplete OnSegmentFunc,
-	parent logger.Writer,
-) *Agent {
-	path = strings.ReplaceAll(path, "%path", pathName)
-
-	switch format {
-	case conf.RecordFormatMPEGTS:
-		path += ".ts"
-
-	default:
-		path += ".mp4"
+// Initialize initializes Agent.
+func (w *Agent) Initialize() {
+	if w.restartPause == 0 {
+		w.restartPause = 2 * time.Second
 	}
 
-	ctx, ctxCancel := context.WithCancel(context.Background())
-
-	a := &Agent{
-		path:              path,
-		partDuration:      partDuration,
-		segmentDuration:   segmentDuration,
-		stream:            stream,
-		onSegmentCreate:   onSegmentCreate,
-		onSegmentComplete: onSegmentComplete,
-		parent:            parent,
-		ctx:               ctx,
-		ctxCancel:         ctxCancel,
-		done:              make(chan struct{}),
-	}
-
-	a.writer = asyncwriter.New(writeQueueSize, a)
+	w.terminate = make(chan struct{})
+	w.done = make(chan struct{})
 
-	switch format {
-	case conf.RecordFormatMPEGTS:
-		a.format = newRecFormatMPEGTS(a)
-
-	default:
-		a.format = newRecFormatFMP4(a)
+	w.currentInstance = &agentInstance{
+		wrapper: w,
 	}
+	w.currentInstance.initialize()
 
-	go a.run()
-
-	return a
-}
-
-// Close closes the Agent.
-func (a *Agent) Close() {
-	a.Log(logger.Info, "recording stopped")
-
-	a.ctxCancel()
-	<-a.done
+	go w.run()
 }
 
 // Log is the main logging function.
-func (a *Agent) Log(level logger.Level, format string, args ...interface{}) {
-	a.parent.Log(level, "[record] "+format, args...)
+func (w *Agent) Log(level logger.Level, format string, args ...interface{}) {
+	w.Parent.Log(level, "[record] "+format, args...)
 }
 
-func (a *Agent) run() {
-	defer close(a.done)
-
-	a.writer.Start()
-
-	select {
-	case err := <-a.writer.Error():
-		a.Log(logger.Error, err.Error())
-		a.stream.RemoveReader(a.writer)
+// Close closes the agent.
+func (w *Agent) Close() {
+	w.Log(logger.Info, "recording stopped")
+	close(w.terminate)
+	<-w.done
+}
 
-	case <-a.ctx.Done():
-		a.stream.RemoveReader(a.writer)
-		a.writer.Stop()
+func (w *Agent) run() {
+	defer close(w.done)
+
+	for {
+		select {
+		case <-w.currentInstance.done:
+			w.currentInstance.close()
+		case <-w.terminate:
+			w.currentInstance.close()
+			return
+		}
+
+		select {
+		case <-time.After(w.restartPause):
+		case <-w.terminate:
+			return
+		}
+
+		w.currentInstance = &agentInstance{
+			wrapper: w,
+		}
+		w.currentInstance.initialize()
 	}
-
-	a.format.close()
 }

internal/record/agent_instance.go

@@ -0,0 +1,87 @@
+package record
+
+import (
+	"strings"
+	"time"
+
+	"github.com/bluenviron/mediacommon/pkg/formats/fmp4"
+
+	"github.com/bluenviron/mediamtx/internal/asyncwriter"
+	"github.com/bluenviron/mediamtx/internal/conf"
+	"github.com/bluenviron/mediamtx/internal/logger"
+)
+
+// OnSegmentFunc is the prototype of the function passed as runOnSegmentStart / runOnSegmentComplete
+type OnSegmentFunc = func(string)
+
+type sample struct {
+	*fmp4.PartSample
+	dts time.Duration
+}
+
+type agentInstance struct {
+	wrapper *Agent
+
+	resolvedPath string
+	writer       *asyncwriter.Writer
+	format       recFormat
+
+	terminate chan struct{}
+	done      chan struct{}
+}
+
+func (a *agentInstance) initialize() {
+	a.resolvedPath = strings.ReplaceAll(a.wrapper.RecordPath, "%path", a.wrapper.PathName)
+
+	switch a.wrapper.Format {
+	case conf.RecordFormatMPEGTS:
+		a.resolvedPath += ".ts"
+
+	default:
+		a.resolvedPath += ".mp4"
+	}
+
+	a.terminate = make(chan struct{})
+	a.done = make(chan struct{})
+
+	a.writer = asyncwriter.New(a.wrapper.WriteQueueSize, a.wrapper)
+
+	switch a.wrapper.Format {
+	case conf.RecordFormatMPEGTS:
+		a.format = &recFormatMPEGTS{
+			a: a,
+		}
+		a.format.initialize()
+
+	default:
+		a.format = &recFormatFMP4{
+			a: a,
+		}
+		a.format.initialize()
+	}
+
+	go a.run()
+}
+
+func (a *agentInstance) close() {
+	close(a.terminate)
+	<-a.done
+}
+
+func (a *agentInstance) run() {
+	defer close(a.done)
+
+	a.writer.Start()
+
+	select {
+	case err := <-a.writer.Error():
+		a.wrapper.Log(logger.Error, err.Error())
+		a.wrapper.Stream.RemoveReader(a.writer)
+
+	case <-a.terminate:
+		a.wrapper.Stream.RemoveReader(a.writer)
+		a.writer.Stop()
+	}
+
+	a.format.close()
+}