Alessandro RosandGitHub
f87d9e659e
sessions are now tracked through cookies or query parameters. This provides the ability to inspect sessions through logs, metrics and API, allows more precise tracking of outbound bytes, decreases load on external HTTP authentication URLs since they are now called once per session and not once per request.
103 lines
2.0 KiB
Go
103 lines
2.0 KiB
Go
package recorder
|
|
|
|
import (
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
|
|
"github.com/bluenviron/mediamtx/internal/conf"
|
|
"github.com/bluenviron/mediamtx/internal/logger"
|
|
"github.com/bluenviron/mediamtx/internal/recordstore"
|
|
"github.com/bluenviron/mediamtx/internal/stream"
|
|
)
|
|
|
|
type recorderInstance struct {
|
|
pathFormat string
|
|
format conf.RecordFormat
|
|
partDuration time.Duration
|
|
maxPartSize conf.StringSize
|
|
segmentDuration time.Duration
|
|
pathName string
|
|
stream *stream.Stream
|
|
onSegmentCreate OnSegmentCreateFunc
|
|
onSegmentComplete OnSegmentCompleteFunc
|
|
parent logger.Writer
|
|
|
|
streamID uuid.UUID
|
|
pathFormat2 string
|
|
format2 format
|
|
skip bool
|
|
reader *stream.Reader
|
|
|
|
terminate chan struct{}
|
|
done chan struct{}
|
|
}
|
|
|
|
// Log implements logger.Writer.
|
|
func (ri *recorderInstance) Log(level logger.Level, format string, args ...any) {
|
|
ri.parent.Log(level, format, args...)
|
|
}
|
|
|
|
func (ri *recorderInstance) initialize() {
|
|
ri.streamID = uuid.New()
|
|
ri.pathFormat2 = ri.pathFormat
|
|
ri.pathFormat2 = recordstore.PathAddExtension(
|
|
strings.ReplaceAll(ri.pathFormat2, "%path", ri.pathName),
|
|
ri.format,
|
|
)
|
|
ri.reader = &stream.Reader{
|
|
SkipOutboundBytes: true,
|
|
Parent: ri,
|
|
}
|
|
|
|
ri.terminate = make(chan struct{})
|
|
ri.done = make(chan struct{})
|
|
|
|
switch ri.format {
|
|
case conf.RecordFormatMPEGTS:
|
|
ri.format2 = &formatMPEGTS{
|
|
ri: ri,
|
|
}
|
|
ok := ri.format2.initialize()
|
|
ri.skip = !ok
|
|
|
|
default:
|
|
ri.format2 = &formatFMP4{
|
|
ri: ri,
|
|
}
|
|
ok := ri.format2.initialize()
|
|
ri.skip = !ok
|
|
}
|
|
|
|
if !ri.skip {
|
|
ri.stream.AddReader(ri.reader)
|
|
}
|
|
|
|
go ri.run()
|
|
}
|
|
|
|
func (ri *recorderInstance) close() {
|
|
close(ri.terminate)
|
|
<-ri.done
|
|
}
|
|
|
|
func (ri *recorderInstance) run() {
|
|
defer close(ri.done)
|
|
|
|
if !ri.skip {
|
|
select {
|
|
case err := <-ri.reader.Error():
|
|
ri.Log(logger.Error, err.Error())
|
|
|
|
case <-ri.terminate:
|
|
}
|
|
|
|
ri.stream.RemoveReader(ri.reader)
|
|
} else {
|
|
<-ri.terminate
|
|
}
|
|
|
|
ri.format2.close()
|
|
}
|