From a9385547ea910e7c8398655ed31d2f484dee68f2 Mon Sep 17 00:00:00 2001 From: aler9 <46489434+aler9@users.noreply.github.com> Date: Thu, 18 Feb 2021 23:26:45 +0100 Subject: [PATCH] add parameter readBufferCount to support non-compliant servers or publishers (#211) --- go.mod | 2 +- go.sum | 4 +- internal/conf/conf.go | 3 +- internal/path/path.go | 8 +++- internal/pathman/pathman.go | 72 +++++++++++++++-------------------- internal/sourcertsp/source.go | 8 +++- main.go | 2 + rtsp-simple-server.yml | 4 ++ 8 files changed, 54 insertions(+), 49 deletions(-) diff --git a/go.mod b/go.mod index 86067f9c..ea92340f 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.15 require ( github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d // indirect - github.com/aler9/gortsplib v0.0.0-20210215170952-6ce21a78419c + github.com/aler9/gortsplib v0.0.0-20210218215907-557fadcd3cf7 github.com/davecgh/go-spew v1.1.1 // indirect github.com/fsnotify/fsnotify v1.4.9 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 diff --git a/go.sum b/go.sum index 71f8ecd6..a7ec770e 100644 --- a/go.sum +++ b/go.sum @@ -2,8 +2,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafo github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d h1:UQZhZ2O0vMHr2cI+DC1Mbh0TJxzA3RcLoMsFw+aXw7E= github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho= -github.com/aler9/gortsplib v0.0.0-20210215170952-6ce21a78419c h1:ckOJiy0/+h6DmG7gAVvf5NI2Fl1AZAYIDf7DAUbDhl8= -github.com/aler9/gortsplib v0.0.0-20210215170952-6ce21a78419c/go.mod h1:8P09VjpiPJFyfkVosyF5/TY82jNwkMN165NS/7sc32I= +github.com/aler9/gortsplib v0.0.0-20210218215907-557fadcd3cf7 h1:2K6yaWw8OCqJB/V4wbcW7Kn8DGTwdpzNC8fmr9jlbwM= +github.com/aler9/gortsplib v0.0.0-20210218215907-557fadcd3cf7/go.mod h1:8P09VjpiPJFyfkVosyF5/TY82jNwkMN165NS/7sc32I= github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= diff --git a/internal/conf/conf.go b/internal/conf/conf.go index b78c4e64..9ac3716a 100644 --- a/internal/conf/conf.go +++ b/internal/conf/conf.go @@ -55,7 +55,7 @@ type Conf struct { ListenIP string `yaml:"listenIP"` ReadTimeout time.Duration `yaml:"readTimeout"` WriteTimeout time.Duration `yaml:"writeTimeout"` - ReadBufferCount uint64 `yaml:"readBufferCount"` + ReadBufferCount int `yaml:"readBufferCount"` Metrics bool `yaml:"metrics"` Pprof bool `yaml:"pprof"` RunOnConnect string `yaml:"runOnConnect"` @@ -73,6 +73,7 @@ type Conf struct { ServerCert string `yaml:"serverCert"` AuthMethods []string `yaml:"authMethods"` AuthMethodsParsed []headers.AuthMethod `yaml:"-" json:"-"` + ReadBufferSize int `yaml:"readBufferSize"` RTMPEnable bool `yaml:"rtmpEnable"` RTMPPort int `yaml:"rtmpPort"` diff --git a/internal/path/path.go b/internal/path/path.go index 9f8f7351..74498595 100644 --- a/internal/path/path.go +++ b/internal/path/path.go @@ -79,7 +79,8 @@ type Path struct { rtspPort int readTimeout time.Duration writeTimeout time.Duration - readBufferCount uint64 + readBufferCount int + readBufferSize int confName string conf *conf.PathConf name string @@ -124,7 +125,8 @@ func New( rtspPort int, readTimeout time.Duration, writeTimeout time.Duration, - readBufferCount uint64, + readBufferCount int, + readBufferSize int, confName string, conf *conf.PathConf, name string, @@ -137,6 +139,7 @@ func New( readTimeout: readTimeout, writeTimeout: writeTimeout, readBufferCount: readBufferCount, + readBufferSize: readBufferSize, confName: confName, conf: conf, name: name, @@ -427,6 +430,7 @@ func (pa *Path) startExternalSource() { pa.readTimeout, pa.writeTimeout, pa.readBufferCount, + pa.readBufferSize, &pa.sourceWg, pa.stats, pa) diff --git a/internal/pathman/pathman.go b/internal/pathman/pathman.go index 211c3e1d..50f0a2de 100644 --- a/internal/pathman/pathman.go +++ b/internal/pathman/pathman.go @@ -24,7 +24,8 @@ type PathManager struct { rtspPort int readTimeout time.Duration writeTimeout time.Duration - readBufferCount uint64 + readBufferCount int + readBufferSize int authMethods []headers.AuthMethod pathConfs map[string]*conf.PathConf stats *stats.Stats @@ -51,7 +52,8 @@ func New( rtspPort int, readTimeout time.Duration, writeTimeout time.Duration, - readBufferCount uint64, + readBufferCount int, + readBufferSize int, authMethods []headers.AuthMethod, pathConfs map[string]*conf.PathConf, stats *stats.Stats, @@ -62,6 +64,7 @@ func New( readTimeout: readTimeout, writeTimeout: writeTimeout, readBufferCount: readBufferCount, + readBufferSize: readBufferSize, authMethods: authMethods, pathConfs: pathConfs, stats: stats, @@ -162,18 +165,7 @@ outer: // create path if it doesn't exist if _, ok := pm.paths[req.PathName]; !ok { - pa := path.New( - pm.rtspPort, - pm.readTimeout, - pm.writeTimeout, - pm.readBufferCount, - pathName, - pathConf, - req.PathName, - &pm.wg, - pm.stats, - pm) - pm.paths[req.PathName] = pa + pm.paths[req.PathName] = pm.createPath(pathName, pathConf, req.PathName) } pm.paths[req.PathName].OnPathManDescribe(req) @@ -195,18 +187,7 @@ outer: // create path if it doesn't exist if _, ok := pm.paths[req.PathName]; !ok { - pa := path.New( - pm.rtspPort, - pm.readTimeout, - pm.writeTimeout, - pm.readBufferCount, - pathName, - pathConf, - req.PathName, - &pm.wg, - pm.stats, - pm) - pm.paths[req.PathName] = pa + pm.paths[req.PathName] = pm.createPath(pathName, pathConf, req.PathName) } pm.paths[req.PathName].OnPathManAnnounce(req) @@ -218,9 +199,13 @@ outer: continue } - err = req.Client.Authenticate(pm.authMethods, - req.PathName, pathConf.ReadIpsParsed, pathConf.ReadUser, - pathConf.ReadPass, req.Req) + err = req.Client.Authenticate( + pm.authMethods, + req.PathName, + pathConf.ReadIpsParsed, + pathConf.ReadUser, + pathConf.ReadPass, + req.Req) if err != nil { req.Res <- client.SetupPlayRes{nil, err} //nolint:govet continue @@ -233,6 +218,7 @@ outer: pm.readTimeout, pm.writeTimeout, pm.readBufferCount, + pm.readBufferSize, pathName, pathConf, req.PathName, @@ -296,21 +282,25 @@ outer: close(pm.clientSetupPlay) } +func (pm *PathManager) createPath(confName string, conf *conf.PathConf, name string) *path.Path { + return path.New( + pm.rtspPort, + pm.readTimeout, + pm.writeTimeout, + pm.readBufferCount, + pm.readBufferSize, + confName, + conf, + name, + &pm.wg, + pm.stats, + pm) +} + func (pm *PathManager) createPaths() { for pathName, pathConf := range pm.pathConfs { if _, ok := pm.paths[pathName]; !ok && pathConf.Regexp == nil { - pa := path.New( - pm.rtspPort, - pm.readTimeout, - pm.writeTimeout, - pm.readBufferCount, - pathName, - pathConf, - pathName, - &pm.wg, - pm.stats, - pm) - pm.paths[pathName] = pa + pm.paths[pathName] = pm.createPath(pathName, pathConf, pathName) } } } diff --git a/internal/sourcertsp/source.go b/internal/sourcertsp/source.go index 30f55af4..3fea9c0d 100644 --- a/internal/sourcertsp/source.go +++ b/internal/sourcertsp/source.go @@ -30,7 +30,8 @@ type Source struct { proto *gortsplib.StreamProtocol readTimeout time.Duration writeTimeout time.Duration - readBufferCount uint64 + readBufferCount int + readBufferSize int wg *sync.WaitGroup stats *stats.Stats parent Parent @@ -44,7 +45,8 @@ func New(ur string, proto *gortsplib.StreamProtocol, readTimeout time.Duration, writeTimeout time.Duration, - readBufferCount uint64, + readBufferCount int, + readBufferSize int, wg *sync.WaitGroup, stats *stats.Stats, parent Parent) *Source { @@ -54,6 +56,7 @@ func New(ur string, readTimeout: readTimeout, writeTimeout: writeTimeout, readBufferCount: readBufferCount, + readBufferSize: readBufferSize, wg: wg, stats: stats, parent: parent, @@ -125,6 +128,7 @@ func (s *Source) runInner() bool { ReadTimeout: s.readTimeout, WriteTimeout: s.writeTimeout, ReadBufferCount: s.readBufferCount, + ReadBufferSize: s.readBufferSize, OnRequest: func(req *base.Request) { s.log(logger.Debug, "c->s %v", req) }, diff --git a/main.go b/main.go index ffd0b172..f8d2b21d 100644 --- a/main.go +++ b/main.go @@ -277,6 +277,7 @@ func (p *program) createResources(initial bool) error { p.conf.ReadTimeout, p.conf.WriteTimeout, p.conf.ReadBufferCount, + p.conf.ReadBufferSize, p.conf.AuthMethodsParsed, p.conf.Paths, p.stats, @@ -380,6 +381,7 @@ func (p *program) closeResources(newConf *conf.Conf) { newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || newConf.ReadBufferCount != p.conf.ReadBufferCount || + newConf.ReadBufferSize != p.conf.ReadBufferSize || !reflect.DeepEqual(newConf.AuthMethodsParsed, p.conf.AuthMethodsParsed) { closePathMan = true } else if !reflect.DeepEqual(newConf.Paths, p.conf.Paths) { diff --git a/rtsp-simple-server.yml b/rtsp-simple-server.yml index a813ea31..76b48d85 100644 --- a/rtsp-simple-server.yml +++ b/rtsp-simple-server.yml @@ -58,6 +58,10 @@ serverKey: server.key serverCert: server.crt # authentication methods. authMethods: [basic, digest] +# read buffer size. +# this doesn't influence throughput and shouldn't be touched unless the server +# reports errors about the buffer size. +readBufferSize: 2048 ############################################### # RTMP options