Browse Source

update gortsplib

pull/340/head
aler9 5 years ago
parent
commit
6e64b4be22
  1. 2
      go.mod
  2. 4
      go.sum
  3. 37
      internal/clientrtsp/client.go
  4. 5
      internal/externalcmd/cmd.go
  5. 68
      internal/path/path.go
  6. 17
      internal/sourcertmp/source.go
  7. 19
      internal/sourcertsp/source.go

2
go.mod

@ -5,7 +5,7 @@ go 1.15
require ( require (
github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect
github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d // indirect github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d // indirect
github.com/aler9/gortsplib v0.0.0-20210313202643-32c10cfb66cd github.com/aler9/gortsplib v0.0.0-20210314154849-d902b7da9320
github.com/davecgh/go-spew v1.1.1 // indirect github.com/davecgh/go-spew v1.1.1 // indirect
github.com/fsnotify/fsnotify v1.4.9 github.com/fsnotify/fsnotify v1.4.9
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51

4
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/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 h1:UQZhZ2O0vMHr2cI+DC1Mbh0TJxzA3RcLoMsFw+aXw7E=
github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho= github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
github.com/aler9/gortsplib v0.0.0-20210313202643-32c10cfb66cd h1:+9AYNCIlkuZF3d3OOqDRC9D+bLdyrDPiVi8q+gmq8mQ= github.com/aler9/gortsplib v0.0.0-20210314154849-d902b7da9320 h1:/WOt00YtNY2eWQy+MZ+LkkP7+XMviepr4yLCrK5PzhE=
github.com/aler9/gortsplib v0.0.0-20210313202643-32c10cfb66cd/go.mod h1:aj4kDzanb3JZ46sFywWShcsnqqXTLE/3PNjwDhQZGM0= github.com/aler9/gortsplib v0.0.0-20210314154849-d902b7da9320/go.mod h1:aj4kDzanb3JZ46sFywWShcsnqqXTLE/3PNjwDhQZGM0=
github.com/aler9/rtmp v0.0.0-20210309202041-2d7177b7300d h1:LAX8pNvYpGgFpKdbPpEZWjNkHbmyvjMrT3vO7s7aaKU= github.com/aler9/rtmp v0.0.0-20210309202041-2d7177b7300d h1:LAX8pNvYpGgFpKdbPpEZWjNkHbmyvjMrT3vO7s7aaKU=
github.com/aler9/rtmp v0.0.0-20210309202041-2d7177b7300d/go.mod h1:vzuE21rowz+lT1NGsWbreIvYulgBpCGnQyeTyFblUHc= github.com/aler9/rtmp v0.0.0-20210309202041-2d7177b7300d/go.mod h1:vzuE21rowz+lT1NGsWbreIvYulgBpCGnQyeTyFblUHc=
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=

37
internal/clientrtsp/client.go

@ -254,7 +254,7 @@ func (c *Client) run() {
}, nil }, nil
} }
onSetup := func(req *base.Request, th *headers.Transport, trackID int) (*base.Response, error) { onSetup := func(req *base.Request, th *headers.Transport, reqPath string, trackID int) (*base.Response, error) {
if th.Protocol == gortsplib.StreamProtocolUDP { if th.Protocol == gortsplib.StreamProtocolUDP {
if _, ok := c.protocols[gortsplib.StreamProtocolUDP]; !ok { if _, ok := c.protocols[gortsplib.StreamProtocolUDP]; !ok {
return &base.Response{ return &base.Response{
@ -271,26 +271,6 @@ func (c *Client) run() {
switch c.conn.State() { switch c.conn.State() {
case gortsplib.ServerConnStateInitial, gortsplib.ServerConnStatePrePlay: // play case gortsplib.ServerConnStateInitial, gortsplib.ServerConnStatePrePlay: // play
pathAndQuery, ok := req.URL.RTSPPathAndQuery()
if !ok {
return &base.Response{
StatusCode: base.StatusBadRequest,
}, fmt.Errorf("invalid path (%s)", req.URL)
}
_, pathAndQuery, ok = base.PathSplitControlAttribute(pathAndQuery)
if !ok {
return &base.Response{
StatusCode: base.StatusBadRequest,
}, fmt.Errorf("invalid path (%s)", req.URL)
}
reqPath, _ := base.PathSplitQuery(pathAndQuery)
// path can end with a slash, remove it
// this is needed to support reading mpegts with ffmpeg
reqPath = strings.TrimSuffix(reqPath, "/")
if c.path != nil && reqPath != c.path.Name() { if c.path != nil && reqPath != c.path.Name() {
return &base.Response{ return &base.Response{
StatusCode: base.StatusBadRequest, StatusCode: base.StatusBadRequest,
@ -333,21 +313,6 @@ func (c *Client) run() {
StatusCode: base.StatusBadRequest, StatusCode: base.StatusBadRequest,
}, fmt.Errorf("track %d does not exist", trackID) }, fmt.Errorf("track %d does not exist", trackID)
} }
default: // record
reqPathAndQuery, ok := req.URL.RTSPPathAndQuery()
if !ok {
return &base.Response{
StatusCode: base.StatusBadRequest,
}, fmt.Errorf("invalid path (%s)", req.URL)
}
if !strings.HasPrefix(reqPathAndQuery, c.path.Name()) {
return &base.Response{
StatusCode: base.StatusBadRequest,
}, fmt.Errorf("invalid path: must begin with '%s', but is '%s'",
c.path.Name(), reqPathAndQuery)
}
} }
return &base.Response{ return &base.Response{

5
internal/externalcmd/cmd.go

@ -62,11 +62,8 @@ func (e *Cmd) run() {
return false return false
} }
t := time.NewTimer(retryPause)
defer t.Stop()
select { select {
case <-t.C: case <-time.After(retryPause):
return true return true
case <-e.terminate: case <-e.terminate:
return false return false

68
internal/path/path.go

@ -38,10 +38,10 @@ type source interface {
IsSource() IsSource()
} }
// sourceExternal is implemented by all source*. // extSource is implemented by all external sources.
type sourceExternal interface { type extSource interface {
IsSource() IsSource()
IsSourceExternal() IsExtSource()
Close() Close()
} }
@ -49,6 +49,10 @@ type sourceRedirect struct{}
func (*sourceRedirect) IsSource() {} func (*sourceRedirect) IsSource() {}
type extSourceSetReadyReq struct {
tracks gortsplib.Tracks
}
type clientState int type clientState int
const ( const (
@ -100,16 +104,16 @@ type Path struct {
closeTimerStarted bool closeTimerStarted bool
// in // in
sourceSetReady chan struct{} // from source extSourceSetReady chan extSourceSetReadyReq // from external source
sourceSetNotReady chan struct{} // from source extSourceSetNotReady chan struct{} // from external source
clientDescribe chan client.DescribeReq clientDescribe chan client.DescribeReq
clientSetupPlay chan client.SetupPlayReq clientSetupPlay chan client.SetupPlayReq
clientAnnounce chan client.AnnounceReq clientAnnounce chan client.AnnounceReq
clientPlay chan client.PlayReq clientPlay chan client.PlayReq
clientRecord chan client.RecordReq clientRecord chan client.RecordReq
clientPause chan client.PauseReq clientPause chan client.PauseReq
clientRemove chan client.RemoveReq clientRemove chan client.RemoveReq
terminate chan struct{} terminate chan struct{}
} }
// New allocates a Path. // New allocates a Path.
@ -144,8 +148,8 @@ func New(
sourceCloseTimer: newEmptyTimer(), sourceCloseTimer: newEmptyTimer(),
runOnDemandCloseTimer: newEmptyTimer(), runOnDemandCloseTimer: newEmptyTimer(),
closeTimer: newEmptyTimer(), closeTimer: newEmptyTimer(),
sourceSetReady: make(chan struct{}), extSourceSetReady: make(chan extSourceSetReadyReq),
sourceSetNotReady: make(chan struct{}), extSourceSetNotReady: make(chan struct{}),
clientDescribe: make(chan client.DescribeReq), clientDescribe: make(chan client.DescribeReq),
clientSetupPlay: make(chan client.SetupPlayReq), clientSetupPlay: make(chan client.SetupPlayReq),
clientAnnounce: make(chan client.AnnounceReq), clientAnnounce: make(chan client.AnnounceReq),
@ -213,7 +217,7 @@ outer:
case <-pa.sourceCloseTimer.C: case <-pa.sourceCloseTimer.C:
pa.sourceCloseTimerStarted = false pa.sourceCloseTimerStarted = false
pa.source.(sourceExternal).Close() pa.source.(extSource).Close()
pa.source = nil pa.source = nil
pa.scheduleClose() pa.scheduleClose()
@ -232,10 +236,11 @@ outer:
<-pa.terminate <-pa.terminate
break outer break outer
case <-pa.sourceSetReady: case req := <-pa.extSourceSetReady:
pa.sourceTracks = req.tracks
pa.onSourceSetReady() pa.onSourceSetReady()
case <-pa.sourceSetNotReady: case <-pa.extSourceSetNotReady:
pa.onSourceSetNotReady() pa.onSourceSetNotReady()
case req := <-pa.clientDescribe: case req := <-pa.clientDescribe:
@ -290,7 +295,7 @@ outer:
onInitCmd.Close() onInitCmd.Close()
} }
if source, ok := pa.source.(sourceExternal); ok { if source, ok := pa.source.(extSource); ok {
source.Close() source.Close()
} }
pa.sourceWg.Wait() pa.sourceWg.Wait()
@ -323,8 +328,8 @@ outer:
} }
pa.clientsWg.Wait() pa.clientsWg.Wait()
close(pa.sourceSetReady) close(pa.extSourceSetReady)
close(pa.sourceSetNotReady) close(pa.extSourceSetNotReady)
close(pa.clientDescribe) close(pa.clientDescribe)
close(pa.clientSetupPlay) close(pa.clientSetupPlay)
close(pa.clientAnnounce) close(pa.clientAnnounce)
@ -338,12 +343,12 @@ func (pa *Path) exhaustChannels() {
go func() { go func() {
for { for {
select { select {
case _, ok := <-pa.sourceSetReady: case _, ok := <-pa.extSourceSetReady:
if !ok { if !ok {
return return
} }
case _, ok := <-pa.sourceSetNotReady: case _, ok := <-pa.extSourceSetNotReady:
if !ok { if !ok {
return return
} }
@ -487,13 +492,13 @@ func (pa *Path) removeClient(c client.Client) {
} }
func (pa *Path) onSourceSetReady() { func (pa *Path) onSourceSetReady() {
pa.sourceState = sourceStateReady
if pa.sourceState == sourceStateWaitingDescribe { if pa.sourceState == sourceStateWaitingDescribe {
pa.describeTimer.Stop() pa.describeTimer.Stop()
pa.describeTimer = newEmptyTimer() pa.describeTimer = newEmptyTimer()
} }
pa.sourceState = sourceStateReady
for _, req := range pa.describeRequests { for _, req := range pa.describeRequests {
req.Res <- client.DescribeRes{pa.sourceTracks.Write(), "", nil} //nolint:govet req.Res <- client.DescribeRes{pa.sourceTracks.Write(), "", nil} //nolint:govet
} }
@ -792,15 +797,14 @@ func (pa *Path) Name() string {
return pa.name return pa.name
} }
// OnSourceSetReady is called by a source. // OnExtSourceSetReady is called by a external source.
func (pa *Path) OnSourceSetReady(tracks gortsplib.Tracks) { func (pa *Path) OnExtSourceSetReady(tracks gortsplib.Tracks) {
pa.sourceTracks = tracks pa.extSourceSetReady <- extSourceSetReadyReq{tracks}
pa.sourceSetReady <- struct{}{}
} }
// OnSourceSetNotReady is called by a source. // OnExtSourceSetNotReady is called by a external source.
func (pa *Path) OnSourceSetNotReady() { func (pa *Path) OnExtSourceSetNotReady() {
pa.sourceSetNotReady <- struct{}{} pa.extSourceSetNotReady <- struct{}{}
} }
// OnPathManDescribe is called by pathman.PathMan. // OnPathManDescribe is called by pathman.PathMan.

17
internal/sourcertmp/source.go

@ -26,8 +26,8 @@ const (
// Parent is implemeneted by path.Path. // Parent is implemeneted by path.Path.
type Parent interface { type Parent interface {
Log(logger.Level, string, ...interface{}) Log(logger.Level, string, ...interface{})
OnSourceSetReady(gortsplib.Tracks) OnExtSourceSetReady(gortsplib.Tracks)
OnSourceSetNotReady() OnExtSourceSetNotReady()
OnFrame(int, gortsplib.StreamType, []byte) OnFrame(int, gortsplib.StreamType, []byte)
} }
@ -76,8 +76,8 @@ func (s *Source) Close() {
// IsSource implements path.source. // IsSource implements path.source.
func (s *Source) IsSource() {} func (s *Source) IsSource() {}
// IsSourceExternal implements path.sourceExternal. // IsExtSource implements path.extSource.
func (s *Source) IsSourceExternal() {} func (s *Source) IsExtSource() {}
func (s *Source) log(level logger.Level, format string, args ...interface{}) { func (s *Source) log(level logger.Level, format string, args ...interface{}) {
s.parent.Log(level, "[rtmp source] "+format, args...) s.parent.Log(level, "[rtmp source] "+format, args...)
@ -93,11 +93,8 @@ func (s *Source) run() {
return false return false
} }
t := time.NewTimer(retryPause)
defer t.Stop()
select { select {
case <-t.C: case <-time.After(retryPause):
return true return true
case <-s.terminate: case <-s.terminate:
return false return false
@ -176,8 +173,8 @@ func (s *Source) runInner() bool {
} }
s.log(logger.Info, "ready") s.log(logger.Info, "ready")
s.parent.OnSourceSetReady(tracks) s.parent.OnExtSourceSetReady(tracks)
defer s.parent.OnSourceSetNotReady() defer s.parent.OnExtSourceSetNotReady()
readerDone := make(chan error) readerDone := make(chan error)
go func() { go func() {

19
internal/sourcertsp/source.go

@ -19,8 +19,8 @@ const (
// Parent is implemented by path.Path. // Parent is implemented by path.Path.
type Parent interface { type Parent interface {
Log(logger.Level, string, ...interface{}) Log(logger.Level, string, ...interface{})
OnSourceSetReady(gortsplib.Tracks) OnExtSourceSetReady(gortsplib.Tracks)
OnSourceSetNotReady() OnExtSourceSetNotReady()
OnFrame(int, gortsplib.StreamType, []byte) OnFrame(int, gortsplib.StreamType, []byte)
} }
@ -81,8 +81,8 @@ func (s *Source) Close() {
// IsSource implements path.source. // IsSource implements path.source.
func (s *Source) IsSource() {} func (s *Source) IsSource() {}
// IsSourceExternal implements path.sourceExternal. // IsExtSource implements path.extSource.
func (s *Source) IsSourceExternal() {} func (s *Source) IsExtSource() {}
func (s *Source) log(level logger.Level, format string, args ...interface{}) { func (s *Source) log(level logger.Level, format string, args ...interface{}) {
s.parent.Log(level, "[rtsp source] "+format, args...) s.parent.Log(level, "[rtsp source] "+format, args...)
@ -98,11 +98,8 @@ func (s *Source) run() {
return false return false
} }
t := time.NewTimer(retryPause)
defer t.Stop()
select { select {
case <-t.C: case <-time.After(retryPause):
return true return true
case <-s.terminate: case <-s.terminate:
return false return false
@ -150,11 +147,9 @@ func (s *Source) runInner() bool {
return true return true
} }
tracks := conn.Tracks()
s.log(logger.Info, "ready") s.log(logger.Info, "ready")
s.parent.OnSourceSetReady(tracks) s.parent.OnExtSourceSetReady(conn.Tracks())
defer s.parent.OnSourceSetNotReady() defer s.parent.OnExtSourceSetNotReady()
done := conn.ReadFrames(func(trackID int, streamType gortsplib.StreamType, payload []byte) { done := conn.ReadFrames(func(trackID int, streamType gortsplib.StreamType, payload []byte) {
s.parent.OnFrame(trackID, streamType, payload) s.parent.OnFrame(trackID, streamType, payload)

Loading…
Cancel
Save