-
Notifications
You must be signed in to change notification settings - Fork 1.4k
/
agent_instance.go
85 lines (64 loc) · 1.5 KB
/
agent_instance.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
package record
import (
"strings"
"time"
"github.com/bluenviron/mediacommon/pkg/formats/fmp4"
"github.com/bluenviron/mediamtx/internal/asyncwriter"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/logger"
)
// OnSegmentFunc is the prototype of the function passed as runOnSegmentStart / runOnSegmentComplete
type OnSegmentFunc = func(string)
type sample struct {
*fmp4.PartSample
dts time.Duration
ntp time.Time
}
type agentInstance struct {
agent *Agent
pathFormat string
writer *asyncwriter.Writer
format format
terminate chan struct{}
done chan struct{}
}
func (a *agentInstance) initialize() {
a.pathFormat = a.agent.PathFormat
a.pathFormat = PathAddExtension(
strings.ReplaceAll(a.pathFormat, "%path", a.agent.PathName),
a.agent.Format,
)
a.terminate = make(chan struct{})
a.done = make(chan struct{})
a.writer = asyncwriter.New(a.agent.WriteQueueSize, a.agent)
switch a.agent.Format {
case conf.RecordFormatMPEGTS:
a.format = &formatMPEGTS{
a: a,
}
a.format.initialize()
default:
a.format = &formatFMP4{
a: a,
}
a.format.initialize()
}
go a.run()
}
func (a *agentInstance) close() {
close(a.terminate)
<-a.done
}
func (a *agentInstance) run() {
defer close(a.done)
a.writer.Start()
select {
case err := <-a.writer.Error():
a.agent.Log(logger.Error, err.Error())
a.agent.Stream.RemoveReader(a.writer)
case <-a.terminate:
a.agent.Stream.RemoveReader(a.writer)
a.writer.Stop()
}
a.format.close()
}