srt: support runOnRead (#2152)

internal/core/core.go

@@ -441,6 +441,7 @@ func (p *Core) createResources(initial bool) error {
 				p.conf.WriteTimeout,
 				p.conf.ReadBufferCount,
 				p.conf.UDPMaxPayloadSize,
+				p.externalCmdPool,
 				p.pathManager,
 				p,
 			)

internal/core/srt_conn.go

@@ -20,6 +20,7 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/bluenviron/mediamtx/internal/conf"
+	"github.com/bluenviron/mediamtx/internal/externalcmd"
 	"github.com/bluenviron/mediamtx/internal/formatprocessor"
 	"github.com/bluenviron/mediamtx/internal/logger"
 	"github.com/bluenviron/mediamtx/internal/stream"
@@ -64,6 +65,7 @@ type srtConn struct {
 	udpMaxPayloadSize int
 	connReq           srt.ConnRequest
 	wg                *sync.WaitGroup
+	externalCmdPool   *externalcmd.Pool
 	pathManager       srtConnPathManager
 	parent            srtConnParent
 
@@ -88,6 +90,7 @@ func newSRTConn(
 	udpMaxPayloadSize int,
 	connReq srt.ConnRequest,
 	wg *sync.WaitGroup,
+	externalCmdPool *externalcmd.Pool,
 	pathManager srtConnPathManager,
 	parent srtConnParent,
 ) *srtConn {
@@ -100,6 +103,7 @@ func newSRTConn(
 		udpMaxPayloadSize: udpMaxPayloadSize,
 		connReq:           connReq,
 		wg:                wg,
+		externalCmdPool:   externalCmdPool,
 		pathManager:       pathManager,
 		parent:            parent,
 		ctx:               ctx,
@@ -746,6 +750,24 @@ func (c *srtConn) runRead(req srtNewConnReq, pathName string, user string, pass
 	c.Log(logger.Info, "is reading from path '%s', %s",
 		res.path.name, sourceMediaInfo(medias))
 
+	pathConf := res.path.safeConf()
+
+	if pathConf.RunOnRead != "" {
+		c.Log(logger.Info, "runOnRead command started")
+		onReadCmd := externalcmd.NewCmd(
+			c.externalCmdPool,
+			pathConf.RunOnRead,
+			pathConf.RunOnReadRestart,
+			res.path.externalCmdEnv(),
+			func(err error) {
+				c.Log(logger.Info, "runOnRead command exited: %v", err)
+			})
+		defer func() {
+			onReadCmd.Close()
+			c.Log(logger.Info, "runOnRead command stopped")
+		}()
+	}
+
 	w = mpegts.NewWriter(bw, tracks)
 
 	// disable read deadline

internal/core/srt_server.go

@@ -11,6 +11,7 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/bluenviron/mediamtx/internal/conf"
+	"github.com/bluenviron/mediamtx/internal/externalcmd"
 	"github.com/bluenviron/mediamtx/internal/logger"
 )
 
@@ -60,6 +61,7 @@ type srtServer struct {
 	writeTimeout      conf.StringDuration
 	readBufferCount   int
 	udpMaxPayloadSize int
+	externalCmdPool   *externalcmd.Pool
 	pathManager       *pathManager
 	parent            srtServerParent
 
@@ -84,6 +86,7 @@ func newSRTServer(
 	writeTimeout conf.StringDuration,
 	readBufferCount int,
 	udpMaxPayloadSize int,
+	externalCmdPool *externalcmd.Pool,
 	pathManager *pathManager,
 	parent srtServerParent,
 ) (*srtServer, error) {
@@ -103,6 +106,7 @@ func newSRTServer(
 		writeTimeout:      writeTimeout,
 		readBufferCount:   readBufferCount,
 		udpMaxPayloadSize: udpMaxPayloadSize,
+		externalCmdPool:   externalCmdPool,
 		pathManager:       pathManager,
 		parent:            parent,
 		ctx:               ctx,
@@ -161,6 +165,7 @@ outer:
 				s.udpMaxPayloadSize,
 				req.connReq,
 				&s.wg,
+				s.externalCmdPool,
 				s.pathManager,
 				s)
 			s.conns[c] = struct{}{}