apps/core
1
0
Fork 0
mirror of https://github.com/datarhei/core.git synced 2025-10-03 23:26:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

Allow probe with individual timeout

Browse Source
This commit is contained in:
Ingo Oppermann
2023-01-23 17:09:55 +01:00
parent 505fbff03f
commit b2cd8f7133
2 changed files with 43 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
restream/restream.go
View File

@@ -30,30 +30,31 @@ import (
// The Restreamer interface // The Restreamer interface
type Restreamer interface { type Restreamer interface {
ID() string // ID of this instance ID() string // ID of this instance
Name() string // Arbitrary name of this instance Name() string // Arbitrary name of this instance
CreatedAt() time.Time // Time of when this instance has been created CreatedAt() time.Time // Time of when this instance has been created
Start() // Start all processes that have a "start" order Start() // Start all processes that have a "start" order
Stop() // Stop all running process but keep their "start" order Stop() // Stop all running process but keep their "start" order
AddProcess(config *app.Config) error // Add a new process AddProcess(config *app.Config) error // Add a new process
GetProcessIDs(idpattern, refpattern string) []string // Get a list of process IDs based on patterns for ID and reference GetProcessIDs(idpattern, refpattern string) []string // Get a list of process IDs based on patterns for ID and reference
DeleteProcess(id string) error // Delete a process DeleteProcess(id string) error // Delete a process
UpdateProcess(id string, config *app.Config) error // Update a process UpdateProcess(id string, config *app.Config) error // Update a process
StartProcess(id string) error // Start a process StartProcess(id string) error // Start a process
StopProcess(id string) error // Stop a process StopProcess(id string) error // Stop a process
RestartProcess(id string) error // Restart a process RestartProcess(id string) error // Restart a process
ReloadProcess(id string) error // Reload a process ReloadProcess(id string) error // Reload a process
GetProcess(id string) (*app.Process, error) // Get a process GetProcess(id string) (*app.Process, error) // Get a process
GetProcessState(id string) (*app.State, error) // Get the state of a process GetProcessState(id string) (*app.State, error) // Get the state of a process
GetProcessLog(id string) (*app.Log, error) // Get the logs of a process GetProcessLog(id string) (*app.Log, error) // Get the logs of a process
GetPlayout(id, inputid string) (string, error) // Get the URL of the playout API for a process GetPlayout(id, inputid string) (string, error) // Get the URL of the playout API for a process
Probe(id string) app.Probe // Probe a process Probe(id string) app.Probe // Probe a process
Skills() skills.Skills // Get the ffmpeg skills ProbeWithTimeout(id string, timeout time.Duration) app.Probe // Probe a process with specific timeout
ReloadSkills() error // Reload the ffmpeg skills Skills() skills.Skills // Get the ffmpeg skills
SetProcessMetadata(id, key string, data interface{}) error // Set metatdata to a process ReloadSkills() error // Reload the ffmpeg skills
GetProcessMetadata(id, key string) (interface{}, error) // Get previously set metadata from a process SetProcessMetadata(id, key string, data interface{}) error // Set metatdata to a process
SetMetadata(key string, data interface{}) error // Set general metadata GetProcessMetadata(id, key string) (interface{}, error) // Get previously set metadata from a process
GetMetadata(key string) (interface{}, error) // Get previously set general metadata SetMetadata(key string, data interface{}) error // Set general metadata
GetMetadata(key string) (interface{}, error) // Get previously set general metadata
} }
// Config is the required configuration for a new restreamer instance. // Config is the required configuration for a new restreamer instance.
@@ -1251,6 +1252,10 @@ func (r *restream) GetProcessLog(id string) (*app.Log, error) {
} }
func (r *restream) Probe(id string) app.Probe { func (r *restream) Probe(id string) app.Probe {
return r.ProbeWithTimeout(id, 20*time.Second)
}
func (r *restream) ProbeWithTimeout(id string, timeout time.Duration) app.Probe {
r.lock.RLock() r.lock.RLock()
appprobe := app.Probe{} appprobe := app.Probe{}
@@ -1288,7 +1293,7 @@ func (r *restream) Probe(id string) app.Probe {
ffmpeg, err := r.ffmpeg.New(ffmpeg.ProcessConfig{ ffmpeg, err := r.ffmpeg.New(ffmpeg.ProcessConfig{
Reconnect: false, Reconnect: false,
ReconnectDelay: 0, ReconnectDelay: 0,
StaleTimeout: 20 * time.Second, StaleTimeout: timeout,
Command: command, Command: command,
Parser: prober, Parser: prober,
Logger: task.logger, Logger: task.logger,

13
restream/restream_test.go
View File

@@ -394,6 +394,19 @@ func TestReloadProcess(t *testing.T) {
rs.StopProcess(process.ID) rs.StopProcess(process.ID)
} }
func TestProbeProcess(t *testing.T) {
rs, err := getDummyRestreamer(nil, nil, nil, nil)
require.NoError(t, err)
process := getDummyProcess()
rs.AddProcess(process)
probe := rs.ProbeWithTimeout(process.ID, 5*time.Second)
require.Equal(t, 3, len(probe.Streams))
}
func TestProcessMetadata(t *testing.T) { func TestProcessMetadata(t *testing.T) {
rs, err := getDummyRestreamer(nil, nil, nil, nil) rs, err := getDummyRestreamer(nil, nil, nil, nil)
require.NoError(t, err) require.NoError(t, err)