mediamtx/path.go

204 lines
4.8 KiB
Go
Raw Normal View History

package main
import (
2020-07-30 11:31:18 +00:00
"fmt"
"os"
"os/exec"
"time"
"github.com/aler9/sdp/v3"
)
// a publisher is either a client or a source
type publisher interface {
isPublisher()
}
type path struct {
p *program
2020-08-05 09:49:36 +00:00
name string
2020-07-30 11:31:18 +00:00
confp *confPath
permanent bool
publisher publisher
publisherReady bool
publisherSdpText []byte
publisherSdpParsed *sdp.SessionDescription
lastRequested time.Time
2020-07-30 11:31:18 +00:00
lastActivation time.Time
onDemandCmd *exec.Cmd
}
2020-08-05 09:49:36 +00:00
func newPath(p *program, name string, confp *confPath, permanent bool) *path {
2020-07-30 11:31:18 +00:00
pa := &path{
p: p,
2020-08-05 09:49:36 +00:00
name: name,
2020-07-30 11:31:18 +00:00
confp: confp,
permanent: permanent,
}
2020-07-30 11:31:18 +00:00
2020-07-30 15:30:50 +00:00
return pa
}
2020-07-30 11:31:18 +00:00
func (pa *path) check() {
hasClientsWaitingDescribe := func() bool {
for c := range pa.p.clients {
2020-08-05 09:49:36 +00:00
if c.state == clientStateWaitingDescription && c.pathName == pa.name {
return true
}
}
return false
}()
2020-07-30 11:31:18 +00:00
// reply to DESCRIBE requests if they are in timeout
if hasClientsWaitingDescribe &&
time.Since(pa.lastActivation) >= 5*time.Second {
for c := range pa.p.clients {
if c.state == clientStateWaitingDescription &&
2020-08-05 09:49:36 +00:00
c.pathName == pa.name {
c.pathName = ""
2020-07-30 11:31:18 +00:00
c.state = clientStateInitial
2020-08-05 09:49:36 +00:00
c.describeRes <- describeRes{nil, fmt.Errorf("publisher of path '%s' has timed out", pa.name)}
2020-07-30 11:31:18 +00:00
}
}
// perform actions below in next run
return
}
if source, ok := pa.publisher.(*source); ok {
// stop on demand source if needed
if pa.confp.SourceOnDemand &&
source.state == sourceStateRunning &&
time.Since(pa.lastRequested) >= 10*time.Second {
2020-07-30 11:31:18 +00:00
hasClients := func() bool {
for c := range pa.p.clients {
2020-08-05 09:49:36 +00:00
if c.pathName == pa.name {
2020-07-30 11:31:18 +00:00
return true
}
}
return false
}()
if !hasClients {
2020-07-30 11:31:18 +00:00
source.log("stopping since we're not requested anymore")
source.state = sourceStateStopped
source.events <- sourceEventApplyState{source.state}
}
}
} else {
// stop on demand command if needed
if pa.onDemandCmd != nil &&
time.Since(pa.lastRequested) >= 10*time.Second {
2020-07-30 11:31:18 +00:00
hasClientReaders := func() bool {
for c := range pa.p.clients {
2020-08-05 09:49:36 +00:00
if c.pathName == pa.name && c != pa.publisher {
2020-07-30 11:31:18 +00:00
return true
}
}
return false
}()
if !hasClientReaders {
2020-07-30 11:31:18 +00:00
pa.p.log("stopping on demand command since it is not requested anymore")
pa.onDemandCmd.Process.Signal(os.Interrupt)
pa.onDemandCmd.Wait()
pa.onDemandCmd = nil
}
}
}
}
2020-07-30 11:31:18 +00:00
func (pa *path) describe(client *client) {
pa.lastRequested = time.Now()
// publisher not found
if pa.publisher == nil {
2020-07-31 15:49:48 +00:00
// on demand command is available: put the client on hold
2020-07-30 11:31:18 +00:00
if pa.confp.RunOnDemand != "" {
2020-07-31 15:49:48 +00:00
// start on demand command if needed
2020-07-30 11:31:18 +00:00
if pa.onDemandCmd == nil {
pa.p.log("starting on demand command")
pa.lastActivation = time.Now()
pa.onDemandCmd = exec.Command("/bin/sh", "-c", pa.confp.RunOnDemand)
pa.onDemandCmd.Env = append(os.Environ(),
2020-08-05 09:49:36 +00:00
"RTSP_SERVER_PATH="+pa.name,
)
2020-07-30 11:31:18 +00:00
pa.onDemandCmd.Stdout = os.Stdout
pa.onDemandCmd.Stderr = os.Stderr
err := pa.onDemandCmd.Start()
if err != nil {
pa.p.log("ERR: %s", err)
}
}
2020-08-05 09:49:36 +00:00
client.pathName = pa.name
2020-07-30 11:31:18 +00:00
client.state = clientStateWaitingDescription
return
}
// no on-demand: reply with 404
2020-08-05 09:49:36 +00:00
client.describeRes <- describeRes{nil, fmt.Errorf("no one is publishing on path '%s'", pa.name)}
return
2020-07-30 11:31:18 +00:00
}
// publisher was found but is not ready: put the client on hold
if !pa.publisherReady {
// start source if needed
2020-07-30 11:31:18 +00:00
if source, ok := pa.publisher.(*source); ok && source.state == sourceStateStopped {
source.log("starting on demand")
2020-07-30 11:31:18 +00:00
pa.lastActivation = time.Now()
source.state = sourceStateRunning
source.events <- sourceEventApplyState{source.state}
}
2020-08-05 09:49:36 +00:00
client.pathName = pa.name
2020-07-30 11:31:18 +00:00
client.state = clientStateWaitingDescription
return
}
// publisher was found and is ready
2020-07-30 11:31:18 +00:00
client.describeRes <- describeRes{pa.publisherSdpText, nil}
}
func (pa *path) publisherRemove() {
for c := range pa.p.clients {
if c.state == clientStateWaitingDescription &&
2020-08-05 09:49:36 +00:00
c.pathName == pa.name {
c.pathName = ""
2020-07-30 11:31:18 +00:00
c.state = clientStateInitial
2020-08-05 09:49:36 +00:00
c.describeRes <- describeRes{nil, fmt.Errorf("publisher of path '%s' is not available anymore", pa.name)}
2020-07-30 11:31:18 +00:00
}
}
pa.publisher = nil
}
2020-07-30 11:31:18 +00:00
func (pa *path) publisherSetReady() {
pa.publisherReady = true
// reply to all clients that are waiting for a description
2020-07-30 11:31:18 +00:00
for c := range pa.p.clients {
if c.state == clientStateWaitingDescription &&
2020-08-05 09:49:36 +00:00
c.pathName == pa.name {
c.pathName = ""
c.state = clientStateInitial
2020-07-30 11:31:18 +00:00
c.describeRes <- describeRes{pa.publisherSdpText, nil}
}
}
}
2020-07-30 11:31:18 +00:00
func (pa *path) publisherSetNotReady() {
pa.publisherReady = false
// close all clients that are reading
2020-07-30 11:31:18 +00:00
for c := range pa.p.clients {
if c.state != clientStateWaitingDescription &&
2020-07-30 11:31:18 +00:00
c != pa.publisher &&
2020-08-05 09:49:36 +00:00
c.pathName == pa.name {
c.conn.NetConn().Close()
}
}
}