pimhwang
/
mediamtx
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

api: add 'readyTime' to paths (#2049) (#2082)

pull/2083/head
Alessandro Ros 3 years ago committed by GitHub
parent
4064a76ec3
commit
0d18076201
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 87 additions and 69 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 6
      apidocs/openapi.yaml
  2. 4
      internal/core/api_defs.go
  3. 26
      internal/core/api_test.go
  4. 32
      internal/core/hls_manager.go
  5. 38
      internal/core/path.go
  6. 50
      internal/core/path_manager.go

6
apidocs/openapi.yaml
Unescape View File

@ -328,6 +328,12 @@ components:
$ref: '#/components/schemas/PathSourceOrReader' $ref: '#/components/schemas/PathSourceOrReader'
sourceReady: sourceReady:
type: boolean type: boolean
description: deprecated, replaced by 'ready'
ready:
type: boolean
readyTime:
type: string
nullable: true
tracks: tracks:
type: array type: array
items: items:

4
internal/core/api_defs.go
Unescape View File

@ -13,7 +13,9 @@ type apiPath struct {
ConfName string `json:"confName"` ConfName string `json:"confName"`
Conf *conf.PathConf `json:"conf"` Conf *conf.PathConf `json:"conf"`
Source interface{} `json:"source"` Source interface{} `json:"source"`
SourceReady bool `json:"sourceReady"` SourceReady bool `json:"sourceReady"` // Deprecated: renamed to Ready
Ready bool `json:"ready"`
ReadyTime *time.Time `json:"readyTime"`
Tracks []string `json:"tracks"` Tracks []string `json:"tracks"`
BytesReceived uint64 `json:"bytesReceived"` BytesReceived uint64 `json:"bytesReceived"`
Readers []interface{} `json:"readers"` Readers []interface{} `json:"readers"`

26
internal/core/api_test.go
Unescape View File

@ -217,7 +217,7 @@ func TestAPIPathsList(t *testing.T) {
type path struct { type path struct {
Name string `json:"name"` Name string `json:"name"`
Source pathSource `json:"source"` Source pathSource `json:"source"`
SourceReady bool `json:"sourceReady"` Ready bool `json:"ready"`
Tracks []string `json:"tracks"` Tracks []string `json:"tracks"`
BytesReceived uint64 `json:"bytesReceived"` BytesReceived uint64 `json:"bytesReceived"`
} }
@ -280,7 +280,7 @@ func TestAPIPathsList(t *testing.T) {
Source: pathSource{ Source: pathSource{
Type: "rtspSession", Type: "rtspSession",
}, },
SourceReady: true, Ready: true,
Tracks: []string{"H264", "MPEG-4 Audio"}, Tracks: []string{"H264", "MPEG-4 Audio"},
BytesReceived: 16, BytesReceived: 16,
}}, }},
@ -343,8 +343,8 @@ func TestAPIPathsList(t *testing.T) {
Source: pathSource{ Source: pathSource{
Type: "rtspsSession", Type: "rtspsSession",
}, },
SourceReady: true, Ready: true,
Tracks: []string{"H264", "MPEG-4 Audio"}, Tracks: []string{"H264", "MPEG-4 Audio"},
}}, }},
}, out) }, out)
}) })
@ -370,8 +370,8 @@ func TestAPIPathsList(t *testing.T) {
Source: pathSource{ Source: pathSource{
Type: "rtspSource", Type: "rtspSource",
}, },
SourceReady: false, Ready: false,
Tracks: []string{}, Tracks: []string{},
}}, }},
}, out) }, out)
}) })
@ -397,8 +397,8 @@ func TestAPIPathsList(t *testing.T) {
Source: pathSource{ Source: pathSource{
Type: "rtmpSource", Type: "rtmpSource",
}, },
SourceReady: false, Ready: false,
Tracks: []string{}, Tracks: []string{},
}}, }},
}, out) }, out)
}) })
@ -424,8 +424,8 @@ func TestAPIPathsList(t *testing.T) {
Source: pathSource{ Source: pathSource{
Type: "hlsSource", Type: "hlsSource",
}, },
SourceReady: false, Ready: false,
Tracks: []string{}, Tracks: []string{},
}}, }},
}, out) }, out)
}) })
@ -449,7 +449,7 @@ func TestAPIPathsGet(t *testing.T) {
type path struct { type path struct {
Name string `json:"name"` Name string `json:"name"`
Source pathSource `json:"source"` Source pathSource `json:"source"`
SourceReady bool `json:"sourceReady"` Ready bool `json:"Ready"`
Tracks []string `json:"tracks"` Tracks []string `json:"tracks"`
BytesReceived uint64 `json:"bytesReceived"` BytesReceived uint64 `json:"bytesReceived"`
} }
@ -478,8 +478,8 @@ func TestAPIPathsGet(t *testing.T) {
Source: pathSource{ Source: pathSource{
Type: "rtspSession", Type: "rtspSession",
}, },
SourceReady: true, Ready: true,
Tracks: []string{"H264"}, Tracks: []string{"H264"},
}, out) }, out)
} else { } else {
res, err := hc.Get("http://localhost:9997/v2/paths/get/" + pathName) res, err := hc.Get("http://localhost:9997/v2/paths/get/" + pathName)

32
internal/core/hls_manager.go
Unescape View File

@ -54,12 +54,12 @@ type hlsManager struct {
muxers map[string]*hlsMuxer muxers map[string]*hlsMuxer
// in // in
chPathSourceReady chan *path chPathReady chan *path
chPathSourceNotReady chan *path chPathNotReady chan *path
chHandleRequest chan hlsMuxerHandleRequestReq chHandleRequest chan hlsMuxerHandleRequestReq
chMuxerClose chan *hlsMuxer chMuxerClose chan *hlsMuxer
chAPIMuxerList chan hlsManagerAPIMuxersListReq chAPIMuxerList chan hlsManagerAPIMuxersListReq
chAPIMuxerGet chan hlsManagerAPIMuxersGetReq chAPIMuxerGet chan hlsManagerAPIMuxersGetReq
} }
func newHLSManager( func newHLSManager(
@ -101,8 +101,8 @@ func newHLSManager(
ctx: ctx, ctx: ctx,
ctxCancel: ctxCancel, ctxCancel: ctxCancel,
muxers: make(map[string]*hlsMuxer), muxers: make(map[string]*hlsMuxer),
chPathSourceReady: make(chan *path), chPathReady: make(chan *path),
chPathSourceNotReady: make(chan *path), chPathNotReady: make(chan *path),
chHandleRequest: make(chan hlsMuxerHandleRequestReq), chHandleRequest: make(chan hlsMuxerHandleRequestReq),
chMuxerClose: make(chan *hlsMuxer), chMuxerClose: make(chan *hlsMuxer),
chAPIMuxerList: make(chan hlsManagerAPIMuxersListReq), chAPIMuxerList: make(chan hlsManagerAPIMuxersListReq),
@ -157,14 +157,14 @@ func (m *hlsManager) run() {
outer: outer:
for { for {
select { select {
case pa := <-m.chPathSourceReady: case pa := <-m.chPathReady:
if m.alwaysRemux && !pa.conf.SourceOnDemand { if m.alwaysRemux && !pa.conf.SourceOnDemand {
if _, ok := m.muxers[pa.name]; !ok { if _, ok := m.muxers[pa.name]; !ok {
m.createMuxer(pa.name, "") m.createMuxer(pa.name, "")
} }
} }
case pa := <-m.chPathSourceNotReady: case pa := <-m.chPathNotReady:
c, ok := m.muxers[pa.name] c, ok := m.muxers[pa.name]
if ok && c.remoteAddr == "" { // created with "always remux" if ok && c.remoteAddr == "" { // created with "always remux"
c.close() c.close()
@ -258,18 +258,18 @@ func (m *hlsManager) muxerClose(c *hlsMuxer) {
} }
} }
// pathSourceReady is called by pathManager. // pathReady is called by pathManager.
func (m *hlsManager) pathSourceReady(pa *path) { func (m *hlsManager) pathReady(pa *path) {
select { select {
case m.chPathSourceReady <- pa: case m.chPathReady <- pa:
case <-m.ctx.Done(): case <-m.ctx.Done():
} }
} }
// pathSourceNotReady is called by pathManager. // pathNotReady is called by pathManager.
func (m *hlsManager) pathSourceNotReady(pa *path) { func (m *hlsManager) pathNotReady(pa *path) {
select { select {
case m.chPathSourceNotReady <- pa: case m.chPathNotReady <- pa:
case <-m.ctx.Done(): case <-m.ctx.Done():
} }
} }

38
internal/core/path.go
Unescape View File

@ -44,8 +44,8 @@ func (e pathErrNoOnePublishing) Error() string {
type pathParent interface { type pathParent interface {
logger.Writer logger.Writer
pathSourceReady(*path) pathReady(*path)
pathSourceNotReady(*path) pathNotReady(*path)
onPathClose(*path) onPathClose(*path)
} }
@ -196,8 +196,9 @@ type path struct {
ctxCancel func() ctxCancel func()
confMutex sync.RWMutex confMutex sync.RWMutex
source source source source
bytesReceived *uint64
stream *stream stream *stream
readyTime time.Time
bytesReceived *uint64
readers map[reader]struct{} readers map[reader]struct{}
describeRequestsOnHold []pathDescribeReq describeRequestsOnHold []pathDescribeReq
readerAddRequestsOnHold []pathReaderAddReq readerAddRequestsOnHold []pathReaderAddReq
@ -359,7 +360,7 @@ func (pa *path) run() {
} }
case <-pa.onDemandStaticSourceCloseTimer.C: case <-pa.onDemandStaticSourceCloseTimer.C:
pa.sourceSetNotReady() pa.setNotReady()
pa.onDemandStaticSourceStop() pa.onDemandStaticSourceStop()
if pa.shouldClose() { if pa.shouldClose() {
@ -400,7 +401,7 @@ func (pa *path) run() {
pa.confMutex.Unlock() pa.confMutex.Unlock()
case req := <-pa.chSourceStaticSetReady: case req := <-pa.chSourceStaticSetReady:
err := pa.sourceSetReady(req.medias, req.generateRTPPackets) err := pa.setReady(req.medias, req.generateRTPPackets)
if err != nil { if err != nil {
req.res <- pathSourceStaticSetReadyRes{err: err} req.res <- pathSourceStaticSetReadyRes{err: err}
} else { } else {
@ -427,7 +428,7 @@ func (pa *path) run() {
} }
case req := <-pa.chSourceStaticSetNotReady: case req := <-pa.chSourceStaticSetNotReady:
pa.sourceSetNotReady() pa.setNotReady()
// send response before calling onDemandStaticSourceStop() // send response before calling onDemandStaticSourceStop()
// in order to avoid a deadlock due to sourceStatic.stop() // in order to avoid a deadlock due to sourceStatic.stop()
@ -511,7 +512,7 @@ func (pa *path) run() {
} }
if pa.stream != nil { if pa.stream != nil {
pa.sourceSetNotReady() pa.setNotReady()
} }
if pa.source != nil { if pa.source != nil {
@ -626,7 +627,7 @@ func (pa *path) onDemandPublisherStop() {
} }
} }
func (pa *path) sourceSetReady(medias media.Medias, allocateEncoder bool) error { func (pa *path) setReady(medias media.Medias, allocateEncoder bool) error {
stream, err := newStream( stream, err := newStream(
pa.udpMaxPayloadSize, pa.udpMaxPayloadSize,
medias, medias,
@ -639,6 +640,7 @@ func (pa *path) sourceSetReady(medias media.Medias, allocateEncoder bool) error
} }
pa.stream = stream pa.stream = stream
pa.readyTime = time.Now()
if pa.conf.RunOnReady != "" { if pa.conf.RunOnReady != "" {
pa.Log(logger.Info, "runOnReady command started") pa.Log(logger.Info, "runOnReady command started")
@ -652,13 +654,13 @@ func (pa *path) sourceSetReady(medias media.Medias, allocateEncoder bool) error
}) })
} }
pa.parent.pathSourceReady(pa) pa.parent.pathReady(pa)
return nil return nil
} }
func (pa *path) sourceSetNotReady() { func (pa *path) setNotReady() {
pa.parent.pathSourceNotReady(pa) pa.parent.pathNotReady(pa)
for r := range pa.readers { for r := range pa.readers {
pa.doReaderRemove(r) pa.doReaderRemove(r)
@ -683,7 +685,7 @@ func (pa *path) doReaderRemove(r reader) {
func (pa *path) doPublisherRemove() { func (pa *path) doPublisherRemove() {
if pa.stream != nil { if pa.stream != nil {
pa.sourceSetNotReady() pa.setNotReady()
} }
pa.source = nil pa.source = nil
@ -777,7 +779,7 @@ func (pa *path) handlePublisherStart(req pathPublisherStartReq) {
return return
} }
err := pa.sourceSetReady(req.medias, req.generateRTPPackets) err := pa.setReady(req.medias, req.generateRTPPackets)
if err != nil { if err != nil {
req.res <- pathPublisherRecordRes{err: err} req.res <- pathPublisherRecordRes{err: err}
return return
@ -807,7 +809,7 @@ func (pa *path) handlePublisherStart(req pathPublisherStartReq) {
func (pa *path) handlePublisherStop(req pathPublisherStopReq) { func (pa *path) handlePublisherStop(req pathPublisherStopReq) {
if req.author == pa.source && pa.stream != nil { if req.author == pa.source && pa.stream != nil {
pa.sourceSetNotReady() pa.setNotReady()
} }
close(req.res) close(req.res)
} }
@ -892,6 +894,14 @@ func (pa *path) handleAPIPathsGet(req pathAPIPathsGetReq) {
return pa.source.apiSourceDescribe() return pa.source.apiSourceDescribe()
}(), }(),
SourceReady: pa.stream != nil, SourceReady: pa.stream != nil,
Ready: pa.stream != nil,
ReadyTime: func() *time.Time {
if pa.stream == nil {
return nil
}
v := pa.readyTime
return &v
}(),
Tracks: func() []string { Tracks: func() []string {
if pa.stream == nil { if pa.stream == nil {
return []string{} return []string{}

50
internal/core/path_manager.go
Unescape View File

@ -55,8 +55,8 @@ func getConfForPath(pathConfs map[string]*conf.PathConf, name string) (string, *
} }
type pathManagerHLSManager interface { type pathManagerHLSManager interface {
pathSourceReady(*path) pathReady(*path)
pathSourceNotReady(*path) pathNotReady(*path)
} }
type pathManagerParent interface { type pathManagerParent interface {
@ -84,17 +84,17 @@ type pathManager struct {
pathsByConf map[string]map[*path]struct{} pathsByConf map[string]map[*path]struct{}
// in // in
chConfReload chan map[string]*conf.PathConf chConfReload chan map[string]*conf.PathConf
chPathClose chan *path chPathClose chan *path
chPathSourceReady chan *path chPathReady chan *path
chPathSourceNotReady chan *path chPathNotReady chan *path
chGetConfForPath chan pathGetConfForPathReq chGetConfForPath chan pathGetConfForPathReq
chDescribe chan pathDescribeReq chDescribe chan pathDescribeReq
chReaderAdd chan pathReaderAddReq chReaderAdd chan pathReaderAddReq
chPublisherAdd chan pathPublisherAddReq chPublisherAdd chan pathPublisherAddReq
chHLSManagerSet chan pathManagerHLSManager chHLSManagerSet chan pathManagerHLSManager
chAPIPathsList chan pathAPIPathsListReq chAPIPathsList chan pathAPIPathsListReq
chAPIPathsGet chan pathAPIPathsGetReq chAPIPathsGet chan pathAPIPathsGetReq
} }
func newPathManager( func newPathManager(
@ -130,8 +130,8 @@ func newPathManager(
pathsByConf: make(map[string]map[*path]struct{}), pathsByConf: make(map[string]map[*path]struct{}),
chConfReload: make(chan map[string]*conf.PathConf), chConfReload: make(chan map[string]*conf.PathConf),
chPathClose: make(chan *path), chPathClose: make(chan *path),
chPathSourceReady: make(chan *path), chPathReady: make(chan *path),
chPathSourceNotReady: make(chan *path), chPathNotReady: make(chan *path),
chGetConfForPath: make(chan pathGetConfForPathReq), chGetConfForPath: make(chan pathGetConfForPathReq),
chDescribe: make(chan pathDescribeReq), chDescribe: make(chan pathDescribeReq),
chReaderAdd: make(chan pathReaderAddReq), chReaderAdd: make(chan pathReaderAddReq),
@ -218,14 +218,14 @@ outer:
} }
pm.removePath(pa) pm.removePath(pa)
case pa := <-pm.chPathSourceReady: case pa := <-pm.chPathReady:
if pm.hlsManager != nil { if pm.hlsManager != nil {
pm.hlsManager.pathSourceReady(pa) pm.hlsManager.pathReady(pa)
} }
case pa := <-pm.chPathSourceNotReady: case pa := <-pm.chPathNotReady:
if pm.hlsManager != nil { if pm.hlsManager != nil {
pm.hlsManager.pathSourceNotReady(pa) pm.hlsManager.pathNotReady(pa)
} }
case req := <-pm.chGetConfForPath: case req := <-pm.chGetConfForPath:
@ -386,19 +386,19 @@ func (pm *pathManager) confReload(pathConfs map[string]*conf.PathConf) {
} }
} }
// pathSourceReady is called by path. // pathReady is called by path.
func (pm *pathManager) pathSourceReady(pa *path) { func (pm *pathManager) pathReady(pa *path) {
select { select {
case pm.chPathSourceReady <- pa: case pm.chPathReady <- pa:
case <-pm.ctx.Done(): case <-pm.ctx.Done():
case <-pa.ctx.Done(): // in case pathManager is blocked by path.wait() case <-pa.ctx.Done(): // in case pathManager is blocked by path.wait()
} }
} }
// pathSourceNotReady is called by path. // pathNotReady is called by path.
func (pm *pathManager) pathSourceNotReady(pa *path) { func (pm *pathManager) pathNotReady(pa *path) {
select { select {
case pm.chPathSourceNotReady <- pa: case pm.chPathNotReady <- pa:
case <-pm.ctx.Done(): case <-pm.ctx.Done():
case <-pa.ctx.Done(): // in case pathManager is blocked by path.wait() case <-pa.ctx.Done(): // in case pathManager is blocked by path.wait()
} }

Write Preview
Loading…
Cancel
Save