From 64d9060560b73ef2730c5e5ffa92d81243b35972 Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Sat, 16 Sep 2023 21:41:49 +0200 Subject: [PATCH] add additional environment variables to custom commands (#1414) (#2356) new variables: MTX_CONN_TYPE, MTX_CONN_ID, MTX_SOURCE_TYPE, MTX_SOURCE_ID, MTX_READER_TYPE, MTX_READ_ID --- README.md | 272 +++++++++++++++++++++--------- apidocs/openapi.yaml | 1 + internal/core/api_defs.go | 25 +-- internal/core/conn.go | 29 ++-- internal/core/hls_muxer.go | 4 +- internal/core/hls_source.go | 4 +- internal/core/path.go | 28 +-- internal/core/reader.go | 2 +- internal/core/rpicamera_source.go | 4 +- internal/core/rtmp_conn.go | 23 ++- internal/core/rtmp_source.go | 4 +- internal/core/rtsp_conn.go | 23 ++- internal/core/rtsp_server.go | 1 + internal/core/rtsp_session.go | 20 ++- internal/core/rtsp_source.go | 4 +- internal/core/source.go | 2 +- internal/core/source_redirect.go | 4 +- internal/core/source_static.go | 4 +- internal/core/srt_conn.go | 25 ++- internal/core/srt_source.go | 4 +- internal/core/udp_source.go | 4 +- internal/core/webrtc_session.go | 20 ++- internal/core/webrtc_source.go | 4 +- mediamtx.yml | 10 +- 24 files changed, 359 insertions(+), 162 deletions(-) diff --git a/README.md b/README.md index 20fd81b9..f7d09453 100644 --- a/README.md +++ b/README.md @@ -54,7 +54,7 @@ And can be read from the server with: * Query and control the server through the API * Reload the configuration without disconnecting existing clients (hot reloading) * Read Prometheus-compatible metrics -* Run external commands when clients connect, disconnect, read or publish streams +* Run external commands (hooks) when clients connect, disconnect, read or publish streams * Compatible with Linux, Windows and macOS, does not require any dependency or interpreter, it's a single executable **Note about rtsp-simple-server** @@ -111,6 +111,10 @@ _rtsp-simple-server_ has been rebranded as _MediaMTX_. The reason is pretty obvi * [Forward streams to another server](#forward-streams-to-another-server) * [On-demand publishing](#on-demand-publishing) * [Start on boot](#start-on-boot) + * [Hooks](#hooks) + * [API](#api) + * [Metrics](#metrics) + * [pprof](#pprof) * [RTSP-specific features](#rtsp-specific-features) * [Transport protocols](#transport-protocols) * [Encryption](#encryption) @@ -119,9 +123,6 @@ _rtsp-simple-server_ has been rebranded as _MediaMTX_. The reason is pretty obvi * [Encryption](#encryption-1) * [WebRTC-specific features](#webrtc-specific-features) * [Connectivity issues](#connectivity-issues) - * [API](#api) - * [Metrics](#metrics) - * [pprof](#pprof) * [Compile from source](#compile-from-source) * [Specifications](#specifications) * [Related projects](#related-projects) @@ -1241,6 +1242,193 @@ WinSW-x64 install The server is now installed as a system service and will start at boot time. +### Hooks + +The server allows to specify commands that are executed when a certain event happens, allowing the propagation of events to external software. + +`runOnConnect` allows to run a command when a client connects to the server: + +```yml +# This is terminated with SIGINT when a client disconnects from the server. +# The following environment variables are available: +# * RTSP_PORT: RTSP server port +# * MTX_CONN_TYPE: connection type +# * MTX_CONN_ID: connection ID +runOnConnect: curl http://my-custom-server/webhook +# Restart the command if it exits. +runOnConnectRestart: no +``` + +`runOnDisconnect` allows to run a command when a client disconnects from the server: + +```yml +# Environment variables are the same of runOnConnect. +runOnDisconnect: curl http://my-custom-server/webhook +``` + +`runOnInit` allows to run a command when a path is initialized. This can be used to publish a stream when the server is launched: + +```yml +paths: + mypath: + # This is terminated with SIGINT when the program closes. + # 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. + runOnInit: ffmpeg -i my_file.mp4 -c copy -f rtsp rtsp://localhost:8554/mypath + # Restart the command if it exits. + runOnInitRestart: no +``` + +`runOnDemand` allows to run a command when a path is requested by a reader. This can be used to publish a stream on demand: + +```yml +paths: + mypath: + # This is terminated with SIGINT when the program closes. + # 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. + runOnDemand: ffmpeg -i my_file.mp4 -c copy -f rtsp rtsp://localhost:8554/mypath + # Restart the command if it exits. + runOnDemandRestart: no +``` + +`runOnReady` allows to run a command when a stream is ready to be read: + +```yml +paths: + mypath: + # This is terminated with SIGINT when the stream is not ready anymore. + # The following environment variables are available: + # * MTX_PATH: path name + # * MTX_SOURCE_TYPE: source type + # * MTX_SOURCE_ID: source ID + # * RTSP_PORT: RTSP server port + # * G1, G2, ...: regular expression groups, if path name is + # a regular expression. + runOnReady: + # Restart the command if it exits. + runOnReadyRestart: no +``` + +`runOnNotReady` allows to run a command when a stream is not available anymore: + +```yml +paths: + mypath: + # Environment variables are the same of runOnReady. + runOnNotReady: +``` + +`runOnRead` allows to run a command when a client starts reading: + +```yml +paths: + mypath: + # This is terminated with SIGINT when a client stops reading. + # The following environment variables are available: + # * MTX_PATH: path name + # * MTX_READER_TYPE: reader type + # * MTX_READER_ID: reader ID + # * RTSP_PORT: RTSP server port + # * G1, G2, ...: regular expression groups, if path name is + # a regular expression. + runOnRead: + # Restart the command if it exits. + runOnReadRestart: no +``` + +`runOnUnread` allows to run a command when a client stops reading: + +```yml +paths: + mypath: + # Command to run when a client stops reading. + # Environment variables are the same of runOnRead. + runOnUnread: +``` + +### API + +The server can be queried and controlled with its API, that must be enabled by setting the `api` parameter in the configuration: + +```yml +api: yes +``` + +The API listens on `apiAddress`, that by default is `127.0.0.1:9997`; for instance, to obtain a list of active paths, run: + +``` +curl http://127.0.0.1:9997/v2/paths/list +``` + +Full documentation of the API is available on the [dedicated site](https://bluenviron.github.io/mediamtx/). + +### Metrics + +A metrics exporter, compatible with [Prometheus](https://prometheus.io/), can be enabled with the parameter `metrics: yes`; then the server can be queried for metrics with Prometheus or with a simple HTTP request: + +``` +curl localhost:9998/metrics +``` + +Obtaining: + +```ini +# metrics of every path +paths{name="[path_name]",state="[state]"} 1 +paths_bytes_received{name="[path_name]",state="[state]"} 1234 + +# metrics of every HLS muxer +hls_muxers{name="[name]"} 1 +hls_muxers_bytes_sent{name="[name]"} 187 + +# metrics of every RTSP connection +rtsp_conns{id="[id]"} 1 +rtsp_conns_bytes_received{id="[id]"} 1234 +rtsp_conns_bytes_sent{id="[id]"} 187 + +# metrics of every RTSP session +rtsp_sessions{id="[id]",state="idle"} 1 +rtsp_sessions_bytes_received{id="[id]",state="[state]"} 1234 +rtsp_sessions_bytes_sent{id="[id]",state="[state]"} 187 + +# metrics of every RTSPS connection +rtsps_conns{id="[id]"} 1 +rtsps_conns_bytes_received{id="[id]"} 1234 +rtsps_conns_bytes_sent{id="[id]"} 187 + +# metrics of every RTSPS session +rtsps_sessions{id="[id]",state="[state]"} 1 +rtsps_sessions_bytes_received{id="[id]",state="[state]"} 1234 +rtsps_sessions_bytes_sent{id="[id]",state="[state]"} 187 + +# metrics of every RTMP connection +rtmp_conns{id="[id]",state="[state]"} 1 +rtmp_conns_bytes_received{id="[id]",state="[state]"} 1234 +rtmp_conns_bytes_sent{id="[id]",state="[state]"} 187 + +# metrics of every WebRTC session +webrtc_sessions{id="[id]"} 1 +webrtc_sessions_bytes_received{id="[id]",state="[state]"} 1234 +webrtc_sessions_bytes_sent{id="[id]",state="[state]"} 187 +``` + +### pprof + +A performance monitor, compatible with pprof, can be enabled with the parameter `pprof: yes`; then the server can be queried for metrics with pprof-compatible tools, like: + +``` +go tool pprof -text http://localhost:9999/debug/pprof/goroutine +go tool pprof -text http://localhost:9999/debug/pprof/heap +go tool pprof -text http://localhost:9999/debug/pprof/profile?seconds=30 +``` + ### RTSP-specific features #### Transport protocols @@ -1395,82 +1583,6 @@ webrtcICEServers2: where secret is the secret of the TURN server. MediaMTX will generate a set of credentials by using the secret, and credentials will be sent to clients before the WebRTC/ICE connection is established. -### API - -The server can be queried and controlled with its API, that must be enabled by setting the `api` parameter in the configuration: - -```yml -api: yes -``` - -The API listens on `apiAddress`, that by default is `127.0.0.1:9997`; for instance, to obtain a list of active paths, run: - -``` -curl http://127.0.0.1:9997/v2/paths/list -``` - -Full documentation of the API is available on the [dedicated site](https://bluenviron.github.io/mediamtx/). - -### Metrics - -A metrics exporter, compatible with [Prometheus](https://prometheus.io/), can be enabled with the parameter `metrics: yes`; then the server can be queried for metrics with Prometheus or with a simple HTTP request: - -``` -curl localhost:9998/metrics -``` - -Obtaining: - -```ini -# metrics of every path -paths{name="[path_name]",state="[state]"} 1 -paths_bytes_received{name="[path_name]",state="[state]"} 1234 - -# metrics of every HLS muxer -hls_muxers{name="[name]"} 1 -hls_muxers_bytes_sent{name="[name]"} 187 - -# metrics of every RTSP connection -rtsp_conns{id="[id]"} 1 -rtsp_conns_bytes_received{id="[id]"} 1234 -rtsp_conns_bytes_sent{id="[id]"} 187 - -# metrics of every RTSP session -rtsp_sessions{id="[id]",state="idle"} 1 -rtsp_sessions_bytes_received{id="[id]",state="[state]"} 1234 -rtsp_sessions_bytes_sent{id="[id]",state="[state]"} 187 - -# metrics of every RTSPS connection -rtsps_conns{id="[id]"} 1 -rtsps_conns_bytes_received{id="[id]"} 1234 -rtsps_conns_bytes_sent{id="[id]"} 187 - -# metrics of every RTSPS session -rtsps_sessions{id="[id]",state="[state]"} 1 -rtsps_sessions_bytes_received{id="[id]",state="[state]"} 1234 -rtsps_sessions_bytes_sent{id="[id]",state="[state]"} 187 - -# metrics of every RTMP connection -rtmp_conns{id="[id]",state="[state]"} 1 -rtmp_conns_bytes_received{id="[id]",state="[state]"} 1234 -rtmp_conns_bytes_sent{id="[id]",state="[state]"} 187 - -# metrics of every WebRTC session -webrtc_sessions{id="[id]"} 1 -webrtc_sessions_bytes_received{id="[id]",state="[state]"} 1234 -webrtc_sessions_bytes_sent{id="[id]",state="[state]"} 187 -``` - -### pprof - -A performance monitor, compatible with pprof, can be enabled with the parameter `pprof: yes`; then the server can be queried for metrics with pprof-compatible tools, like: - -``` -go tool pprof -text http://localhost:9999/debug/pprof/goroutine -go tool pprof -text http://localhost:9999/debug/pprof/heap -go tool pprof -text http://localhost:9999/debug/pprof/profile?seconds=30 -``` - ## Compile from source ### Standard diff --git a/apidocs/openapi.yaml b/apidocs/openapi.yaml index 8c01f92d..ad1f9955 100644 --- a/apidocs/openapi.yaml +++ b/apidocs/openapi.yaml @@ -357,6 +357,7 @@ components: $ref: '#/components/schemas/PathConf' source: $ref: '#/components/schemas/PathSourceOrReader' + nullable: true ready: type: boolean readyTime: diff --git a/internal/core/api_defs.go b/internal/core/api_defs.go index 48588b02..6be6978d 100644 --- a/internal/core/api_defs.go +++ b/internal/core/api_defs.go @@ -8,17 +8,22 @@ import ( "github.com/bluenviron/mediamtx/internal/conf" ) +type apiPathSourceOrReader struct { + Type string `json:"type"` + ID string `json:"id"` +} + type apiPath struct { - Name string `json:"name"` - ConfName string `json:"confName"` - Conf *conf.PathConf `json:"conf"` - Source interface{} `json:"source"` - SourceReady bool `json:"sourceReady"` // Deprecated: renamed to Ready - Ready bool `json:"ready"` - ReadyTime *time.Time `json:"readyTime"` - Tracks []string `json:"tracks"` - BytesReceived uint64 `json:"bytesReceived"` - Readers []interface{} `json:"readers"` + Name string `json:"name"` + ConfName string `json:"confName"` + Conf *conf.PathConf `json:"conf"` + Source *apiPathSourceOrReader `json:"source"` + SourceReady bool `json:"sourceReady"` // Deprecated: renamed to Ready + Ready bool `json:"ready"` + ReadyTime *time.Time `json:"readyTime"` + Tracks []string `json:"tracks"` + BytesReceived uint64 `json:"bytesReceived"` + Readers []apiPathSourceOrReader `json:"readers"` } type apiPathsList struct { diff --git a/internal/core/conn.go b/internal/core/conn.go index 8c2cf465..8f8a5c2c 100644 --- a/internal/core/conn.go +++ b/internal/core/conn.go @@ -36,27 +36,29 @@ func newConn( } } -func (c *conn) open() { +func (c *conn) open(desc apiPathSourceOrReader) { if c.runOnConnect != "" { c.logger.Log(logger.Info, "runOnConnect command started") _, port, _ := net.SplitHostPort(c.rtspAddress) + env := externalcmd.Environment{ + "RTSP_PORT": port, + "MTX_CONN_TYPE": desc.Type, + "MTX_CONN_ID": desc.ID, + } + c.onConnectCmd = externalcmd.NewCmd( c.externalCmdPool, c.runOnConnect, c.runOnConnectRestart, - externalcmd.Environment{ - "MTX_PATH": "", - "RTSP_PATH": "", // deprecated - "RTSP_PORT": port, - }, + env, func(err error) { c.logger.Log(logger.Info, "runOnConnect command exited: %v", err) }) } } -func (c *conn) close() { +func (c *conn) close(desc apiPathSourceOrReader) { if c.onConnectCmd != nil { c.onConnectCmd.Close() c.logger.Log(logger.Info, "runOnConnect command stopped") @@ -64,16 +66,19 @@ func (c *conn) close() { if c.runOnDisconnect != "" { c.logger.Log(logger.Info, "runOnDisconnect command launched") + _, port, _ := net.SplitHostPort(c.rtspAddress) + env := externalcmd.Environment{ + "RTSP_PORT": port, + "MTX_CONN_TYPE": desc.Type, + "MTX_CONN_ID": desc.ID, + } + externalcmd.NewCmd( c.externalCmdPool, c.runOnDisconnect, false, - externalcmd.Environment{ - "MTX_PATH": "", - "RTSP_PATH": "", // deprecated - "RTSP_PORT": port, - }, + env, nil) } } diff --git a/internal/core/hls_muxer.go b/internal/core/hls_muxer.go index 22443a8a..75e001a4 100644 --- a/internal/core/hls_muxer.go +++ b/internal/core/hls_muxer.go @@ -557,8 +557,8 @@ func (m *hlsMuxer) processRequest(req *hlsMuxerHandleRequestReq) { } // apiReaderDescribe implements reader. -func (m *hlsMuxer) apiReaderDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (m *hlsMuxer) apiReaderDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "hlsMuxer", ID: "", } diff --git a/internal/core/hls_source.go b/internal/core/hls_source.go index 37df2544..e10d40c3 100644 --- a/internal/core/hls_source.go +++ b/internal/core/hls_source.go @@ -236,8 +236,8 @@ func (s *hlsSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan } // apiSourceDescribe implements sourceStaticImpl. -func (*hlsSource) apiSourceDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (*hlsSource) apiSourceDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "hlsSource", ID: "", } diff --git a/internal/core/path.go b/internal/core/path.go index 31bc5fe2..016e981b 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -145,11 +145,6 @@ type pathStopPublisherReq struct { res chan struct{} } -type pathAPISourceOrReader struct { - Type string `json:"type"` - ID string `json:"id"` -} - type pathAPIPathsListRes struct { data *apiPathsList paths map[string]*path @@ -744,11 +739,12 @@ func (pa *path) doAPIPathsGet(req pathAPIPathsGetReq) { Name: pa.name, ConfName: pa.confName, Conf: pa.conf, - Source: func() interface{} { + Source: func() *apiPathSourceOrReader { if pa.source == nil { return nil } - return pa.source.apiSourceDescribe() + v := pa.source.apiSourceDescribe() + return &v }(), SourceReady: pa.stream != nil, Ready: pa.stream != nil, @@ -771,8 +767,8 @@ func (pa *path) doAPIPathsGet(req pathAPIPathsGetReq) { } return pa.stream.BytesReceived() }(), - Readers: func() []interface{} { - ret := []interface{}{} + Readers: func() []apiPathSourceOrReader { + ret := []apiPathSourceOrReader{} for r := range pa.readers { ret = append(ret, r.apiReaderDescribe()) } @@ -905,12 +901,17 @@ func (pa *path) setReady(desc *description.Session, allocateEncoder bool) error pa.readyTime = time.Now() if pa.conf.RunOnReady != "" { + env := pa.externalCmdEnv() + desc := pa.source.apiSourceDescribe() + env["MTX_SOURCE_TYPE"] = desc.Type + env["MTX_SOURCE_ID"] = desc.ID + pa.Log(logger.Info, "runOnReady command started") pa.onReadyCmd = externalcmd.NewCmd( pa.externalCmdPool, pa.conf.RunOnReady, pa.conf.RunOnReadyRestart, - pa.externalCmdEnv(), + env, func(err error) { pa.Log(logger.Info, "runOnReady command exited: %v", err) }) @@ -936,12 +937,17 @@ func (pa *path) setNotReady() { } if pa.conf.RunOnNotReady != "" { + env := pa.externalCmdEnv() + desc := pa.source.apiSourceDescribe() + env["MTX_SOURCE_TYPE"] = desc.Type + env["MTX_SOURCE_ID"] = desc.ID + pa.Log(logger.Info, "runOnNotReady command launched") externalcmd.NewCmd( pa.externalCmdPool, pa.conf.RunOnNotReady, false, - pa.externalCmdEnv(), + env, nil) } diff --git a/internal/core/reader.go b/internal/core/reader.go index 58cf4ae9..f03bcb0b 100644 --- a/internal/core/reader.go +++ b/internal/core/reader.go @@ -3,5 +3,5 @@ package core // reader is an entity that can read a stream. type reader interface { close() - apiReaderDescribe() pathAPISourceOrReader + apiReaderDescribe() apiPathSourceOrReader } diff --git a/internal/core/rpicamera_source.go b/internal/core/rpicamera_source.go index de437276..26282b48 100644 --- a/internal/core/rpicamera_source.go +++ b/internal/core/rpicamera_source.go @@ -131,8 +131,8 @@ func (s *rpiCameraSource) run(ctx context.Context, cnf *conf.PathConf, reloadCon } // apiSourceDescribe implements sourceStaticImpl. -func (*rpiCameraSource) apiSourceDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (*rpiCameraSource) apiSourceDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "rpiCameraSource", ID: "", } diff --git a/internal/core/rtmp_conn.go b/internal/core/rtmp_conn.go index 46574182..7f014649 100644 --- a/internal/core/rtmp_conn.go +++ b/internal/core/rtmp_conn.go @@ -148,8 +148,9 @@ func (c *rtmpConn) ip() net.IP { func (c *rtmpConn) run() { //nolint:dupl defer c.wg.Done() - c.conn.open() - defer c.conn.close() + desc := c.apiReaderDescribe() + c.conn.open(desc) + defer c.conn.close(desc) err := c.runInner() @@ -262,12 +263,17 @@ func (c *rtmpConn) runRead(conn *rtmp.Conn, u *url.URL) error { pathConf := res.path.safeConf() if pathConf.RunOnRead != "" { + env := res.path.externalCmdEnv() + desc := c.apiReaderDescribe() + env["MTX_READER_TYPE"] = desc.Type + env["MTX_READER_ID"] = desc.ID + c.Log(logger.Info, "runOnRead command started") onReadCmd := externalcmd.NewCmd( c.externalCmdPool, pathConf.RunOnRead, pathConf.RunOnReadRestart, - res.path.externalCmdEnv(), + env, func(err error) { c.Log(logger.Info, "runOnRead command exited: %v", err) }) @@ -279,6 +285,11 @@ func (c *rtmpConn) runRead(conn *rtmp.Conn, u *url.URL) error { if pathConf.RunOnUnread != "" { defer func() { + env := res.path.externalCmdEnv() + desc := c.apiReaderDescribe() + env["MTX_READER_TYPE"] = desc.Type + env["MTX_READER_ID"] = desc.ID + c.Log(logger.Info, "runOnUnread command launched") externalcmd.NewCmd( c.externalCmdPool, @@ -630,8 +641,8 @@ func (c *rtmpConn) runPublish(conn *rtmp.Conn, u *url.URL) error { } // apiReaderDescribe implements reader. -func (c *rtmpConn) apiReaderDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (c *rtmpConn) apiReaderDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: func() string { if c.isTLS { return "rtmpsConn" @@ -643,7 +654,7 @@ func (c *rtmpConn) apiReaderDescribe() pathAPISourceOrReader { } // apiSourceDescribe implements source. -func (c *rtmpConn) apiSourceDescribe() pathAPISourceOrReader { +func (c *rtmpConn) apiSourceDescribe() apiPathSourceOrReader { return c.apiReaderDescribe() } diff --git a/internal/core/rtmp_source.go b/internal/core/rtmp_source.go index fb6fcd0d..2dbf6ffe 100644 --- a/internal/core/rtmp_source.go +++ b/internal/core/rtmp_source.go @@ -200,8 +200,8 @@ func (s *rtmpSource) runReader(u *url.URL, nconn net.Conn) error { } // apiSourceDescribe implements sourceStaticImpl. -func (*rtmpSource) apiSourceDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (*rtmpSource) apiSourceDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "rtmpSource", ID: "", } diff --git a/internal/core/rtsp_conn.go b/internal/core/rtsp_conn.go index 0cbbe609..45e0c44d 100644 --- a/internal/core/rtsp_conn.go +++ b/internal/core/rtsp_conn.go @@ -29,6 +29,7 @@ type rtspConnParent interface { type rtspConn struct { *conn + isTLS bool rtspAddress string authMethods []headers.AuthMethod readTimeout conf.StringDuration @@ -43,6 +44,7 @@ type rtspConn struct { } func newRTSPConn( + isTLS bool, rtspAddress string, authMethods []headers.AuthMethod, readTimeout conf.StringDuration, @@ -55,6 +57,7 @@ func newRTSPConn( parent rtspConnParent, ) *rtspConn { c := &rtspConn{ + isTLS: isTLS, rtspAddress: rtspAddress, authMethods: authMethods, readTimeout: readTimeout, @@ -76,7 +79,15 @@ func newRTSPConn( c.Log(logger.Info, "opened") - c.conn.open() + c.conn.open(apiPathSourceOrReader{ + Type: func() string { + if isTLS { + return "rtspsConn" + } + return "rtspConn" + }(), + ID: c.uuid.String(), + }) return c } @@ -102,7 +113,15 @@ func (c *rtspConn) ip() net.IP { func (c *rtspConn) onClose(err error) { c.Log(logger.Info, "closed: %v", err) - c.conn.close() + c.conn.close(apiPathSourceOrReader{ + Type: func() string { + if c.isTLS { + return "rtspsConn" + } + return "rtspConn" + }(), + ID: c.uuid.String(), + }) } // onRequest is called by rtspServer. diff --git a/internal/core/rtsp_server.go b/internal/core/rtsp_server.go index 8d4251c3..bcab3d11 100644 --- a/internal/core/rtsp_server.go +++ b/internal/core/rtsp_server.go @@ -217,6 +217,7 @@ outer: // OnConnOpen implements gortsplib.ServerHandlerOnConnOpen. func (s *rtspServer) OnConnOpen(ctx *gortsplib.ServerHandlerOnConnOpenCtx) { c := newRTSPConn( + s.isTLS, s.rtspAddress, s.authMethods, s.readTimeout, diff --git a/internal/core/rtsp_session.go b/internal/core/rtsp_session.go index cf744f9e..28b0fe9b 100644 --- a/internal/core/rtsp_session.go +++ b/internal/core/rtsp_session.go @@ -103,12 +103,17 @@ func (s *rtspSession) onUnread() { } if s.path.conf.RunOnUnread != "" { + env := s.path.externalCmdEnv() + desc := s.apiReaderDescribe() + env["MTX_READER_TYPE"] = desc.Type + env["MTX_READER_ID"] = desc.ID + s.Log(logger.Info, "runOnUnread command launched") externalcmd.NewCmd( s.externalCmdPool, s.path.conf.RunOnUnread, false, - s.path.externalCmdEnv(), + env, nil) } } @@ -308,12 +313,17 @@ func (s *rtspSession) onPlay(_ *gortsplib.ServerHandlerOnPlayCtx) (*base.Respons pathConf := s.path.safeConf() if pathConf.RunOnRead != "" { + env := s.path.externalCmdEnv() + desc := s.apiReaderDescribe() + env["MTX_READER_TYPE"] = desc.Type + env["MTX_READER_ID"] = desc.ID + s.Log(logger.Info, "runOnRead command started") s.onReadCmd = externalcmd.NewCmd( s.externalCmdPool, pathConf.RunOnRead, pathConf.RunOnReadRestart, - s.path.externalCmdEnv(), + env, func(err error) { s.Log(logger.Info, "runOnRead command exited: %v", err) }) @@ -396,8 +406,8 @@ func (s *rtspSession) onPause(_ *gortsplib.ServerHandlerOnPauseCtx) (*base.Respo } // apiReaderDescribe implements reader. -func (s *rtspSession) apiReaderDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (s *rtspSession) apiReaderDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: func() string { if s.isTLS { return "rtspsSession" @@ -409,7 +419,7 @@ func (s *rtspSession) apiReaderDescribe() pathAPISourceOrReader { } // apiSourceDescribe implements source. -func (s *rtspSession) apiSourceDescribe() pathAPISourceOrReader { +func (s *rtspSession) apiSourceDescribe() apiPathSourceOrReader { return s.apiReaderDescribe() } diff --git a/internal/core/rtsp_source.go b/internal/core/rtsp_source.go index e6481767..97bcc0cc 100644 --- a/internal/core/rtsp_source.go +++ b/internal/core/rtsp_source.go @@ -200,8 +200,8 @@ func (s *rtspSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf cha } // apiSourceDescribe implements sourceStaticImpl. -func (*rtspSource) apiSourceDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (*rtspSource) apiSourceDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "rtspSource", ID: "", } diff --git a/internal/core/source.go b/internal/core/source.go index d4b76ef3..03137761 100644 --- a/internal/core/source.go +++ b/internal/core/source.go @@ -16,7 +16,7 @@ import ( // - sourceRedirect type source interface { logger.Writer - apiSourceDescribe() pathAPISourceOrReader + apiSourceDescribe() apiPathSourceOrReader } func mediaDescription(media *description.Media) string { diff --git a/internal/core/source_redirect.go b/internal/core/source_redirect.go index fafbda34..4931b20d 100644 --- a/internal/core/source_redirect.go +++ b/internal/core/source_redirect.go @@ -11,8 +11,8 @@ func (*sourceRedirect) Log(logger.Level, string, ...interface{}) { } // apiSourceDescribe implements source. -func (*sourceRedirect) apiSourceDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (*sourceRedirect) apiSourceDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "redirect", ID: "", } diff --git a/internal/core/source_static.go b/internal/core/source_static.go index 540df6af..246fc6be 100644 --- a/internal/core/source_static.go +++ b/internal/core/source_static.go @@ -17,7 +17,7 @@ const ( type sourceStaticImpl interface { logger.Writer run(context.Context, *conf.PathConf, chan *conf.PathConf) error - apiSourceDescribe() pathAPISourceOrReader + apiSourceDescribe() apiPathSourceOrReader } type sourceStaticParent interface { @@ -216,7 +216,7 @@ func (s *sourceStatic) reloadConf(newConf *conf.PathConf) { } // apiSourceDescribe implements source. -func (s *sourceStatic) apiSourceDescribe() pathAPISourceOrReader { +func (s *sourceStatic) apiSourceDescribe() apiPathSourceOrReader { return s.impl.apiSourceDescribe() } diff --git a/internal/core/srt_conn.go b/internal/core/srt_conn.go index e4c66b12..1ae83f48 100644 --- a/internal/core/srt_conn.go +++ b/internal/core/srt_conn.go @@ -143,8 +143,9 @@ func (c *srtConn) ip() net.IP { func (c *srtConn) run() { //nolint:dupl defer c.wg.Done() - c.conn.open() - defer c.conn.close() + desc := c.apiReaderDescribe() + c.conn.open(desc) + defer c.conn.close(desc) err := c.runInner() @@ -716,12 +717,17 @@ func (c *srtConn) runRead(req srtNewConnReq, pathName string, user string, pass pathConf := res.path.safeConf() if pathConf.RunOnRead != "" { + env := res.path.externalCmdEnv() + desc := c.apiReaderDescribe() + env["MTX_READER_TYPE"] = desc.Type + env["MTX_READER_ID"] = desc.ID + c.Log(logger.Info, "runOnRead command started") onReadCmd := externalcmd.NewCmd( c.externalCmdPool, pathConf.RunOnRead, pathConf.RunOnReadRestart, - res.path.externalCmdEnv(), + env, func(err error) { c.Log(logger.Info, "runOnRead command exited: %v", err) }) @@ -733,12 +739,17 @@ func (c *srtConn) runRead(req srtNewConnReq, pathName string, user string, pass if pathConf.RunOnUnread != "" { defer func() { + env := res.path.externalCmdEnv() + desc := c.apiReaderDescribe() + env["MTX_READER_TYPE"] = desc.Type + env["MTX_READER_ID"] = desc.ID + c.Log(logger.Info, "runOnUnread command launched") externalcmd.NewCmd( c.externalCmdPool, pathConf.RunOnUnread, false, - res.path.externalCmdEnv(), + env, nil) }() } @@ -792,15 +803,15 @@ func (c *srtConn) setConn(sconn srt.Conn) { } // apiReaderDescribe implements reader. -func (c *srtConn) apiReaderDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (c *srtConn) apiReaderDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "srtConn", ID: c.uuid.String(), } } // apiSourceDescribe implements source. -func (c *srtConn) apiSourceDescribe() pathAPISourceOrReader { +func (c *srtConn) apiSourceDescribe() apiPathSourceOrReader { return c.apiReaderDescribe() } diff --git a/internal/core/srt_source.go b/internal/core/srt_source.go index 54c01696..bfff2e3d 100644 --- a/internal/core/srt_source.go +++ b/internal/core/srt_source.go @@ -278,8 +278,8 @@ func (s *srtSource) runReader(sconn srt.Conn) error { } // apiSourceDescribe implements sourceStaticImpl. -func (*srtSource) apiSourceDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (*srtSource) apiSourceDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "srtSource", ID: "", } diff --git a/internal/core/udp_source.go b/internal/core/udp_source.go index 26dffd88..6f56ae8f 100644 --- a/internal/core/udp_source.go +++ b/internal/core/udp_source.go @@ -312,8 +312,8 @@ func (s *udpSource) runReader(pc net.PacketConn) error { } // apiSourceDescribe implements sourceStaticImpl. -func (*udpSource) apiSourceDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (*udpSource) apiSourceDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "udpSource", ID: "", } diff --git a/internal/core/webrtc_session.go b/internal/core/webrtc_session.go index 69d55125..493a9921 100644 --- a/internal/core/webrtc_session.go +++ b/internal/core/webrtc_session.go @@ -530,12 +530,17 @@ func (s *webRTCSession) runRead() (int, error) { pathConf := res.path.safeConf() if pathConf.RunOnRead != "" { + env := res.path.externalCmdEnv() + desc := s.apiReaderDescribe() + env["MTX_READER_TYPE"] = desc.Type + env["MTX_READER_ID"] = desc.ID + s.Log(logger.Info, "runOnRead command started") onReadCmd := externalcmd.NewCmd( s.externalCmdPool, pathConf.RunOnRead, pathConf.RunOnReadRestart, - res.path.externalCmdEnv(), + env, func(err error) { s.Log(logger.Info, "runOnRead command exited: %v", err) }) @@ -547,12 +552,17 @@ func (s *webRTCSession) runRead() (int, error) { if pathConf.RunOnUnread != "" { defer func() { + env := res.path.externalCmdEnv() + desc := s.apiReaderDescribe() + env["MTX_READER_TYPE"] = desc.Type + env["MTX_READER_ID"] = desc.ID + s.Log(logger.Info, "runOnUnread command launched") externalcmd.NewCmd( s.externalCmdPool, pathConf.RunOnUnread, false, - res.path.externalCmdEnv(), + env, nil) }() } @@ -623,15 +633,15 @@ func (s *webRTCSession) addCandidates( } // apiSourceDescribe implements sourceStaticImpl. -func (s *webRTCSession) apiSourceDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (s *webRTCSession) apiSourceDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "webRTCSession", ID: s.uuid.String(), } } // apiReaderDescribe implements reader. -func (s *webRTCSession) apiReaderDescribe() pathAPISourceOrReader { +func (s *webRTCSession) apiReaderDescribe() apiPathSourceOrReader { return s.apiSourceDescribe() } diff --git a/internal/core/webrtc_source.go b/internal/core/webrtc_source.go index 0d61cf65..a9bc04f0 100644 --- a/internal/core/webrtc_source.go +++ b/internal/core/webrtc_source.go @@ -171,8 +171,8 @@ func (s *webRTCSource) run(ctx context.Context, cnf *conf.PathConf, _ chan *conf } // apiSourceDescribe implements sourceStaticImpl. -func (*webRTCSource) apiSourceDescribe() pathAPISourceOrReader { - return pathAPISourceOrReader{ +func (*webRTCSource) apiSourceDescribe() apiPathSourceOrReader { + return apiPathSourceOrReader{ Type: "webRTCSource", ID: "", } diff --git a/mediamtx.yml b/mediamtx.yml index e1d00678..43608217 100644 --- a/mediamtx.yml +++ b/mediamtx.yml @@ -56,6 +56,8 @@ pprofAddress: 127.0.0.1:9999 # This is terminated with SIGINT when a client disconnects from the server. # The following environment variables are available: # * RTSP_PORT: RTSP server port +# * MTX_CONN_TYPE: connection type +# * MTX_CONN_ID: connection ID runOnConnect: # Restart the command if it exits. runOnConnectRestart: no @@ -449,7 +451,7 @@ paths: # External commands path settings # Command to run when this path is initialized. - # This can be used to publish a stream and keep it always opened. + # This can be used to publish a stream when the server is launched. # This is terminated with SIGINT when the program closes. # The following environment variables are available: # * MTX_PATH: path name @@ -460,7 +462,7 @@ paths: # Restart the command if it exits. runOnInitRestart: no - # Command to run when this path is requested. + # Command to run when this path is requested by a reader. # This can be used to publish a stream on demand. # This is terminated with SIGINT when the path is not requested anymore. # The following environment variables are available: @@ -483,6 +485,8 @@ paths: # This is terminated with SIGINT when the stream is not ready anymore. # The following environment variables are available: # * MTX_PATH: path name + # * MTX_SOURCE_TYPE: source type + # * MTX_SOURCE_ID: source ID # * RTSP_PORT: RTSP server port # * G1, G2, ...: regular expression groups, if path name is # a regular expression. @@ -497,6 +501,8 @@ paths: # This is terminated with SIGINT when a client stops reading. # The following environment variables are available: # * MTX_PATH: path name + # * MTX_READER_TYPE: reader type + # * MTX_READER_ID: reader ID # * RTSP_PORT: RTSP server port # * G1, G2, ...: regular expression groups, if path name is # a regular expression.