add runOnRecordSegmentCreate hook (#2451) (#2503)

README.md

@@ -1321,8 +1321,7 @@ paths:
 `runOnReady` allows to run a command when a stream is ready to be read:
 
 ```yml
-paths:
-  mypath:
+pathDefaults:
   # This is terminated with SIGINT when the stream is not ready anymore.
   # The following environment variables are available:
   # * MTX_PATH: path name
@@ -1339,8 +1338,7 @@ paths:
 `runOnNotReady` allows to run a command when a stream is not available anymore:
 
 ```yml
-paths:
-  mypath:
+pathDefaults:
   # Environment variables are the same of runOnReady.
   runOnNotReady: curl http://my-custom-server/webhook?path=$MTX_PATH&source_type=$MTX_SOURCE_TYPE&source_id=$MTX_SOURCE_ID
 ```
@@ -1348,8 +1346,7 @@ paths:
 `runOnRead` allows to run a command when a client starts reading:
 
 ```yml
-paths:
-  mypath:
+pathDefaults:
   # This is terminated with SIGINT when a client stops reading.
   # The following environment variables are available:
   # * MTX_PATH: path name
@@ -1366,18 +1363,30 @@ paths:
 `runOnUnread` allows to run a command when a client stops reading:
 
 ```yml
-paths:
-  mypath:
+pathDefaults:
   # Command to run when a client stops reading.
   # Environment variables are the same of runOnRead.
   runOnUnread: curl http://my-custom-server/webhook?path=$MTX_PATH&reader_type=$MTX_READER_TYPE&reader_id=$MTX_READER_ID
 ```
 
+`runOnRecordSegmentCreate` allows to run a command when a recording segment is created:
+
+```yml
+pathDefaults:
+  # Command to run when a recording segment is created.
+  # The following environment variables are available:
+  # * MTX_PATH: path name
+  # * RTSP_PORT: RTSP server port
+  # * G1, G2, ...: regular expression groups, if path name is
+  #   a regular expression.
+  # * MTX_SEGMENT_PATH: segment file path
+  runOnRecordSegmentCreate: curl http://my-custom-server/webhook?path=$MTX_PATH&segment_path=$MTX_SEGMENT_PATH
+```
+
 `runOnRecordSegmentComplete` allows to run a command when a recording segment is complete:
 
 ```yml
-paths:
-  mypath:
+pathDefaults:
   # Command to run when a recording segment is complete.
   # The following environment variables are available:
   # * MTX_PATH: path name

apidocs/openapi.yaml

@@ -347,6 +347,8 @@ components:
           type: boolean
         runOnUnread:
           type: string
+        runOnRecordSegmentCreate:
+          type: string
         runOnRecordSegmentComplete:
           type: string
 

internal/conf/path.go

@@ -138,6 +138,7 @@ type Path struct {
 	RunOnRead                  string         `json:"runOnRead"`
 	RunOnReadRestart           bool           `json:"runOnReadRestart"`
 	RunOnUnread                string         `json:"runOnUnread"`
+	RunOnRecordSegmentCreate   string         `json:"runOnRecordSegmentCreate"`
 	RunOnRecordSegmentComplete string         `json:"runOnRecordSegmentComplete"`
 }
 

internal/core/path.go

@@ -957,6 +957,20 @@ func (pa *path) startRecording() {
 		time.Duration(pa.conf.RecordSegmentDuration),
 		pa.name,
 		pa.stream,
+		func(segmentPath string) {
+			if pa.conf.RunOnRecordSegmentCreate != "" {
+				env := pa.externalCmdEnv()
+				env["MTX_SEGMENT_PATH"] = segmentPath
+
+				pa.Log(logger.Info, "runOnRecordSegmentCreate command launched")
+				externalcmd.NewCmd(
+					pa.externalCmdPool,
+					pa.conf.RunOnRecordSegmentCreate,
+					false,
+					env,
+					nil)
+			}
+		},
 		func(segmentPath string) {
 			if pa.conf.RunOnRecordSegmentComplete != "" {
 				env := pa.externalCmdEnv()

internal/record/agent.go

@@ -26,6 +26,9 @@ import (
 	"github.com/bluenviron/mediamtx/internal/unit"
 )
 
+// OnSegmentFunc is the prototype of the function passed as runOnSegmentStart / runOnSegmentComplete
+type OnSegmentFunc = func(string)
+
 func durationGoToMp4(v time.Duration, timeScale uint32) uint64 {
 	timeScale64 := uint64(timeScale)
 	secs := v / time.Second
@@ -111,7 +114,8 @@ type Agent struct {
 	partDuration      time.Duration
 	segmentDuration   time.Duration
 	stream            *stream.Stream
-	onSegmentComplete func(string)
+	onSegmentCreate   OnSegmentFunc
+	onSegmentComplete OnSegmentFunc
 	parent            logger.Writer
 
 	ctx                context.Context
@@ -125,7 +129,7 @@ type Agent struct {
 	done chan struct{}
 }
 
-// NewAgent allocates a nAgent.
+// NewAgent allocates an Agent.
 func NewAgent(
 	writeQueueSize int,
 	recordPath string,
@@ -133,16 +137,13 @@ func NewAgent(
 	segmentDuration time.Duration,
 	pathName string,
 	stream *stream.Stream,
-	onSegmentComplete func(string),
+	onSegmentCreate OnSegmentFunc,
+	onSegmentComplete OnSegmentFunc,
 	parent logger.Writer,
 ) *Agent {
 	recordPath = strings.ReplaceAll(recordPath, "%path", pathName)
 	recordPath += ".mp4"
 
-	if onSegmentComplete == nil {
-		onSegmentComplete = func(_ string) {}
-	}
-
 	ctx, ctxCancel := context.WithCancel(context.Background())
 
 	r := &Agent{
@@ -150,6 +151,7 @@ func NewAgent(
 		partDuration:      partDuration,
 		segmentDuration:   segmentDuration,
 		stream:            stream,
+		onSegmentCreate:   onSegmentCreate,
 		onSegmentComplete: onSegmentComplete,
 		parent:            parent,
 		ctx:               ctx,

internal/record/agent_test.go

@@ -77,6 +77,7 @@ func TestAgent(t *testing.T) {
 
 	recordPath := filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f")
 
+	segCreated := make(chan struct{}, 2)
 	segDone := make(chan struct{}, 2)
 
 	a := NewAgent(
@@ -86,6 +87,9 @@ func TestAgent(t *testing.T) {
 		1*time.Second,
 		"mypath",
 		stream,
+		func(fpath string) {
+			segCreated <- struct{}{}
+		},
 		func(fpath string) {
 			segDone <- struct{}{}
 		},
@@ -145,8 +149,11 @@ func TestAgent(t *testing.T) {
 		})
 	}
 
+	for i := 0; i < 2; i++ {
+		<-segCreated
 		<-segDone
-	<-segDone
+	}
+
 	a.Close()
 
 	_, err = os.Stat(filepath.Join(dir, "mypath", "2008-05-20_22-15-25-000125.mp4"))

internal/record/part.go

@@ -60,7 +60,7 @@ func newPart(
 func (p *part) close() error {
 	if p.s.f == nil {
 		p.s.fpath = encodeRecordPath(&recordPathParams{time: timeNow()}, p.s.r.path)
-		p.s.r.Log(logger.Debug, "opening segment %s", p.s.fpath)
+		p.s.r.Log(logger.Debug, "creating segment %s", p.s.fpath)
 
 		err := os.MkdirAll(filepath.Dir(p.s.fpath), 0o755)
 		if err != nil {
@@ -72,6 +72,8 @@ func (p *part) close() error {
 			return err
 		}
 
+		p.s.r.onSegmentCreate(p.s.fpath)
+
 		err = writeInit(f, p.s.r.tracks)
 		if err != nil {
 			f.Close()

mediamtx.yml

@@ -518,6 +518,15 @@ pathDefaults:
   # Environment variables are the same of runOnRead.
   runOnUnread:
 
+  # Command to run when a recording segment is created.
+  # The following environment variables are available:
+  # * MTX_PATH: path name
+  # * RTSP_PORT: RTSP server port
+  # * G1, G2, ...: regular expression groups, if path name is
+  #   a regular expression.
+  # * MTX_SEGMENT_PATH: segment file path
+  runOnRecordSegmentCreate:
+
   # Command to run when a recording segment is complete.
   # The following environment variables are available:
   # * MTX_PATH: path name
@@ -531,7 +540,8 @@ pathDefaults:
 # Path settings
 
 # Settings in "paths" are applied to specific paths, and the map key
-# is the name of the path. Any setting in "pathDefaults" can be overridden.
+# is the name of the path.
+# Any setting in "pathDefaults" can be overridden here.
 # It's possible to use regular expressions by using a tilde as prefix,
 # for example "~^(test1|test2)$" will match both "test1" and "test2",
 # for example "~^prefix" will match all paths that start with "prefix".