Browse Source

support static sources in paths with regular expressions (#824) (#2799)

This allows to proxy requests to other servers by using regular
expressions.
pull/2801/head v1.4.0
Alessandro Ros 1 year ago committed by GitHub
parent
commit
d261bfe773
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 22
      README.md
  2. 33
      internal/conf/path.go
  3. 22
      internal/core/path.go
  4. 74
      internal/core/path_test.go
  5. 84
      internal/core/static_source_handler.go
  6. 7
      internal/staticsources/hls/source.go
  7. 7
      internal/staticsources/hls/source_test.go
  8. 9
      internal/staticsources/rtmp/source.go
  9. 19
      internal/staticsources/rtmp/source_test.go
  10. 3
      internal/staticsources/rtsp/source.go
  11. 14
      internal/staticsources/rtsp/source_test.go
  12. 7
      internal/staticsources/srt/source.go
  13. 11
      internal/staticsources/srt/source_test.go
  14. 7
      internal/staticsources/udp/source.go
  15. 9
      internal/staticsources/udp/source_test.go
  16. 7
      internal/staticsources/webrtc/source.go
  17. 9
      internal/staticsources/webrtc/source_test.go
  18. 2
      mediamtx.yml

22
README.md

@ -115,7 +115,8 @@ _rtsp-simple-server_ has been rebranded as _MediaMTX_. The reason is pretty obvi @@ -115,7 +115,8 @@ _rtsp-simple-server_ has been rebranded as _MediaMTX_. The reason is pretty obvi
* [Encrypt the configuration](#encrypt-the-configuration)
* [Remuxing, re-encoding, compression](#remuxing-re-encoding-compression)
* [Record streams to disk](#record-streams-to-disk)
* [Forward streams to another server](#forward-streams-to-another-server)
* [Forward streams to other servers](#forward-streams-to-other-servers)
* [Proxy requests to other servers](#proxy-requests-to-other-servers)
* [On-demand publishing](#on-demand-publishing)
* [Start on boot](#start-on-boot)
* [Linux](#linux)
@ -1164,7 +1165,7 @@ To upload recordings to a remote location, you can use _MediaMTX_ together with @@ -1164,7 +1165,7 @@ To upload recordings to a remote location, you can use _MediaMTX_ together with
If you want to delete local segments after they are uploaded, replace `rclone sync` with `rclone move`.
### Forward streams to another server
### Forward streams to other servers
To forward incoming streams to another server, use _FFmpeg_ inside the `runOnReady` parameter:
@ -1173,10 +1174,25 @@ pathDefaults: @@ -1173,10 +1174,25 @@ pathDefaults:
runOnReady: >
ffmpeg -i rtsp://localhost:$RTSP_PORT/$MTX_PATH
-c copy
-f rtsp rtsp://another-server/another-path
-f rtsp rtsp://other-server:8554/another-path
runOnReadyRestart: yes
```
### Proxy requests to other servers
The server allows to proxy incoming requests to other servers or cameras. This is useful to expose servers or cameras behind a NAT. Edit `mediamtx.yml` and replace everything inside section `paths` with the following content:
```yml
paths:
"~^proxy_(.+)$":
# If path name is a regular expression, $G1, G2, etc will be replaced
# with regular expression groups.
source: rtsp://other-server:8554/$G1
sourceOnDemand: yes
```
All requests addressed to `rtsp://server:8854/proxy_a` will be forwarded to `rtsp://other-server:8854/a` and so on.
### On-demand publishing
Edit `mediamtx.yml` and replace everything inside section `paths` with the following content:

33
internal/conf/path.go

@ -235,15 +235,16 @@ func (pconf *Path) check(conf *Conf, name string) error { @@ -235,15 +235,16 @@ func (pconf *Path) check(conf *Conf, name string) error {
// General
if pconf.Source != "publisher" && pconf.Source != "redirect" &&
pconf.Regexp != nil && !pconf.SourceOnDemand {
return fmt.Errorf("a path with a regular expression (or path 'all') and a static source" +
" must have 'sourceOnDemand' set to true")
}
switch {
case pconf.Source == "publisher":
case strings.HasPrefix(pconf.Source, "rtsp://") ||
strings.HasPrefix(pconf.Source, "rtsps://"):
if pconf.Regexp != nil {
return fmt.Errorf("a path with a regular expression (or path 'all') cannot have a RTSP source. use another path")
}
_, err := base.ParseURL(pconf.Source)
if err != nil {
return fmt.Errorf("'%s' is not a valid URL", pconf.Source)
@ -251,10 +252,6 @@ func (pconf *Path) check(conf *Conf, name string) error { @@ -251,10 +252,6 @@ func (pconf *Path) check(conf *Conf, name string) error {
case strings.HasPrefix(pconf.Source, "rtmp://") ||
strings.HasPrefix(pconf.Source, "rtmps://"):
if pconf.Regexp != nil {
return fmt.Errorf("a path with a regular expression (or path 'all') cannot have a RTMP source. use another path")
}
u, err := gourl.Parse(pconf.Source)
if err != nil {
return fmt.Errorf("'%s' is not a valid URL", pconf.Source)
@ -271,10 +268,6 @@ func (pconf *Path) check(conf *Conf, name string) error { @@ -271,10 +268,6 @@ func (pconf *Path) check(conf *Conf, name string) error {
case strings.HasPrefix(pconf.Source, "http://") ||
strings.HasPrefix(pconf.Source, "https://"):
if pconf.Regexp != nil {
return fmt.Errorf("a path with a regular expression (or path 'all') cannot have a HLS source. use another path")
}
u, err := gourl.Parse(pconf.Source)
if err != nil {
return fmt.Errorf("'%s' is not a valid URL", pconf.Source)
@ -293,19 +286,12 @@ func (pconf *Path) check(conf *Conf, name string) error { @@ -293,19 +286,12 @@ func (pconf *Path) check(conf *Conf, name string) error {
}
case strings.HasPrefix(pconf.Source, "udp://"):
if pconf.Regexp != nil {
return fmt.Errorf("a path with a regular expression (or path 'all') cannot have a HLS source. use another path")
}
_, _, err := net.SplitHostPort(pconf.Source[len("udp://"):])
if err != nil {
return fmt.Errorf("'%s' is not a valid UDP URL", pconf.Source)
}
case strings.HasPrefix(pconf.Source, "srt://"):
if pconf.Regexp != nil {
return fmt.Errorf("a path with a regular expression (or path 'all') cannot have a SRT source. use another path")
}
_, err := gourl.Parse(pconf.Source)
if err != nil {
@ -314,11 +300,6 @@ func (pconf *Path) check(conf *Conf, name string) error { @@ -314,11 +300,6 @@ func (pconf *Path) check(conf *Conf, name string) error {
case strings.HasPrefix(pconf.Source, "whep://") ||
strings.HasPrefix(pconf.Source, "wheps://"):
if pconf.Regexp != nil {
return fmt.Errorf("a path with a regular expression (or path 'all') " +
"cannot have a WebRTC/WHEP source. use another path")
}
_, err := gourl.Parse(pconf.Source)
if err != nil {
return fmt.Errorf("'%s' is not a valid URL", pconf.Source)
@ -327,10 +308,6 @@ func (pconf *Path) check(conf *Conf, name string) error { @@ -327,10 +308,6 @@ func (pconf *Path) check(conf *Conf, name string) error {
case pconf.Source == "redirect":
case pconf.Source == "rpiCamera":
if pconf.Regexp != nil {
return fmt.Errorf(
"a path with a regular expression (or path 'all') cannot have 'rpiCamera' as source. use another path")
}
default:
return fmt.Errorf("invalid source: '%s'", pconf.Source)

22
internal/core/path.go

@ -197,12 +197,22 @@ func (pa *path) run() { @@ -197,12 +197,22 @@ func (pa *path) run() {
if pa.conf.Source == "redirect" {
pa.source = &sourceRedirect{}
} else if pa.conf.HasStaticSource() {
pa.source = newStaticSourceHandler(
pa.conf,
pa.readTimeout,
pa.writeTimeout,
pa.writeQueueSize,
pa)
resolvedSource := pa.conf.Source
if len(pa.matches) > 1 {
for i, ma := range pa.matches[1:] {
resolvedSource = strings.ReplaceAll(resolvedSource, "$G"+strconv.FormatInt(int64(i+1), 10), ma)
}
}
pa.source = &staticSourceHandler{
conf: pa.conf,
readTimeout: pa.readTimeout,
writeTimeout: pa.writeTimeout,
writeQueueSize: pa.writeQueueSize,
resolvedSource: resolvedSource,
parent: pa,
}
pa.source.(*staticSourceHandler).initialize()
if !pa.conf.SourceOnDemand {
pa.source.(*staticSourceHandler).start(false)

74
internal/core/path_test.go

@ -82,6 +82,25 @@ func main() { @@ -82,6 +82,25 @@ func main() {
}
`
type testServer struct {
onDescribe func(*gortsplib.ServerHandlerOnDescribeCtx) (*base.Response, *gortsplib.ServerStream, error)
onSetup func(*gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error)
onPlay func(*gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error)
}
func (sh *testServer) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx,
) (*base.Response, *gortsplib.ServerStream, error) {
return sh.onDescribe(ctx)
}
func (sh *testServer) OnSetup(ctx *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error) {
return sh.onSetup(ctx)
}
func (sh *testServer) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) {
return sh.onPlay(ctx)
}
var _ defs.Path = &path{}
func TestPathRunOnDemand(t *testing.T) {
@ -573,3 +592,58 @@ func TestPathFallback(t *testing.T) { @@ -573,3 +592,58 @@ func TestPathFallback(t *testing.T) {
})
}
}
func TestPathSourceRegexp(t *testing.T) {
var stream *gortsplib.ServerStream
s := gortsplib.Server{
Handler: &testServer{
onDescribe: func(ctx *gortsplib.ServerHandlerOnDescribeCtx,
) (*base.Response, *gortsplib.ServerStream, error) {
require.Equal(t, "/a", ctx.Path)
return &base.Response{
StatusCode: base.StatusOK,
}, stream, nil
},
onSetup: func(ctx *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, stream, nil
},
onPlay: func(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil
},
},
RTSPAddress: "127.0.0.1:8555",
}
err := s.Start()
require.NoError(t, err)
defer s.Close()
stream = gortsplib.NewServerStream(&s, &description.Session{Medias: []*description.Media{testMediaH264}})
defer stream.Close()
p, ok := newInstance(
"paths:\n" +
" '~^test_(.+)$':\n" +
" source: rtsp://127.0.0.1:8555/$G1\n" +
" sourceOnDemand: yes\n" +
" 'all':\n")
require.Equal(t, true, ok)
defer p.Close()
reader := gortsplib.Client{}
u, err := base.ParseURL("rtsp://127.0.0.1:8554/test_a")
require.NoError(t, err)
err = reader.Start(u.Scheme, u.Host)
require.NoError(t, err)
defer reader.Close()
_, _, err = reader.Describe(u)
require.NoError(t, err)
}

84
internal/core/static_source_handler.go

@ -30,8 +30,12 @@ type staticSourceHandlerParent interface { @@ -30,8 +30,12 @@ type staticSourceHandlerParent interface {
// staticSourceHandler is a static source handler.
type staticSourceHandler struct {
conf *conf.Path
parent staticSourceHandlerParent
conf *conf.Path
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
writeQueueSize int
resolvedSource string
parent staticSourceHandlerParent
ctx context.Context
ctxCancel func()
@ -47,72 +51,66 @@ type staticSourceHandler struct { @@ -47,72 +51,66 @@ type staticSourceHandler struct {
done chan struct{}
}
func newStaticSourceHandler(
cnf *conf.Path,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
writeQueueSize int,
parent staticSourceHandlerParent,
) *staticSourceHandler {
s := &staticSourceHandler{
conf: cnf,
parent: parent,
chReloadConf: make(chan *conf.Path),
chInstanceSetReady: make(chan defs.PathSourceStaticSetReadyReq),
chInstanceSetNotReady: make(chan defs.PathSourceStaticSetNotReadyReq),
}
func (s *staticSourceHandler) initialize() {
s.chReloadConf = make(chan *conf.Path)
s.chInstanceSetReady = make(chan defs.PathSourceStaticSetReadyReq)
s.chInstanceSetNotReady = make(chan defs.PathSourceStaticSetNotReadyReq)
switch {
case strings.HasPrefix(cnf.Source, "rtsp://") ||
strings.HasPrefix(cnf.Source, "rtsps://"):
case strings.HasPrefix(s.resolvedSource, "rtsp://") ||
strings.HasPrefix(s.resolvedSource, "rtsps://"):
s.instance = &rtspsource.Source{
ReadTimeout: readTimeout,
WriteTimeout: writeTimeout,
WriteQueueSize: writeQueueSize,
ResolvedSource: s.resolvedSource,
ReadTimeout: s.readTimeout,
WriteTimeout: s.writeTimeout,
WriteQueueSize: s.writeQueueSize,
Parent: s,
}
case strings.HasPrefix(cnf.Source, "rtmp://") ||
strings.HasPrefix(cnf.Source, "rtmps://"):
case strings.HasPrefix(s.resolvedSource, "rtmp://") ||
strings.HasPrefix(s.resolvedSource, "rtmps://"):
s.instance = &rtmpsource.Source{
ReadTimeout: readTimeout,
WriteTimeout: writeTimeout,
Parent: s,
ResolvedSource: s.resolvedSource,
ReadTimeout: s.readTimeout,
WriteTimeout: s.writeTimeout,
Parent: s,
}
case strings.HasPrefix(cnf.Source, "http://") ||
strings.HasPrefix(cnf.Source, "https://"):
case strings.HasPrefix(s.resolvedSource, "http://") ||
strings.HasPrefix(s.resolvedSource, "https://"):
s.instance = &hlssource.Source{
ReadTimeout: readTimeout,
Parent: s,
ResolvedSource: s.resolvedSource,
ReadTimeout: s.readTimeout,
Parent: s,
}
case strings.HasPrefix(cnf.Source, "udp://"):
case strings.HasPrefix(s.resolvedSource, "udp://"):
s.instance = &udpsource.Source{
ReadTimeout: readTimeout,
Parent: s,
ResolvedSource: s.resolvedSource,
ReadTimeout: s.readTimeout,
Parent: s,
}
case strings.HasPrefix(cnf.Source, "srt://"):
case strings.HasPrefix(s.resolvedSource, "srt://"):
s.instance = &srtsource.Source{
ReadTimeout: readTimeout,
Parent: s,
ResolvedSource: s.resolvedSource,
ReadTimeout: s.readTimeout,
Parent: s,
}
case strings.HasPrefix(cnf.Source, "whep://") ||
strings.HasPrefix(cnf.Source, "wheps://"):
case strings.HasPrefix(s.resolvedSource, "whep://") ||
strings.HasPrefix(s.resolvedSource, "wheps://"):
s.instance = &webrtcsource.Source{
ReadTimeout: readTimeout,
Parent: s,
ResolvedSource: s.resolvedSource,
ReadTimeout: s.readTimeout,
Parent: s,
}
case cnf.Source == "rpiCamera":
case s.resolvedSource == "rpiCamera":
s.instance = &rpicamerasource.Source{
Parent: s,
}
}
return s
}
func (s *staticSourceHandler) close(reason string) {

7
internal/staticsources/hls/source.go

@ -20,8 +20,9 @@ import ( @@ -20,8 +20,9 @@ import (
// Source is a HLS static source.
type Source struct {
ReadTimeout conf.StringDuration
Parent defs.StaticSourceParent
ResolvedSource string
ReadTimeout conf.StringDuration
Parent defs.StaticSourceParent
}
// Log implements logger.Writer.
@ -43,7 +44,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { @@ -43,7 +44,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
var c *gohlslib.Client
c = &gohlslib.Client{
URI: params.Conf.Source,
URI: s.ResolvedSource,
HTTPClient: &http.Client{
Timeout: time.Duration(s.ReadTimeout),
Transport: &http.Transport{

7
internal/staticsources/hls/source_test.go

@ -104,12 +104,11 @@ func TestSource(t *testing.T) { @@ -104,12 +104,11 @@ func TestSource(t *testing.T) {
te := tester.New(
func(p defs.StaticSourceParent) defs.StaticSource {
return &Source{
Parent: p,
ResolvedSource: "http://localhost:5780/stream.m3u8",
Parent: p,
}
},
&conf.Path{
Source: "http://localhost:5780/stream.m3u8",
},
&conf.Path{},
)
defer te.Close()

9
internal/staticsources/rtmp/source.go

@ -23,9 +23,10 @@ import ( @@ -23,9 +23,10 @@ import (
// Source is a RTMP static source.
type Source struct {
ReadTimeout conf.StringDuration
WriteTimeout conf.StringDuration
Parent defs.StaticSourceParent
ResolvedSource string
ReadTimeout conf.StringDuration
WriteTimeout conf.StringDuration
Parent defs.StaticSourceParent
}
// Log implements logger.Writer.
@ -37,7 +38,7 @@ func (s *Source) Log(level logger.Level, format string, args ...interface{}) { @@ -37,7 +38,7 @@ func (s *Source) Log(level logger.Level, format string, args ...interface{}) {
func (s *Source) Run(params defs.StaticSourceRunParams) error {
s.Log(logger.Debug, "connecting")
u, err := url.Parse(params.Conf.Source)
u, err := url.Parse(s.ResolvedSource)
if err != nil {
return err
}

19
internal/staticsources/rtmp/source_test.go

@ -156,26 +156,25 @@ func TestSource(t *testing.T) { @@ -156,26 +156,25 @@ func TestSource(t *testing.T) {
te = tester.New(
func(p defs.StaticSourceParent) defs.StaticSource {
return &Source{
ReadTimeout: conf.StringDuration(10 * time.Second),
WriteTimeout: conf.StringDuration(10 * time.Second),
Parent: p,
ResolvedSource: "rtmp://localhost:1937/teststream",
ReadTimeout: conf.StringDuration(10 * time.Second),
WriteTimeout: conf.StringDuration(10 * time.Second),
Parent: p,
}
},
&conf.Path{
Source: "rtmp://localhost:1937/teststream",
},
&conf.Path{},
)
} else {
te = tester.New(
func(p defs.StaticSourceParent) defs.StaticSource {
return &Source{
ReadTimeout: conf.StringDuration(10 * time.Second),
WriteTimeout: conf.StringDuration(10 * time.Second),
Parent: p,
ResolvedSource: "rtmps://localhost:1937/teststream",
ReadTimeout: conf.StringDuration(10 * time.Second),
WriteTimeout: conf.StringDuration(10 * time.Second),
Parent: p,
}
},
&conf.Path{
Source: "rtmps://localhost:1937/teststream",
SourceFingerprint: "33949E05FFFB5FF3E8AA16F8213A6251B4D9363804BA53233C4DA9A46D6F2739",
},
)

3
internal/staticsources/rtsp/source.go

@ -62,6 +62,7 @@ func createRangeHeader(cnf *conf.Path) (*headers.Range, error) { @@ -62,6 +62,7 @@ func createRangeHeader(cnf *conf.Path) (*headers.Range, error) {
// Source is a RTSP static source.
type Source struct {
ResolvedSource string
ReadTimeout conf.StringDuration
WriteTimeout conf.StringDuration
WriteQueueSize int
@ -103,7 +104,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { @@ -103,7 +104,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
},
}
u, err := base.ParseURL(params.Conf.Source)
u, err := base.ParseURL(s.ResolvedSource)
if err != nil {
return err
}

14
internal/staticsources/rtsp/source_test.go

@ -201,7 +201,6 @@ func TestSource(t *testing.T) { @@ -201,7 +201,6 @@ func TestSource(t *testing.T) {
err = s.Start()
require.NoError(t, err)
defer s.Wait() //nolint:errcheck
defer s.Close()
stream = gortsplib.NewServerStream(&s, &description.Session{Medias: []*description.Media{testMediaH264}})
@ -216,6 +215,7 @@ func TestSource(t *testing.T) { @@ -216,6 +215,7 @@ func TestSource(t *testing.T) {
te = tester.New(
func(p defs.StaticSourceParent) defs.StaticSource {
return &Source{
ResolvedSource: "rtsp://testuser:testpass@localhost:8555/teststream",
ReadTimeout: conf.StringDuration(10 * time.Second),
WriteTimeout: conf.StringDuration(10 * time.Second),
WriteQueueSize: 2048,
@ -223,7 +223,6 @@ func TestSource(t *testing.T) { @@ -223,7 +223,6 @@ func TestSource(t *testing.T) {
}
},
&conf.Path{
Source: "rtsp://testuser:testpass@localhost:8555/teststream",
RTSPTransport: sp,
},
)
@ -231,6 +230,7 @@ func TestSource(t *testing.T) { @@ -231,6 +230,7 @@ func TestSource(t *testing.T) {
te = tester.New(
func(p defs.StaticSourceParent) defs.StaticSource {
return &Source{
ResolvedSource: "rtsps://testuser:testpass@localhost:8555/teststream",
ReadTimeout: conf.StringDuration(10 * time.Second),
WriteTimeout: conf.StringDuration(10 * time.Second),
WriteQueueSize: 2048,
@ -238,7 +238,6 @@ func TestSource(t *testing.T) { @@ -238,7 +238,6 @@ func TestSource(t *testing.T) {
}
},
&conf.Path{
Source: "rtsps://testuser:testpass@localhost:8555/teststream",
SourceFingerprint: "33949E05FFFB5FF3E8AA16F8213A6251B4D9363804BA53233C4DA9A46D6F2739",
},
)
@ -306,7 +305,6 @@ func TestRTSPSourceNoPassword(t *testing.T) { @@ -306,7 +305,6 @@ func TestRTSPSourceNoPassword(t *testing.T) {
err = s.Start()
require.NoError(t, err)
defer s.Wait() //nolint:errcheck
defer s.Close()
stream = gortsplib.NewServerStream(&s, &description.Session{Medias: []*description.Media{testMediaH264}})
@ -318,6 +316,7 @@ func TestRTSPSourceNoPassword(t *testing.T) { @@ -318,6 +316,7 @@ func TestRTSPSourceNoPassword(t *testing.T) {
te := tester.New(
func(p defs.StaticSourceParent) defs.StaticSource {
return &Source{
ResolvedSource: "rtsp://testuser:@127.0.0.1:8555/teststream",
ReadTimeout: conf.StringDuration(10 * time.Second),
WriteTimeout: conf.StringDuration(10 * time.Second),
WriteQueueSize: 2048,
@ -325,7 +324,6 @@ func TestRTSPSourceNoPassword(t *testing.T) { @@ -325,7 +324,6 @@ func TestRTSPSourceNoPassword(t *testing.T) {
}
},
&conf.Path{
Source: "rtsp://testuser:@127.0.0.1:8555/teststream",
RTSPTransport: sp,
},
)
@ -389,15 +387,12 @@ func TestRTSPSourceRange(t *testing.T) { @@ -389,15 +387,12 @@ func TestRTSPSourceRange(t *testing.T) {
err := s.Start()
require.NoError(t, err)
defer s.Wait() //nolint:errcheck
defer s.Close()
stream = gortsplib.NewServerStream(&s, &description.Session{Medias: []*description.Media{testMediaH264}})
defer stream.Close()
cnf := &conf.Path{
Source: "rtsp://127.0.0.1:8555/teststream",
}
cnf := &conf.Path{}
switch ca {
case "clock":
@ -416,6 +411,7 @@ func TestRTSPSourceRange(t *testing.T) { @@ -416,6 +411,7 @@ func TestRTSPSourceRange(t *testing.T) {
te := tester.New(
func(p defs.StaticSourceParent) defs.StaticSource {
return &Source{
ResolvedSource: "rtsp://127.0.0.1:8555/teststream",
ReadTimeout: conf.StringDuration(10 * time.Second),
WriteTimeout: conf.StringDuration(10 * time.Second),
WriteQueueSize: 2048,

7
internal/staticsources/srt/source.go

@ -17,8 +17,9 @@ import ( @@ -17,8 +17,9 @@ import (
// Source is a SRT static source.
type Source struct {
ReadTimeout conf.StringDuration
Parent defs.StaticSourceParent
ResolvedSource string
ReadTimeout conf.StringDuration
Parent defs.StaticSourceParent
}
// Log implements logger.Writer.
@ -31,7 +32,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { @@ -31,7 +32,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
s.Log(logger.Debug, "connecting")
conf := srt.DefaultConfig()
address, err := conf.UnmarshalURL(params.Conf.Source)
address, err := conf.UnmarshalURL(s.ResolvedSource)
if err != nil {
return err
}

11
internal/staticsources/srt/source_test.go

@ -6,7 +6,7 @@ import ( @@ -6,7 +6,7 @@ import (
"time"
"github.com/bluenviron/mediacommon/pkg/formats/mpegts"
"github.com/datarhei/gosrt"
srt "github.com/datarhei/gosrt"
"github.com/stretchr/testify/require"
"github.com/bluenviron/mediamtx/internal/conf"
@ -59,13 +59,12 @@ func TestSource(t *testing.T) { @@ -59,13 +59,12 @@ func TestSource(t *testing.T) {
te := tester.New(
func(p defs.StaticSourceParent) defs.StaticSource {
return &Source{
ReadTimeout: conf.StringDuration(10 * time.Second),
Parent: p,
ResolvedSource: "srt://localhost:9002?streamid=sidname&passphrase=ttest1234567",
ReadTimeout: conf.StringDuration(10 * time.Second),
Parent: p,
}
},
&conf.Path{
Source: "srt://localhost:9002?streamid=sidname&passphrase=ttest1234567",
},
&conf.Path{},
)
defer te.Close()

7
internal/staticsources/udp/source.go

@ -45,8 +45,9 @@ type packetConn interface { @@ -45,8 +45,9 @@ type packetConn interface {
// Source is a UDP static source.
type Source struct {
ReadTimeout conf.StringDuration
Parent defs.StaticSourceParent
ResolvedSource string
ReadTimeout conf.StringDuration
Parent defs.StaticSourceParent
}
// Log implements logger.Writer.
@ -58,7 +59,7 @@ func (s *Source) Log(level logger.Level, format string, args ...interface{}) { @@ -58,7 +59,7 @@ func (s *Source) Log(level logger.Level, format string, args ...interface{}) {
func (s *Source) Run(params defs.StaticSourceRunParams) error {
s.Log(logger.Debug, "connecting")
hostPort := params.Conf.Source[len("udp://"):]
hostPort := s.ResolvedSource[len("udp://"):]
addr, err := net.ResolveUDPAddr("udp", hostPort)
if err != nil {

9
internal/staticsources/udp/source_test.go

@ -18,13 +18,12 @@ func TestSource(t *testing.T) { @@ -18,13 +18,12 @@ func TestSource(t *testing.T) {
te := tester.New(
func(p defs.StaticSourceParent) defs.StaticSource {
return &Source{
ReadTimeout: conf.StringDuration(10 * time.Second),
Parent: p,
ResolvedSource: "udp://localhost:9001",
ReadTimeout: conf.StringDuration(10 * time.Second),
Parent: p,
}
},
&conf.Path{
Source: "udp://localhost:9001",
},
&conf.Path{},
)
defer te.Close()

7
internal/staticsources/webrtc/source.go

@ -19,8 +19,9 @@ import ( @@ -19,8 +19,9 @@ import (
// Source is a WebRTC static source.
type Source struct {
ReadTimeout conf.StringDuration
Parent defs.StaticSourceParent
ResolvedSource string
ReadTimeout conf.StringDuration
Parent defs.StaticSourceParent
}
// Log implements logger.Writer.
@ -32,7 +33,7 @@ func (s *Source) Log(level logger.Level, format string, args ...interface{}) { @@ -32,7 +33,7 @@ func (s *Source) Log(level logger.Level, format string, args ...interface{}) {
func (s *Source) Run(params defs.StaticSourceRunParams) error {
s.Log(logger.Debug, "connecting")
u, err := url.Parse(params.Conf.Source)
u, err := url.Parse(s.ResolvedSource)
if err != nil {
return err
}

9
internal/staticsources/webrtc/source_test.go

@ -127,13 +127,12 @@ func TestSource(t *testing.T) { @@ -127,13 +127,12 @@ func TestSource(t *testing.T) {
te := tester.New(
func(p defs.StaticSourceParent) defs.StaticSource {
return &Source{
ReadTimeout: conf.StringDuration(10 * time.Second),
Parent: p,
ResolvedSource: "whep://localhost:9003/my/resource",
ReadTimeout: conf.StringDuration(10 * time.Second),
Parent: p,
}
},
&conf.Path{
Source: "whep://localhost:9003/my/resource",
},
&conf.Path{},
)
defer te.Close()

2
mediamtx.yml

@ -265,6 +265,8 @@ pathDefaults: @@ -265,6 +265,8 @@ pathDefaults:
# * wheps://existing-url -> the stream is pulled from another WebRTC server / camera with HTTPS
# * redirect -> the stream is provided by another path or server
# * rpiCamera -> the stream is provided by a Raspberry Pi Camera
# If path name is a regular expression, $G1, G2, etc will be replaced
# with regular expression groups.
source: publisher
# If the source is a URL, and the source certificate is self-signed
# or invalid, you can provide the fingerprint of the certificate in order to

Loading…
Cancel
Save