add parameter readBufferCount to support non-compliant servers or publishers (#211)

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

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-20210218215907-557fadcd3cf7 h1:2K6yaWw8OCqJB/V4wbcW7Kn8DGTwdpzNC8fmr9jlbwM=
-github.com/aler9/gortsplib v0.0.0-20210215170952-6ce21a78419c/go.mod h1:8P09VjpiPJFyfkVosyF5/TY82jNwkMN165NS/7sc32I=
+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=

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"`

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)

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.paths[req.PathName] = pm.createPath(pathName, pathConf, req.PathName)
-					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].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.paths[req.PathName] = pm.createPath(pathName, pathConf, req.PathName)
-					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].OnPathManAnnounce(req)
@@ -218,9 +199,13 @@ outer:
 				continue
 			}
 
-			err = req.Client.Authenticate(pm.authMethods,
+			err = req.Client.Authenticate(
-				req.PathName, pathConf.ReadIpsParsed, pathConf.ReadUser,
+				pm.authMethods,
-				pathConf.ReadPass, req.Req)
+				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.paths[pathName] = pm.createPath(pathName, pathConf, pathName)
-				pm.rtspPort,
-				pm.readTimeout,
-				pm.writeTimeout,
-				pm.readBufferCount,
-				pathName,
-				pathConf,
-				pathName,
-				&pm.wg,
-				pm.stats,
-				pm)
-			pm.paths[pathName] = pa
 		}
 	}
 }

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)
 			},

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) {

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