do not wait for external commands to exit during runtime

wait for them during shutdown.

README.md

@@ -23,7 +23,7 @@ Features:
 * Query and control the server through an HTTP API
 * Read Prometheus-compatible metrics
 * Redirect readers to other RTSP servers (load balancing)
-* Run custom commands when clients connect, disconnect, read or publish streams
+* Run external commands when clients connect, disconnect, read or publish streams
 * Reload the configuration without disconnecting existing clients (hot reloading)
 * Compatible with Linux, Windows and macOS, does not require any dependency or interpreter, it's a single executable
 
@@ -99,7 +99,7 @@ The `--network=host` flag is mandatory since Docker can change the source port o
 docker run --rm -it -e RTSP_PROTOCOLS=tcp -p 8554:8554 -p 1935:1935 -p 8888:8888 aler9/rtsp-simple-server
 ```
 
-Please keep in mind that the Docker image doesn't include _FFmpeg_. if you need to use _FFmpeg_ for a custom command or anything else, you need to build a Docker image that contains both _rtsp-simple-server_ and _FFmpeg_, by following instructions [here](https://github.com/aler9/rtsp-simple-server/discussions/278#discussioncomment-549104).
+Please keep in mind that the Docker image doesn't include _FFmpeg_. if you need to use _FFmpeg_ for an external command or anything else, you need to build a Docker image that contains both _rtsp-simple-server_ and _FFmpeg_, by following instructions [here](https://github.com/aler9/rtsp-simple-server/discussions/278#discussioncomment-549104).
 
 ## Basic usage
 

internal/core/core.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/aler9/rtsp-simple-server/internal/conf"
 	"github.com/aler9/rtsp-simple-server/internal/confwatcher"
+	"github.com/aler9/rtsp-simple-server/internal/externalcmd"
 	"github.com/aler9/rtsp-simple-server/internal/logger"
 	"github.com/aler9/rtsp-simple-server/internal/rlimit"
 )
@@ -21,21 +22,22 @@ var version = "v0.0.0"
 
 // Core is an instance of rtsp-simple-server.
 type Core struct {
-	ctx         context.Context
-	ctxCancel   func()
-	confPath    string
-	conf        *conf.Conf
-	confFound   bool
-	logger      *logger.Logger
-	metrics     *metrics
-	pprof       *pprof
-	pathManager *pathManager
-	rtspServer  *rtspServer
-	rtspsServer *rtspServer
-	rtmpServer  *rtmpServer
-	hlsServer   *hlsServer
-	api         *api
-	confWatcher *confwatcher.ConfWatcher
+	ctx             context.Context
+	ctxCancel       func()
+	confPath        string
+	conf            *conf.Conf
+	confFound       bool
+	logger          *logger.Logger
+	externalCmdPool *externalcmd.Pool
+	metrics         *metrics
+	pprof           *pprof
+	pathManager     *pathManager
+	rtspServer      *rtspServer
+	rtspsServer     *rtspServer
+	rtmpServer      *rtmpServer
+	hlsServer       *hlsServer
+	api             *api
+	confWatcher     *confwatcher.ConfWatcher
 
 	// in
 	apiConfigSet chan *conf.Conf
@@ -95,15 +97,6 @@ func New(args []string) (*Core, bool) {
 		return nil, false
 	}
 
-	if p.confFound {
-		p.confWatcher, err = confwatcher.New(p.confPath)
-		if err != nil {
-			p.Log(logger.Error, "%s", err)
-			p.closeResources(nil, false)
-			return nil, false
-		}
-	}
-
 	go p.run()
 
 	return p, true
@@ -176,10 +169,6 @@ outer:
 	p.ctxCancel()
 
 	p.closeResources(nil, false)
-
-	if p.confWatcher != nil {
-		p.confWatcher.Close()
-	}
 }
 
 func (p *Core) createResources(initial bool) error {
@@ -202,6 +191,10 @@ func (p *Core) createResources(initial bool) error {
 		}
 	}
 
+	if initial {
+		p.externalCmdPool = externalcmd.NewPool()
+	}
+
 	if p.conf.Metrics {
 		if p.metrics == nil {
 			p.metrics, err = newMetrics(
@@ -227,6 +220,7 @@ func (p *Core) createResources(initial bool) error {
 	if p.pathManager == nil {
 		p.pathManager = newPathManager(
 			p.ctx,
+			p.externalCmdPool,
 			p.conf.RTSPAddress,
 			p.conf.ReadTimeout,
 			p.conf.WriteTimeout,
@@ -245,6 +239,7 @@ func (p *Core) createResources(initial bool) error {
 			_, useMulticast := p.conf.Protocols[conf.Protocol(gortsplib.TransportUDPMulticast)]
 			p.rtspServer, err = newRTSPServer(
 				p.ctx,
+				p.externalCmdPool,
 				p.conf.RTSPAddress,
 				p.conf.AuthMethods,
 				p.conf.ReadTimeout,
@@ -280,6 +275,7 @@ func (p *Core) createResources(initial bool) error {
 		if p.rtspsServer == nil {
 			p.rtspsServer, err = newRTSPServer(
 				p.ctx,
+				p.externalCmdPool,
 				p.conf.RTSPSAddress,
 				p.conf.AuthMethods,
 				p.conf.ReadTimeout,
@@ -313,6 +309,7 @@ func (p *Core) createResources(initial bool) error {
 		if p.rtmpServer == nil {
 			p.rtmpServer, err = newRTMPServer(
 				p.ctx,
+				p.externalCmdPool,
 				p.conf.RTMPAddress,
 				p.conf.ReadTimeout,
 				p.conf.WriteTimeout,
@@ -365,6 +362,13 @@ func (p *Core) createResources(initial bool) error {
 		}
 	}
 
+	if initial && p.confFound {
+		p.confWatcher, err = confwatcher.New(p.confPath)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -488,6 +492,11 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
 		closeAPI = true
 	}
 
+	if newConf == nil && p.confWatcher != nil {
+		p.confWatcher.Close()
+		p.confWatcher = nil
+	}
+
 	if p.api != nil {
 		if closeAPI {
 			p.api.close()
@@ -532,7 +541,12 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
 		p.metrics = nil
 	}
 
-	if closeLogger && p.logger != nil {
+	if newConf == nil {
+		p.Log(logger.Info, "waiting for external commands")
+		p.externalCmdPool.Close()
+	}
+
+	if closeLogger {
 		p.logger.Close()
 		p.logger = nil
 	}

internal/core/path.go

@@ -208,6 +208,7 @@ type pathAPIPathsListSubReq struct {
 }
 
 type path struct {
+	externalCmdPool *externalcmd.Pool
 	rtspAddress     string
 	readTimeout     conf.StringDuration
 	writeTimeout    conf.StringDuration
@@ -252,6 +253,7 @@ type path struct {
 
 func newPath(
 	parentCtx context.Context,
+	externalCmdPool *externalcmd.Pool,
 	rtspAddress string,
 	readTimeout conf.StringDuration,
 	writeTimeout conf.StringDuration,
@@ -266,6 +268,7 @@ func newPath(
 	ctx, ctxCancel := context.WithCancel(parentCtx)
 
 	pa := &path{
+		externalCmdPool:         externalCmdPool,
 		rtspAddress:             rtspAddress,
 		readTimeout:             readTimeout,
 		writeTimeout:            writeTimeout,
@@ -340,7 +343,8 @@ func (pa *path) run() {
 	var onInitCmd *externalcmd.Cmd
 	if pa.conf.RunOnInit != "" {
 		pa.log(logger.Info, "runOnInit command started")
-		onInitCmd = externalcmd.New(
+		onInitCmd = externalcmd.NewCmd(
+			pa.externalCmdPool,
 			pa.conf.RunOnInit,
 			pa.conf.RunOnInitRestart,
 			pa.externalCmdEnv(),
@@ -485,11 +489,6 @@ func (pa *path) run() {
 		}
 	}
 
-	// close onDemandCmd after the source has been closed.
-	// this avoids a deadlock in which onDemandCmd is a
-	// RTSP publisher that sends a TEARDOWN request and waits
-	// for the response (like FFmpeg), but it can't since
-	// the path is already waiting for the command to close.
 	if pa.onDemandCmd != nil {
 		pa.onDemandCmd.Close()
 		pa.log(logger.Info, "runOnDemand command stopped")
@@ -543,7 +542,8 @@ func (pa *path) onDemandStartSource() {
 		pa.onDemandReadyTimer = time.NewTimer(time.Duration(pa.conf.SourceOnDemandStartTimeout))
 	} else {
 		pa.log(logger.Info, "runOnDemand command started")
-		pa.onDemandCmd = externalcmd.New(
+		pa.onDemandCmd = externalcmd.NewCmd(
+			pa.externalCmdPool,
 			pa.conf.RunOnDemand,
 			pa.conf.RunOnDemandRestart,
 			pa.externalCmdEnv(),
@@ -588,11 +588,6 @@ func (pa *path) onDemandCloseSource() {
 			pa.doPublisherRemove()
 		}
 
-		// close onDemandCmd after the source has been closed.
-		// this avoids a deadlock in which onDemandCmd is a
-		// RTSP publisher that sends a TEARDOWN request and waits
-		// for the response (like FFmpeg), but it can't since
-		// the path is already waiting for the command to close.
 		if pa.onDemandCmd != nil {
 			pa.onDemandCmd.Close()
 			pa.onDemandCmd = nil
@@ -637,11 +632,6 @@ func (pa *path) sourceSetNotReady() {
 		r.close()
 	}
 
-	// close onPublishCmd after all readers have been closed.
-	// this avoids a deadlock in which onPublishCmd is a
-	// RTSP reader that sends a TEARDOWN request and waits
-	// for the response (like FFmpeg), but it can't since
-	// the path is already waiting for the command to close.
 	if pa.onPublishCmd != nil {
 		pa.onPublishCmd.Close()
 		pa.onPublishCmd = nil
@@ -799,7 +789,8 @@ func (pa *path) handlePublisherRecord(req pathPublisherRecordReq) {
 
 	if pa.conf.RunOnPublish != "" {
 		pa.log(logger.Info, "runOnPublish command started")
-		pa.onPublishCmd = externalcmd.New(
+		pa.onPublishCmd = externalcmd.NewCmd(
+			pa.externalCmdPool,
 			pa.conf.RunOnPublish,
 			pa.conf.RunOnPublishRestart,
 			pa.externalCmdEnv(),

internal/core/path_manager.go

@@ -9,6 +9,7 @@ import (
 	"github.com/aler9/gortsplib/pkg/base"
 
 	"github.com/aler9/rtsp-simple-server/internal/conf"
+	"github.com/aler9/rtsp-simple-server/internal/externalcmd"
 	"github.com/aler9/rtsp-simple-server/internal/logger"
 )
 
@@ -21,6 +22,7 @@ type pathManagerParent interface {
 }
 
 type pathManager struct {
+	externalCmdPool *externalcmd.Pool
 	rtspAddress     string
 	readTimeout     conf.StringDuration
 	writeTimeout    conf.StringDuration
@@ -49,6 +51,7 @@ type pathManager struct {
 
 func newPathManager(
 	parentCtx context.Context,
+	externalCmdPool *externalcmd.Pool,
 	rtspAddress string,
 	readTimeout conf.StringDuration,
 	writeTimeout conf.StringDuration,
@@ -60,6 +63,7 @@ func newPathManager(
 	ctx, ctxCancel := context.WithCancel(parentCtx)
 
 	pm := &pathManager{
+		externalCmdPool:   externalCmdPool,
 		rtspAddress:       rtspAddress,
 		readTimeout:       readTimeout,
 		writeTimeout:      writeTimeout,
@@ -277,6 +281,7 @@ func (pm *pathManager) createPath(
 	matches []string) {
 	pm.paths[name] = newPath(
 		pm.ctx,
+		pm.externalCmdPool,
 		pm.rtspAddress,
 		pm.readTimeout,
 		pm.writeTimeout,

internal/core/rtmp_conn.go

@@ -53,6 +53,7 @@ type rtmpConnParent interface {
 }
 
 type rtmpConn struct {
+	externalCmdPool     *externalcmd.Pool
 	id                  string
 	rtspAddress         string
 	readTimeout         conf.StringDuration
@@ -75,6 +76,7 @@ type rtmpConn struct {
 
 func newRTMPConn(
 	parentCtx context.Context,
+	externalCmdPool *externalcmd.Pool,
 	id string,
 	rtspAddress string,
 	readTimeout conf.StringDuration,
@@ -89,6 +91,7 @@ func newRTMPConn(
 	ctx, ctxCancel := context.WithCancel(parentCtx)
 
 	c := &rtmpConn{
+		externalCmdPool:     externalCmdPool,
 		id:                  id,
 		rtspAddress:         rtspAddress,
 		readTimeout:         readTimeout,
@@ -148,7 +151,8 @@ func (c *rtmpConn) run() {
 		if c.runOnConnect != "" {
 			c.log(logger.Info, "runOnConnect command started")
 			_, port, _ := net.SplitHostPort(c.rtspAddress)
-			onConnectCmd := externalcmd.New(
+			onConnectCmd := externalcmd.NewCmd(
+				c.externalCmdPool,
 				c.runOnConnect,
 				c.runOnConnectRestart,
 				externalcmd.Environment{
@@ -289,7 +293,8 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
 
 	if c.path.Conf().RunOnRead != "" {
 		c.log(logger.Info, "runOnRead command started")
-		onReadCmd := externalcmd.New(
+		onReadCmd := externalcmd.NewCmd(
+			c.externalCmdPool,
 			c.path.Conf().RunOnRead,
 			c.path.Conf().RunOnReadRestart,
 			c.path.externalCmdEnv(),

internal/core/rtmp_server.go

@@ -12,6 +12,7 @@ import (
 	"github.com/aler9/gortsplib"
 
 	"github.com/aler9/rtsp-simple-server/internal/conf"
+	"github.com/aler9/rtsp-simple-server/internal/externalcmd"
 	"github.com/aler9/rtsp-simple-server/internal/logger"
 )
 
@@ -47,6 +48,7 @@ type rtmpServerParent interface {
 }
 
 type rtmpServer struct {
+	externalCmdPool     *externalcmd.Pool
 	readTimeout         conf.StringDuration
 	writeTimeout        conf.StringDuration
 	readBufferCount     int
@@ -71,6 +73,7 @@ type rtmpServer struct {
 
 func newRTMPServer(
 	parentCtx context.Context,
+	externalCmdPool *externalcmd.Pool,
 	address string,
 	readTimeout conf.StringDuration,
 	writeTimeout conf.StringDuration,
@@ -89,6 +92,7 @@ func newRTMPServer(
 	ctx, ctxCancel := context.WithCancel(parentCtx)
 
 	s := &rtmpServer{
+		externalCmdPool:     externalCmdPool,
 		readTimeout:         readTimeout,
 		writeTimeout:        writeTimeout,
 		readBufferCount:     readBufferCount,
@@ -170,6 +174,7 @@ outer:
 
 			c := newRTMPConn(
 				s.ctx,
+				s.externalCmdPool,
 				id,
 				s.rtspAddress,
 				s.readTimeout,

internal/core/rtsp_conn.go

@@ -24,6 +24,7 @@ type rtspConnParent interface {
 }
 
 type rtspConn struct {
+	externalCmdPool     *externalcmd.Pool
 	rtspAddress         string
 	authMethods         []headers.AuthMethod
 	readTimeout         conf.StringDuration
@@ -41,6 +42,7 @@ type rtspConn struct {
 }
 
 func newRTSPConn(
+	externalCmdPool *externalcmd.Pool,
 	rtspAddress string,
 	authMethods []headers.AuthMethod,
 	readTimeout conf.StringDuration,
@@ -50,6 +52,7 @@ func newRTSPConn(
 	conn *gortsplib.ServerConn,
 	parent rtspConnParent) *rtspConn {
 	c := &rtspConn{
+		externalCmdPool:     externalCmdPool,
 		rtspAddress:         rtspAddress,
 		authMethods:         authMethods,
 		readTimeout:         readTimeout,
@@ -65,7 +68,8 @@ func newRTSPConn(
 	if c.runOnConnect != "" {
 		c.log(logger.Info, "runOnConnect command started")
 		_, port, _ := net.SplitHostPort(c.rtspAddress)
-		c.onConnectCmd = externalcmd.New(
+		c.onConnectCmd = externalcmd.NewCmd(
+			c.externalCmdPool,
 			c.runOnConnect,
 			c.runOnConnectRestart,
 			externalcmd.Environment{

internal/core/rtsp_server.go

@@ -17,6 +17,7 @@ import (
 	"github.com/aler9/gortsplib/pkg/liberrors"
 
 	"github.com/aler9/rtsp-simple-server/internal/conf"
+	"github.com/aler9/rtsp-simple-server/internal/externalcmd"
 	"github.com/aler9/rtsp-simple-server/internal/logger"
 )
 
@@ -49,6 +50,7 @@ type rtspServerParent interface {
 }
 
 type rtspServer struct {
+	externalCmdPool     *externalcmd.Pool
 	authMethods         []headers.AuthMethod
 	readTimeout         conf.StringDuration
 	isTLS               bool
@@ -71,6 +73,7 @@ type rtspServer struct {
 
 func newRTSPServer(
 	parentCtx context.Context,
+	externalCmdPool *externalcmd.Pool,
 	address string,
 	authMethods []headers.AuthMethod,
 	readTimeout conf.StringDuration,
@@ -97,18 +100,19 @@ func newRTSPServer(
 	ctx, ctxCancel := context.WithCancel(parentCtx)
 
 	s := &rtspServer{
-		authMethods: authMethods,
-		readTimeout: readTimeout,
-		isTLS:       isTLS,
-		rtspAddress: rtspAddress,
-		protocols:   protocols,
-		metrics:     metrics,
-		pathManager: pathManager,
-		parent:      parent,
-		ctx:         ctx,
-		ctxCancel:   ctxCancel,
-		conns:       make(map[*gortsplib.ServerConn]*rtspConn),
-		sessions:    make(map[*gortsplib.ServerSession]*rtspSession),
+		externalCmdPool: externalCmdPool,
+		authMethods:     authMethods,
+		readTimeout:     readTimeout,
+		isTLS:           isTLS,
+		rtspAddress:     rtspAddress,
+		protocols:       protocols,
+		metrics:         metrics,
+		pathManager:     pathManager,
+		parent:          parent,
+		ctx:             ctx,
+		ctxCancel:       ctxCancel,
+		conns:           make(map[*gortsplib.ServerConn]*rtspConn),
+		sessions:        make(map[*gortsplib.ServerSession]*rtspSession),
 	}
 
 	s.srv = &gortsplib.Server{
@@ -251,6 +255,7 @@ func (s *rtspServer) newSessionID() (string, error) {
 // OnConnOpen implements gortsplib.ServerHandlerOnConnOpen.
 func (s *rtspServer) OnConnOpen(ctx *gortsplib.ServerHandlerOnConnOpenCtx) {
 	c := newRTSPConn(
+		s.externalCmdPool,
 		s.rtspAddress,
 		s.authMethods,
 		s.readTimeout,
@@ -300,6 +305,7 @@ func (s *rtspServer) OnSessionOpen(ctx *gortsplib.ServerHandlerOnSessionOpenCtx)
 	id, _ := s.newSessionID()
 
 	se := newRTSPSession(
+		s.externalCmdPool,
 		s.isTLS,
 		s.protocols,
 		id,

internal/core/rtsp_session.go

@@ -29,13 +29,14 @@ type rtspSessionParent interface {
 }
 
 type rtspSession struct {
-	isTLS       bool
-	protocols   map[conf.Protocol]struct{}
-	id          string
-	ss          *gortsplib.ServerSession
-	author      *gortsplib.ServerConn
-	pathManager rtspSessionPathManager
-	parent      rtspSessionParent
+	externalCmdPool *externalcmd.Pool
+	isTLS           bool
+	protocols       map[conf.Protocol]struct{}
+	id              string
+	ss              *gortsplib.ServerSession
+	author          *gortsplib.ServerConn
+	pathManager     rtspSessionPathManager
+	parent          rtspSessionParent
 
 	path            *path
 	state           gortsplib.ServerSessionState
@@ -47,6 +48,7 @@ type rtspSession struct {
 }
 
 func newRTSPSession(
+	externalCmdPool *externalcmd.Pool,
 	isTLS bool,
 	protocols map[conf.Protocol]struct{},
 	id string,
@@ -55,13 +57,14 @@ func newRTSPSession(
 	pathManager rtspSessionPathManager,
 	parent rtspSessionParent) *rtspSession {
 	s := &rtspSession{
-		isTLS:       isTLS,
-		protocols:   protocols,
-		id:          id,
-		ss:          ss,
-		author:      sc,
-		pathManager: pathManager,
-		parent:      parent,
+		externalCmdPool: externalCmdPool,
+		isTLS:           isTLS,
+		protocols:       protocols,
+		id:              id,
+		ss:              ss,
+		author:          sc,
+		pathManager:     pathManager,
+		parent:          parent,
 	}
 
 	s.log(logger.Info, "opened by %v", s.author.NetConn().RemoteAddr())
@@ -276,7 +279,8 @@ func (s *rtspSession) onPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo
 
 		if s.path.Conf().RunOnRead != "" {
 			s.log(logger.Info, "runOnRead command started")
-			s.onReadCmd = externalcmd.New(
+			s.onReadCmd = externalcmd.NewCmd(
+				s.externalCmdPool,
 				s.path.Conf().RunOnRead,
 				s.path.Conf().RunOnReadRestart,
 				s.path.externalCmdEnv(),

internal/externalcmd/cmd.go

@@ -13,6 +13,7 @@ type Environment map[string]string
 
 // Cmd is an external command.
 type Cmd struct {
+	pool    *Pool
 	cmdstr  string
 	restart bool
 	env     Environment
@@ -20,40 +21,39 @@ type Cmd struct {
 
 	// in
 	terminate chan struct{}
-
-	// out
-	done chan struct{}
 }
 
-// New allocates an Cmd.
-func New(
+// NewCmd allocates a Cmd.
+func NewCmd(
+	pool *Pool,
 	cmdstr string,
 	restart bool,
 	env Environment,
 	onExit func(int),
 ) *Cmd {
 	e := &Cmd{
+		pool:      pool,
 		cmdstr:    cmdstr,
 		restart:   restart,
 		env:       env,
 		onExit:    onExit,
 		terminate: make(chan struct{}),
-		done:      make(chan struct{}),
 	}
 
+	pool.wg.Add(1)
+
 	go e.run()
 
 	return e
 }
 
-// Close closes an Cmd.
+// Close closes the command. It doesn't wait for the command to exit.
 func (e *Cmd) Close() {
 	close(e.terminate)
-	<-e.done
 }
 
 func (e *Cmd) run() {
-	defer close(e.done)
+	defer e.pool.wg.Done()
 
 	for {
 		ok := func() bool {

internal/externalcmd/pool.go

@@ -0,0 +1,20 @@
+package externalcmd
+
+import (
+	"sync"
+)
+
+// Pool is a pool of external commands.
+type Pool struct {
+	wg sync.WaitGroup
+}
+
+// NewPool allocates a Pool.
+func NewPool() *Pool {
+	return &Pool{}
+}
+
+// Close waits for all external commands to exit.
+func (p *Pool) Close() {
+	p.wg.Wait()
+}