CEF/mediamtx
1
0
Fork 0
forked from External/mediamtx
Code Pull requests Activity

make regexp groups available to custom commands (#642)

Browse source
This commit is contained in:
aler9 2021-12-08 20:50:09 +01:00
parent bf8e835cab
commit ebc201bda2
11 changed files with 121 additions and 82 deletions
Download patch file Download diff file Expand all files Collapse all files

6
internal/core/core_test.go
View file

@ -214,6 +214,10 @@ import (
)
func main() {
if os.Getenv("1") != "on" {
panic("environment not set")
}
track, err := gortsplib.NewTrackH264(96,
&gortsplib.TrackConfigH264{SPS: []byte{0x01, 0x02, 0x03, 0x04}, PPS: []byte{0x01, 0x02, 0x03, 0x04}})
if err != nil {
@ -259,7 +263,7 @@ func main() {
p1, ok := newInstance(fmt.Sprintf("rtmpDisable: yes\n"+
"hlsDisable: yes\n"+
"paths:\n"+
" all:\n"+
" '~^(on)demand$':\n"+
" runOnDemand: %s\n"+
" runOnDemandCloseAfter: 1s\n", execFile))
require.Equal(t, true, ok)

45
internal/core/path.go
View file

@ -4,6 +4,7 @@ import (
"context"
"fmt"
"net"
"strconv"
"strings"
"sync"
"time"
@ -215,6 +216,7 @@ type path struct {
confName string
conf *conf.PathConf
name string
matches []string
wg *sync.WaitGroup
parent pathParent
@ -258,6 +260,7 @@ func newPath(
confName string,
conf *conf.PathConf,
name string,
matches []string,
wg *sync.WaitGroup,
parent pathParent) *path {
ctx, ctxCancel := context.WithCancel(parentCtx)
@ -271,6 +274,7 @@ func newPath(
confName: confName,
conf: conf,
name: name,
matches: matches,
wg: wg,
parent: parent,
ctx: ctx,
@ -336,11 +340,10 @@ func (pa *path) run() {
var onInitCmd *externalcmd.Cmd
if pa.conf.RunOnInit != "" {
pa.log(logger.Info, "runOnInit command started")
_, port, _ := net.SplitHostPort(pa.rtspAddress)
onInitCmd = externalcmd.New(pa.conf.RunOnInit, pa.conf.RunOnInitRestart, externalcmd.Environment{
Path: pa.name,
Port: port,
})
onInitCmd = externalcmd.New(
pa.conf.RunOnInit,
pa.conf.RunOnInitRestart,
pa.externalCmdEnv())
}
err := func() error {
@ -514,6 +517,20 @@ func (pa *path) isOnDemand() bool {
return (pa.hasStaticSource() && pa.conf.SourceOnDemand) || pa.conf.RunOnDemand != ""
}
func (pa *path) externalCmdEnv() externalcmd.Environment {
_, port, _ := net.SplitHostPort(pa.rtspAddress)
env := externalcmd.Environment{
"RTSP_PATH": pa.name,
"RTSP_PORT": port,
}
for i, ma := range pa.matches[1:] {
env[strconv.FormatInt(int64(i+1), 10)] = ma
}
return env
}
func (pa *path) onDemandStartSource() {
pa.onDemandReadyTimer.Stop()
if pa.hasStaticSource() {
@ -521,11 +538,10 @@ func (pa *path) onDemandStartSource() {
pa.onDemandReadyTimer = time.NewTimer(time.Duration(pa.conf.SourceOnDemandStartTimeout))
} else {
pa.log(logger.Info, "runOnDemand command started")
_, port, _ := net.SplitHostPort(pa.rtspAddress)
pa.onDemandCmd = externalcmd.New(pa.conf.RunOnDemand, pa.conf.RunOnDemandRestart, externalcmd.Environment{
Path: pa.name,
Port: port,
})
pa.onDemandCmd = externalcmd.New(
pa.conf.RunOnDemand,
pa.conf.RunOnDemandRestart,
pa.externalCmdEnv())
pa.onDemandReadyTimer = time.NewTimer(time.Duration(pa.conf.RunOnDemandStartTimeout))
}
@ -775,11 +791,10 @@ func (pa *path) handlePublisherRecord(req pathPublisherRecordReq) {
if pa.conf.RunOnPublish != "" {
pa.log(logger.Info, "runOnPublish command started")
_, port, _ := net.SplitHostPort(pa.rtspAddress)
pa.onPublishCmd = externalcmd.New(pa.conf.RunOnPublish, pa.conf.RunOnPublishRestart, externalcmd.Environment{
Path: pa.name,
Port: port,
})
pa.onPublishCmd = externalcmd.New(
pa.conf.RunOnPublish,
pa.conf.RunOnPublishRestart,
pa.externalCmdEnv())
}
req.Res <- pathPublisherRecordRes{Stream: pa.stream}

38
internal/core/path_manager.go
View file

@ -83,7 +83,7 @@ func newPathManager(
for pathName, pathConf := range pm.pathConfs {
if pathConf.Regexp == nil {
pm.createPath(pathName, pathConf, pathName)
pm.createPath(pathName, pathConf, pathName, nil)
}
}
@ -147,7 +147,7 @@ outer:
// add new paths
for pathName, pathConf := range pm.pathConfs {
if _, ok := pm.paths[pathName]; !ok && pathConf.Regexp == nil {
pm.createPath(pathName, pathConf, pathName)
pm.createPath(pathName, pathConf, pathName, nil)
}
}
@ -164,7 +164,7 @@ outer:
}
case req := <-pm.describe:
pathName, pathConf, err := pm.findPathConf(req.PathName)
pathName, pathConf, pathMatches, err := pm.findPathConf(req.PathName)
if err != nil {
req.Res <- pathDescribeRes{Err: err}
continue
@ -185,13 +185,13 @@ outer:
// create path if it doesn't exist
if _, ok := pm.paths[req.PathName]; !ok {
pm.createPath(pathName, pathConf, req.PathName)
pm.createPath(pathName, pathConf, req.PathName, pathMatches)
}
req.Res <- pathDescribeRes{Path: pm.paths[req.PathName]}
case req := <-pm.readerSetupPlay:
pathName, pathConf, err := pm.findPathConf(req.PathName)
pathName, pathConf, pathMatches, err := pm.findPathConf(req.PathName)
if err != nil {
req.Res <- pathReaderSetupPlayRes{Err: err}
continue
@ -212,13 +212,13 @@ outer:
// create path if it doesn't exist
if _, ok := pm.paths[req.PathName]; !ok {
pm.createPath(pathName, pathConf, req.PathName)
pm.createPath(pathName, pathConf, req.PathName, pathMatches)
}
req.Res <- pathReaderSetupPlayRes{Path: pm.paths[req.PathName]}
case req := <-pm.publisherAnnounce:
pathName, pathConf, err := pm.findPathConf(req.PathName)
pathName, pathConf, pathMatches, err := pm.findPathConf(req.PathName)
if err != nil {
req.Res <- pathPublisherAnnounceRes{Err: err}
continue
@ -239,7 +239,7 @@ outer:
// create path if it doesn't exist
if _, ok := pm.paths[req.PathName]; !ok {
pm.createPath(pathName, pathConf, req.PathName)
pm.createPath(pathName, pathConf, req.PathName, pathMatches)
}
req.Res <- pathPublisherAnnounceRes{Path: pm.paths[req.PathName]}
@ -270,7 +270,11 @@ outer:
}
}
func (pm *pathManager) createPath(confName string, conf *conf.PathConf, name string) {
func (pm *pathManager) createPath(
confName string,
conf *conf.PathConf,
name string,
matches []string) {
pm.paths[name] = newPath(
pm.ctx,
pm.rtspAddress,
@ -281,29 +285,33 @@ func (pm *pathManager) createPath(confName string, conf *conf.PathConf, name str
confName,
conf,
name,
matches,
&pm.wg,
pm)
}
func (pm *pathManager) findPathConf(name string) (string, *conf.PathConf, error) {
func (pm *pathManager) findPathConf(name string) (string, *conf.PathConf, []string, error) {
err := conf.IsValidPathName(name)
if err != nil {
return "", nil, fmt.Errorf("invalid path name: %s (%s)", err, name)
return "", nil, nil, fmt.Errorf("invalid path name: %s (%s)", err, name)
}
// normal path
if pathConf, ok := pm.pathConfs[name]; ok {
return name, pathConf, nil
return name, pathConf, nil, nil
}
// regular expression path
for pathName, pathConf := range pm.pathConfs {
if pathConf.Regexp != nil && pathConf.Regexp.MatchString(name) {
return pathName, pathConf, nil
if pathConf.Regexp != nil {
m := pathConf.Regexp.FindStringSubmatch(name)
if m != nil {
return pathName, pathConf, m, nil
}
}
}
return "", nil, fmt.Errorf("path '%s' is not configured", name)
return "", nil, nil, fmt.Errorf("path '%s' is not configured", name)
}
func (pm *pathManager) authenticate(

20
internal/core/rtmp_conn.go
View file

@ -148,10 +148,13 @@ func (c *rtmpConn) run() {
if c.runOnConnect != "" {
c.log(logger.Info, "runOnConnect command started")
_, port, _ := net.SplitHostPort(c.rtspAddress)
onConnectCmd := externalcmd.New(c.runOnConnect, c.runOnConnectRestart, externalcmd.Environment{
Path: "",
Port: port,
})
onConnectCmd := externalcmd.New(
c.runOnConnect,
c.runOnConnectRestart,
externalcmd.Environment{
"RTSP_PATH": "",
"RTSP_PORT": port,
})
defer func() {
onConnectCmd.Close()
@ -283,11 +286,10 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
if c.path.Conf().RunOnRead != "" {
c.log(logger.Info, "runOnRead command started")
_, port, _ := net.SplitHostPort(c.rtspAddress)
onReadCmd := externalcmd.New(c.path.Conf().RunOnRead, c.path.Conf().RunOnReadRestart, externalcmd.Environment{
Path: c.path.Name(),
Port: port,
})
onReadCmd := externalcmd.New(
c.path.Conf().RunOnRead,
c.path.Conf().RunOnReadRestart,
c.path.externalCmdEnv())
defer func() {
onReadCmd.Close()
c.log(logger.Info, "runOnRead command stopped")

11
internal/core/rtsp_conn.go
View file

@ -65,10 +65,13 @@ func newRTSPConn(
if c.runOnConnect != "" {
c.log(logger.Info, "runOnConnect command started")
_, port, _ := net.SplitHostPort(c.rtspAddress)
c.onConnectCmd = externalcmd.New(c.runOnConnect, c.runOnConnectRestart, externalcmd.Environment{
Path: "",
Port: port,
})
c.onConnectCmd = externalcmd.New(
c.runOnConnect,
c.runOnConnectRestart,
externalcmd.Environment{
"RTSP_PATH": "",
"RTSP_PORT": port,
})
}
return c

1
internal/core/rtsp_server.go
View file

@ -301,7 +301,6 @@ func (s *rtspServer) OnSessionOpen(ctx *gortsplib.ServerHandlerOnSessionOpenCtx)
se := newRTSPSession(
s.isTLS,
s.rtspAddress,
s.protocols,
id,
ctx.Session,

12
internal/core/rtsp_session.go
View file

@ -30,7 +30,6 @@ type rtspSessionParent interface {
type rtspSession struct {
isTLS bool
rtspAddress string
protocols map[conf.Protocol]struct{}
id string
ss *gortsplib.ServerSession
@ -49,7 +48,6 @@ type rtspSession struct {
func newRTSPSession(
isTLS bool,
rtspAddress string,
protocols map[conf.Protocol]struct{},
id string,
ss *gortsplib.ServerSession,
@ -58,7 +56,6 @@ func newRTSPSession(
parent rtspSessionParent) *rtspSession {
s := &rtspSession{
isTLS: isTLS,
rtspAddress: rtspAddress,
protocols: protocols,
id: id,
ss: ss,
@ -279,11 +276,10 @@ func (s *rtspSession) onPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo
if s.path.Conf().RunOnRead != "" {
s.log(logger.Info, "runOnRead command started")
_, port, _ := net.SplitHostPort(s.rtspAddress)
s.onReadCmd = externalcmd.New(s.path.Conf().RunOnRead, s.path.Conf().RunOnReadRestart, externalcmd.Environment{
Path: s.path.Name(),
Port: port,
})
s.onReadCmd = externalcmd.New(
s.path.Conf().RunOnRead,
s.path.Conf().RunOnReadRestart,
s.path.externalCmdEnv())
}
s.stateMutex.Lock()

5
internal/externalcmd/cmd.go
View file

@ -9,10 +9,7 @@ const (
)
// Environment is a Cmd environment.
type Environment struct {
Path string
Port string
}
type Environment map[string]string
// Cmd is an external command.
type Cmd struct {

8
internal/externalcmd/cmd_unix.go
View file

@ -12,10 +12,10 @@ import (
func (e *Cmd) runInner() bool {
cmd := exec.Command("/bin/sh", "-c", "exec "+e.cmdstr)
cmd.Env = append(os.Environ(),
"RTSP_PATH="+e.env.Path,
"RTSP_PORT="+e.env.Port,
)
cmd.Env = append([]string(nil), os.Environ()...)
for key, val := range e.env {
cmd.Env = append(cmd.Env, key+"="+val)
}
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr

24
internal/externalcmd/cmd_win.go
View file

@ -12,11 +12,13 @@ import (
)
func (e *Cmd) runInner() bool {
// on Windows the shell is not used and command is started directly
// variables are replaced manually in order to guarantee compatibility
// with Linux commands
tmp := strings.ReplaceAll(e.cmdstr, "$RTSP_PATH", e.env.Path)
tmp = strings.ReplaceAll(tmp, "$RTSP_PORT", e.env.Port)
// On Windows, the shell is not used and command is started directly.
// Variables are replaced manually in order to guarantee compatibility
// with Linux commands.
tmp := e.cmdstr
for key, val := range e.env {
tmp = strings.ReplaceAll(tmp, "$"+key, val)
}
parts, err := shellquote.Split(tmp)
if err != nil {
return true
@ -24,10 +26,10 @@ func (e *Cmd) runInner() bool {
cmd := exec.Command(parts[0], parts[1:]...)
cmd.Env = append(os.Environ(),
"RTSP_PATH="+e.env.Path,
"RTSP_PORT="+e.env.Port,
)
cmd.Env = append([]string(nil), os.Environ()...)
for key, val := range e.env {
cmd.Env = append(cmd.Env, key+"="+val)
}
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
@ -45,8 +47,8 @@ func (e *Cmd) runInner() bool {
select {
case <-e.terminate:
// on Windows it's not possible to send os.Interrupt to a process
// Kill() is the only supported way
// on Windows, it's not possible to send os.Interrupt to a process.
// Kill() is the only supported way.
cmd.Process.Kill()
<-cmdDone
return false

33
rtsp-simple-server.yml
View file

@ -35,7 +35,8 @@ pprofAddress: 127.0.0.1:9999
# command to run when a client connects to the server.
# this is terminated with SIGINT when a client disconnects from the server.
# the server port is available in the RTSP_PORT variable.
# the following environment variables are available:
# * RTSP_PORT: server port
runOnConnect:
# the restart parameter allows to restart the command if it exits suddenly.
runOnConnectRestart: no
@ -116,7 +117,7 @@ hlsAllowOrigin: '*'
###############################################
# Path parameters
# these settings are path-dependent.
# these settings are path-dependent, and the map key is the name of the path.
# 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".
@ -195,8 +196,11 @@ paths:
# command to run when this path is initialized.
# this can be used to publish a stream and keep it always opened.
# this is terminated with SIGINT when the program closes.
# the path name is available in the RTSP_PATH variable.
# the server port is available in the RTSP_PORT variable.
# the following environment variables are available:
# * RTSP_PATH: path name
# * RTSP_PORT: server port
# * 1, 2, ...: regular expression groups, if path name is
# a regular expression.
runOnInit:
# the restart parameter allows to restart the command if it exits suddenly.
runOnInitRestart: no
@ -204,8 +208,11 @@ paths:
# command to run when this path is requested.
# this can be used to publish a stream on demand.
# this is terminated with SIGINT when the path is not requested anymore.
# the path name is available in the RTSP_PATH variable.
# the server port is available in the RTSP_PORT variable.
# the following environment variables are available:
# * RTSP_PATH: path name
# * RTSP_PORT: server port
# * 1, 2, ...: regular expression groups, if path name is
# a regular expression.
runOnDemand:
# the restart parameter allows to restart the command if it exits suddenly.
runOnDemandRestart: no
@ -218,16 +225,22 @@ paths:
# command to run when a client starts publishing.
# this is terminated with SIGINT when a client stops publishing.
# the path name is available in the RTSP_PATH variable.
# the server port is available in the RTSP_PORT variable.
# the following environment variables are available:
# * RTSP_PATH: path name
# * RTSP_PORT: server port
# * 1, 2, ...: regular expression groups, if path name is
# a regular expression.
runOnPublish:
# the restart parameter allows to restart the command if it exits suddenly.
runOnPublishRestart: no
# command to run when a clients starts reading.
# this is terminated with SIGINT when a client stops reading.
# the path name is available in the RTSP_PATH variable.
# the server port is available in the RTSP_PORT variable.
# the following environment variables are available:
# * RTSP_PATH: path name
# * RTSP_PORT: server port
# * 1, 2, ...: regular expression groups, if path name is
# a regular expression.
runOnRead:
# the restart parameter allows to restart the command if it exits suddenly.
runOnReadRestart: no