OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

fix cleaning of recordings in case of multiple recordDeleteAfter values (#3557) (#3741)

Browse Source
This commit is contained in:
Alessandro Ros
2024-09-08 20:33:18 +02:00
committed by GitHub
parent 1f478329eb
commit 73a300afd0
34 changed files with 738 additions and 588 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/api.go
+30 -9
View File
@@ -22,7 +22,7 @@ import (
"github.com/bluenviron/mediamtx/internal/defs"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/protocols/httpp"
"github.com/bluenviron/mediamtx/internal/record"
"github.com/bluenviron/mediamtx/internal/recordstore"
"github.com/bluenviron/mediamtx/internal/restrictnetwork"
"github.com/bluenviron/mediamtx/internal/servers/hls"
"github.com/bluenviron/mediamtx/internal/servers/rtmp"
@@ -56,6 +56,27 @@ func paramName(ctx *gin.Context) (string, bool) {
return name[1:], true
}
func recordingsOfPath(
pathConf *conf.Path,
pathName string,
) *defs.APIRecording {
ret := &defs.APIRecording{
Name: pathName,
}
segments, _ := recordstore.FindSegments(pathConf, pathName)
ret.Segments = make([]*defs.APIRecordingSegment, len(segments))
for i, seg := range segments {
ret.Segments[i] = &defs.APIRecordingSegment{
Start: seg.Start,
}
}
return ret
}
// PathManager contains methods used by the API and Metrics server.
type PathManager interface {
APIPathsList() (*defs.APIPathList, error)
@@ -1062,7 +1083,7 @@ func (a *API) onRecordingsList(ctx *gin.Context) {
c := a.Conf
a.mutex.RUnlock()
pathNames := getAllPathsWithRecordings(c.Paths)
pathNames := recordstore.FindAllPathsWithSegments(c.Paths)
data := defs.APIRecordingList{}
@@ -1077,8 +1098,8 @@ func (a *API) onRecordingsList(ctx *gin.Context) {
data.Items = make([]*defs.APIRecording, len(pathNames))
for i, pathName := range pathNames {
_, pathConf, _, _ := conf.FindPathConf(c.Paths, pathName)
data.Items[i] = recordingEntry(pathConf, pathName)
pathConf, _, _ := conf.FindPathConf(c.Paths, pathName)
data.Items[i] = recordingsOfPath(pathConf, pathName)
}
ctx.JSON(http.StatusOK, data)
@@ -1095,13 +1116,13 @@ func (a *API) onRecordingsGet(ctx *gin.Context) {
c := a.Conf
a.mutex.RUnlock()
_, pathConf, _, err := conf.FindPathConf(c.Paths, pathName)
pathConf, _, err := conf.FindPathConf(c.Paths, pathName)
if err != nil {
a.writeError(ctx, http.StatusBadRequest, err)
return
}
ctx.JSON(http.StatusOK, recordingEntry(pathConf, pathName))
ctx.JSON(http.StatusOK, recordingsOfPath(pathConf, pathName))
}
func (a *API) onRecordingDeleteSegment(ctx *gin.Context) {
@@ -1117,18 +1138,18 @@ func (a *API) onRecordingDeleteSegment(ctx *gin.Context) {
c := a.Conf
a.mutex.RUnlock()
_, pathConf, _, err := conf.FindPathConf(c.Paths, pathName)
pathConf, _, err := conf.FindPathConf(c.Paths, pathName)
if err != nil {
a.writeError(ctx, http.StatusBadRequest, err)
return
}
pathFormat := record.PathAddExtension(
pathFormat := recordstore.PathAddExtension(
strings.ReplaceAll(pathConf.RecordPath, "%path", pathName),
pathConf.RecordFormat,
)
segmentPath := record.Path{
segmentPath := recordstore.Path{
Start: start,
}.Encode(pathFormat)
internal/api/api_test.go
+1
View File
@@ -595,6 +595,7 @@ func TestRecordingsList(t *testing.T) {
cnf := tempConf(t, "pathDefaults:\n"+
" recordPath: "+filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f")+"\n"+
"paths:\n"+
" mypath1:\n"+
" all_others:\n")
api := API{
internal/api/record.go
-137
View File
@@ -1,137 +0,0 @@
package api
import (
"errors"
"io/fs"
"path/filepath"
"sort"
"strings"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/defs"
"github.com/bluenviron/mediamtx/internal/playback"
"github.com/bluenviron/mediamtx/internal/record"
)
var errFound = errors.New("found")
func fixedPathHasRecordings(pathConf *conf.Path) bool {
recordPath := record.PathAddExtension(
strings.ReplaceAll(pathConf.RecordPath, "%path", pathConf.Name),
pathConf.RecordFormat,
)
// we have to convert to absolute paths
// otherwise, recordPath and fpath inside Walk() won't have common elements
recordPath, _ = filepath.Abs(recordPath)
commonPath := record.CommonPath(recordPath)
err := filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error {
if err != nil {
return err
}
if !info.IsDir() {
var pa record.Path
ok := pa.Decode(recordPath, fpath)
if ok {
return errFound
}
}
return nil
})
if err != nil && !errors.Is(err, errFound) {
return false
}
return errors.Is(err, errFound)
}
func regexpPathGetRecordings(pathConf *conf.Path) []string {
recordPath := record.PathAddExtension(
pathConf.RecordPath,
pathConf.RecordFormat,
)
// we have to convert to absolute paths
// otherwise, recordPath and fpath inside Walk() won't have common elements
recordPath, _ = filepath.Abs(recordPath)
commonPath := record.CommonPath(recordPath)
var ret []string
filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error { //nolint:errcheck
if err != nil {
return err
}
if !info.IsDir() {
var pa record.Path
ok := pa.Decode(recordPath, fpath)
if ok && pathConf.Regexp.FindStringSubmatch(pa.Path) != nil {
ret = append(ret, pa.Path)
}
}
return nil
})
return ret
}
func removeDuplicatesAndSort(in []string) []string {
ma := make(map[string]struct{}, len(in))
for _, i := range in {
ma[i] = struct{}{}
}
out := []string{}
for k := range ma {
out = append(out, k)
}
sort.Strings(out)
return out
}
func getAllPathsWithRecordings(paths map[string]*conf.Path) []string {
pathNames := []string{}
for _, pathConf := range paths {
if pathConf.Regexp == nil {
if fixedPathHasRecordings(pathConf) {
pathNames = append(pathNames, pathConf.Name)
}
} else {
pathNames = append(pathNames, regexpPathGetRecordings(pathConf)...)
}
}
return removeDuplicatesAndSort(pathNames)
}
func recordingEntry(
pathConf *conf.Path,
pathName string,
) *defs.APIRecording {
ret := &defs.APIRecording{
Name: pathName,
}
segments, _ := playback.FindSegments(pathConf, pathName)
ret.Segments = make([]*defs.APIRecordingSegment, len(segments))
for i, seg := range segments {
ret.Segments[i] = &defs.APIRecordingSegment{
Start: seg.Start,
}
}
return ret
}
internal/conf/path.go
+7 -7
View File
@@ -46,15 +46,15 @@ func srtCheckPassphrase(passphrase string) error {
}
// FindPathConf returns the configuration corresponding to the given path name.
func FindPathConf(pathConfs map[string]*Path, name string) (string, *Path, []string, error) {
func FindPathConf(pathConfs map[string]*Path, name string) (*Path, []string, error) {
err := isValidPathName(name)
if err != nil {
return "", nil, nil, fmt.Errorf("invalid path name: %w (%s)", err, name)
return nil, nil, fmt.Errorf("invalid path name: %w (%s)", err, name)
}
// normal path
if pathConf, ok := pathConfs[name]; ok {
return name, pathConf, nil, nil
return pathConf, nil, nil
}
// regular expression-based path
@@ -62,22 +62,22 @@ func FindPathConf(pathConfs map[string]*Path, name string) (string, *Path, []str
if pathConf.Regexp != nil && pathConfName != "all" && pathConfName != "all_others" {
m := pathConf.Regexp.FindStringSubmatch(name)
if m != nil {
return pathConfName, pathConf, m, nil
return pathConf, m, nil
}
}
}
// all_others
// process all_others after every other entry
for pathConfName, pathConf := range pathConfs {
if pathConfName == "all" || pathConfName == "all_others" {
m := pathConf.Regexp.FindStringSubmatch(name)
if m != nil {
return pathConfName, pathConf, m, nil
return pathConf, m, nil
}
}
}
return "", nil, nil, fmt.Errorf("path '%s' is not configured", name)
return nil, nil, fmt.Errorf("path '%s' is not configured", name)
}
// Path is a path configuration.
internal/core/core.go
+9 -42
View File
@@ -8,7 +8,6 @@ import (
"os/signal"
"path/filepath"
"reflect"
"sort"
"strings"
"time"
@@ -25,7 +24,7 @@ import (
"github.com/bluenviron/mediamtx/internal/metrics"
"github.com/bluenviron/mediamtx/internal/playback"
"github.com/bluenviron/mediamtx/internal/pprof"
"github.com/bluenviron/mediamtx/internal/record"
"github.com/bluenviron/mediamtx/internal/recordcleaner"
"github.com/bluenviron/mediamtx/internal/rlimit"
"github.com/bluenviron/mediamtx/internal/servers/hls"
"github.com/bluenviron/mediamtx/internal/servers/rtmp"
@@ -44,38 +43,6 @@ var defaultConfPaths = []string{
"/etc/mediamtx/mediamtx.yml",
}
func gatherCleanerEntries(paths map[string]*conf.Path) []record.CleanerEntry {
out := make(map[record.CleanerEntry]struct{})
for _, pa := range paths {
if pa.Record && pa.RecordDeleteAfter != 0 {
entry := record.CleanerEntry{
Path: pa.RecordPath,
Format: pa.RecordFormat,
DeleteAfter: time.Duration(pa.RecordDeleteAfter),
}
out[entry] = struct{}{}
}
}
out2 := make([]record.CleanerEntry, len(out))
i := 0
for v := range out {
out2[i] = v
i++
}
sort.Slice(out2, func(i, j int) bool {
if out2[i].Path != out2[j].Path {
return out2[i].Path < out2[j].Path
}
return out2[i].DeleteAfter < out2[j].DeleteAfter
})
return out2
}
var cli struct {
Version bool `help:"print version"`
Confpath string `arg:"" default:""`
@@ -92,7 +59,7 @@ type Core struct {
authManager *auth.Manager
metrics *metrics.Metrics
pprof *pprof.PPROF
recordCleaner *record.Cleaner
recordCleaner *recordcleaner.Cleaner
playbackServer *playback.Server
pathManager *pathManager
rtspServer *rtsp.Server
@@ -333,12 +300,10 @@ func (p *Core) createResources(initial bool) error {
p.pprof = i
}
cleanerEntries := gatherCleanerEntries(p.conf.Paths)
if len(cleanerEntries) != 0 &&
p.recordCleaner == nil {
p.recordCleaner = &record.Cleaner{
Entries: cleanerEntries,
Parent: p,
if p.recordCleaner == nil {
p.recordCleaner = &recordcleaner.Cleaner{
PathConfs: p.conf.Paths,
Parent: p,
}
p.recordCleaner.Initialize()
}
@@ -707,8 +672,10 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
closeLogger
closeRecorderCleaner := newConf == nil ||
!reflect.DeepEqual(gatherCleanerEntries(newConf.Paths), gatherCleanerEntries(p.conf.Paths)) ||
closeLogger
if !closeRecorderCleaner && !reflect.DeepEqual(newConf.Paths, p.conf.Paths) {
p.recordCleaner.ReloadPathConfs(newConf.Paths)
}
closePlaybackServer := newConf == nil ||
newConf.Playback != p.conf.Playback ||
internal/core/path.go
+12 -13
View File
@@ -17,7 +17,7 @@ import (
"github.com/bluenviron/mediamtx/internal/externalcmd"
"github.com/bluenviron/mediamtx/internal/hooks"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/record"
"github.com/bluenviron/mediamtx/internal/recorder"
"github.com/bluenviron/mediamtx/internal/stream"
)
@@ -71,7 +71,6 @@ type path struct {
writeTimeout conf.StringDuration
writeQueueSize int
udpMaxPayloadSize int
confName string
conf *conf.Path
name string
matches []string
@@ -85,7 +84,7 @@ type path struct {
source defs.Source
publisherQuery string
stream *stream.Stream
recordAgent *record.Agent
recorder *recorder.Recorder
readyTime time.Time
onUnDemandHook func(string)
onNotReadyHook func()
@@ -368,12 +367,12 @@ func (pa *path) doReloadConf(newConf *conf.Path) {
}
if pa.conf.Record {
if pa.stream != nil && pa.recordAgent == nil {
if pa.stream != nil && pa.recorder == nil {
pa.startRecording()
}
} else if pa.recordAgent != nil {
pa.recordAgent.Close()
pa.recordAgent = nil
} else if pa.recorder != nil {
pa.recorder.Close()
pa.recorder = nil
}
}
@@ -572,7 +571,7 @@ func (pa *path) doAPIPathsGet(req pathAPIPathsGetReq) {
req.res <- pathAPIPathsGetRes{
data: &defs.APIPath{
Name: pa.name,
ConfName: pa.confName,
ConfName: pa.conf.Name,
Source: func() *defs.APIPathSourceOrReader {
if pa.source == nil {
return nil
@@ -765,9 +764,9 @@ func (pa *path) setNotReady() {
pa.onNotReadyHook()
if pa.recordAgent != nil {
pa.recordAgent.Close()
pa.recordAgent = nil
if pa.recorder != nil {
pa.recorder.Close()
pa.recorder = nil
}
if pa.stream != nil {
@@ -777,7 +776,7 @@ func (pa *path) setNotReady() {
}
func (pa *path) startRecording() {
pa.recordAgent = &record.Agent{
pa.recorder = &recorder.Recorder{
WriteQueueSize: pa.writeQueueSize,
PathFormat: pa.conf.RecordPath,
Format: pa.conf.RecordFormat,
@@ -816,7 +815,7 @@ func (pa *path) startRecording() {
},
Parent: pa,
}
pa.recordAgent.Initialize()
pa.recorder.Initialize()
}
func (pa *path) executeRemoveReader(r defs.Reader) {
internal/core/path_manager.go
+16 -18
View File
@@ -100,9 +100,9 @@ func (pm *pathManager) initialize() {
pm.chAPIPathsList = make(chan pathAPIPathsListReq)
pm.chAPIPathsGet = make(chan pathAPIPathsGetReq)
for pathConfName, pathConf := range pm.pathConfs {
for _, pathConf := range pm.pathConfs {
if pathConf.Regexp == nil {
pm.createPath(pathConfName, pathConf, pathConfName, nil)
pm.createPath(pathConf, pathConf.Name, nil)
}
}
@@ -202,7 +202,7 @@ func (pm *pathManager) doReloadConf(newPaths map[string]*conf.Path) {
// add new paths
for pathConfName, pathConf := range pm.pathConfs {
if _, ok := pm.paths[pathConfName]; !ok && pathConf.Regexp == nil {
pm.createPath(pathConfName, pathConf, pathConfName, nil)
pm.createPath(pathConf, pathConfName, nil)
}
}
}
@@ -231,7 +231,7 @@ func (pm *pathManager) doPathNotReady(pa *path) {
}
func (pm *pathManager) doFindPathConf(req defs.PathFindPathConfReq) {
_, pathConf, _, err := conf.FindPathConf(pm.pathConfs, req.AccessRequest.Name)
pathConf, _, err := conf.FindPathConf(pm.pathConfs, req.AccessRequest.Name)
if err != nil {
req.Res <- defs.PathFindPathConfRes{Err: err}
return
@@ -247,7 +247,7 @@ func (pm *pathManager) doFindPathConf(req defs.PathFindPathConfReq) {
}
func (pm *pathManager) doDescribe(req defs.PathDescribeReq) {
pathConfName, pathConf, pathMatches, err := conf.FindPathConf(pm.pathConfs, req.AccessRequest.Name)
pathConf, pathMatches, err := conf.FindPathConf(pm.pathConfs, req.AccessRequest.Name)
if err != nil {
req.Res <- defs.PathDescribeRes{Err: err}
return
@@ -261,14 +261,14 @@ func (pm *pathManager) doDescribe(req defs.PathDescribeReq) {
// create path if it doesn't exist
if _, ok := pm.paths[req.AccessRequest.Name]; !ok {
pm.createPath(pathConfName, pathConf, req.AccessRequest.Name, pathMatches)
pm.createPath(pathConf, req.AccessRequest.Name, pathMatches)
}
req.Res <- defs.PathDescribeRes{Path: pm.paths[req.AccessRequest.Name]}
}
func (pm *pathManager) doAddReader(req defs.PathAddReaderReq) {
pathConfName, pathConf, pathMatches, err := conf.FindPathConf(pm.pathConfs, req.AccessRequest.Name)
pathConf, pathMatches, err := conf.FindPathConf(pm.pathConfs, req.AccessRequest.Name)
if err != nil {
req.Res <- defs.PathAddReaderRes{Err: err}
return
@@ -284,14 +284,14 @@ func (pm *pathManager) doAddReader(req defs.PathAddReaderReq) {
// create path if it doesn't exist
if _, ok := pm.paths[req.AccessRequest.Name]; !ok {
pm.createPath(pathConfName, pathConf, req.AccessRequest.Name, pathMatches)
pm.createPath(pathConf, req.AccessRequest.Name, pathMatches)
}
req.Res <- defs.PathAddReaderRes{Path: pm.paths[req.AccessRequest.Name]}
}
func (pm *pathManager) doAddPublisher(req defs.PathAddPublisherReq) {
pathConfName, pathConf, pathMatches, err := conf.FindPathConf(pm.pathConfs, req.AccessRequest.Name)
pathConf, pathMatches, err := conf.FindPathConf(pm.pathConfs, req.AccessRequest.Name)
if err != nil {
req.Res <- defs.PathAddPublisherRes{Err: err}
return
@@ -307,7 +307,7 @@ func (pm *pathManager) doAddPublisher(req defs.PathAddPublisherReq) {
// create path if it doesn't exist
if _, ok := pm.paths[req.AccessRequest.Name]; !ok {
pm.createPath(pathConfName, pathConf, req.AccessRequest.Name, pathMatches)
pm.createPath(pathConf, req.AccessRequest.Name, pathMatches)
}
req.Res <- defs.PathAddPublisherRes{Path: pm.paths[req.AccessRequest.Name]}
@@ -334,7 +334,6 @@ func (pm *pathManager) doAPIPathsGet(req pathAPIPathsGetReq) {
}
func (pm *pathManager) createPath(
pathConfName string,
pathConf *conf.Path,
name string,
matches []string,
@@ -347,7 +346,6 @@ func (pm *pathManager) createPath(
writeTimeout: pm.writeTimeout,
writeQueueSize: pm.writeQueueSize,
udpMaxPayloadSize: pm.udpMaxPayloadSize,
confName: pathConfName,
conf: pathConf,
name: name,
matches: matches,
@@ -359,16 +357,16 @@ func (pm *pathManager) createPath(
pm.paths[name] = pa
if _, ok := pm.pathsByConf[pathConfName]; !ok {
pm.pathsByConf[pathConfName] = make(map[*path]struct{})
if _, ok := pm.pathsByConf[pathConf.Name]; !ok {
pm.pathsByConf[pathConf.Name] = make(map[*path]struct{})
}
pm.pathsByConf[pathConfName][pa] = struct{}{}
pm.pathsByConf[pathConf.Name][pa] = struct{}{}
}
func (pm *pathManager) removePath(pa *path) {
delete(pm.pathsByConf[pa.confName], pa)
if len(pm.pathsByConf[pa.confName]) == 0 {
delete(pm.pathsByConf, pa.confName)
delete(pm.pathsByConf[pa.conf.Name], pa)
if len(pm.pathsByConf[pa.conf.Name]) == 0 {
delete(pm.pathsByConf, pa.conf.Name)
}
delete(pm.paths, pa.name)
}
internal/playback/on_get.go
+5 -4
View File
@@ -12,6 +12,7 @@ import (
"github.com/bluenviron/mediacommon/pkg/formats/fmp4"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/recordstore"
"github.com/gin-gonic/gin"
)
@@ -41,7 +42,7 @@ func parseDuration(raw string) (time.Duration, error) {
func seekAndMux(
recordFormat conf.RecordFormat,
segments []*Segment,
segments []*recordstore.Segment,
start time.Time,
duration time.Duration,
m muxer,
@@ -152,9 +153,9 @@ func (s *Server) onGet(ctx *gin.Context) {
return
}
segments, err := findSegmentsInTimespan(pathConf, pathName, start, duration)
segments, err := recordstore.FindSegmentsInTimespan(pathConf, pathName, start, duration)
if err != nil {
if errors.Is(err, errNoSegmentsFound) {
if errors.Is(err, recordstore.ErrNoSegmentsFound) {
s.writeError(ctx, http.StatusNotFound, err)
} else {
s.writeError(ctx, http.StatusBadRequest, err)
@@ -172,7 +173,7 @@ func (s *Server) onGet(ctx *gin.Context) {
// nothing has been written yet; send back JSON
if !ww.written {
if errors.Is(err, errNoSegmentsFound) {
if errors.Is(err, recordstore.ErrNoSegmentsFound) {
s.writeError(ctx, http.StatusNotFound, err)
} else {
s.writeError(ctx, http.StatusBadRequest, err)
internal/playback/on_get_test.go
+3
View File
@@ -235,6 +235,7 @@ func TestOnGet(t *testing.T) {
ReadTimeout: conf.StringDuration(10 * time.Second),
PathConfs: map[string]*conf.Path{
"mypath": {
Name: "mypath",
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
},
},
@@ -520,6 +521,7 @@ func TestOnGetDifferentInit(t *testing.T) {
ReadTimeout: conf.StringDuration(10 * time.Second),
PathConfs: map[string]*conf.Path{
"mypath": {
Name: "mypath",
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
},
},
@@ -595,6 +597,7 @@ func TestOnGetNTPCompensation(t *testing.T) {
ReadTimeout: conf.StringDuration(10 * time.Second),
PathConfs: map[string]*conf.Path{
"mypath": {
Name: "mypath",
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
},
},
internal/playback/on_list.go
+7 -3
View File
@@ -13,6 +13,7 @@ import (
"github.com/bluenviron/mediacommon/pkg/formats/fmp4"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/recordstore"
"github.com/gin-gonic/gin"
)
@@ -28,7 +29,10 @@ type listEntry struct {
URL string `json:"url"`
}
func computeDurationAndConcatenate(recordFormat conf.RecordFormat, segments []*Segment) ([]listEntry, error) {
func computeDurationAndConcatenate(
recordFormat conf.RecordFormat,
segments []*recordstore.Segment,
) ([]listEntry, error) {
if recordFormat == conf.RecordFormatFMP4 {
out := []listEntry{}
var prevInit *fmp4.Init
@@ -99,9 +103,9 @@ func (s *Server) onList(ctx *gin.Context) {
return
}
segments, err := FindSegments(pathConf, pathName)
segments, err := recordstore.FindSegments(pathConf, pathName)
if err != nil {
if errors.Is(err, errNoSegmentsFound) {
if errors.Is(err, recordstore.ErrNoSegmentsFound) {
s.writeError(ctx, http.StatusNotFound, err)
} else {
s.writeError(ctx, http.StatusBadRequest, err)
internal/playback/on_list_test.go
+2
View File
@@ -32,6 +32,7 @@ func TestOnList(t *testing.T) {
ReadTimeout: conf.StringDuration(10 * time.Second),
PathConfs: map[string]*conf.Path{
"mypath": {
Name: "mypath",
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
},
},
@@ -106,6 +107,7 @@ func TestOnListDifferentInit(t *testing.T) {
ReadTimeout: conf.StringDuration(10 * time.Second),
PathConfs: map[string]*conf.Path{
"mypath": {
Name: "mypath",
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
},
},
internal/playback/segment.go
-140
View File
@@ -1,140 +0,0 @@
package playback
import (
"io/fs"
"path/filepath"
"sort"
"strings"
"time"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/record"
)
// Segment is a recording segment.
type Segment struct {
Fpath string
Start time.Time
}
func findSegmentsInTimespan(
pathConf *conf.Path,
pathName string,
start time.Time,
duration time.Duration,
) ([]*Segment, error) {
recordPath := record.PathAddExtension(
strings.ReplaceAll(pathConf.RecordPath, "%path", pathName),
pathConf.RecordFormat,
)
// we have to convert to absolute paths
// otherwise, recordPath and fpath inside Walk() won't have common elements
recordPath, _ = filepath.Abs(recordPath)
commonPath := record.CommonPath(recordPath)
end := start.Add(duration)
var segments []*Segment
err := filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error {
if err != nil {
return err
}
if !info.IsDir() {
var pa record.Path
ok := pa.Decode(recordPath, fpath)
// gather all segments that starts before the end of the playback
if ok && !end.Before(pa.Start) {
segments = append(segments, &Segment{
Fpath: fpath,
Start: pa.Start,
})
}
}
return nil
})
if err != nil {
return nil, err
}
if segments == nil {
return nil, errNoSegmentsFound
}
sort.Slice(segments, func(i, j int) bool {
return segments[i].Start.Before(segments[j].Start)
})
// find the segment that may contain the start of the playback and remove all previous ones
found := false
for i := 0; i < len(segments)-1; i++ {
if !start.Before(segments[i].Start) && start.Before(segments[i+1].Start) {
segments = segments[i:]
found = true
break
}
}
// otherwise, keep the last segment only and check if it may contain the start of the playback
if !found {
segments = segments[len(segments)-1:]
if segments[len(segments)-1].Start.After(start) {
return nil, errNoSegmentsFound
}
}
return segments, nil
}
// FindSegments returns all segments of a path.
func FindSegments(
pathConf *conf.Path,
pathName string,
) ([]*Segment, error) {
recordPath := record.PathAddExtension(
strings.ReplaceAll(pathConf.RecordPath, "%path", pathName),
pathConf.RecordFormat,
)
// we have to convert to absolute paths
// otherwise, recordPath and fpath inside Walk() won't have common elements
recordPath, _ = filepath.Abs(recordPath)
commonPath := record.CommonPath(recordPath)
var segments []*Segment
err := filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error {
if err != nil {
return err
}
if !info.IsDir() {
var pa record.Path
ok := pa.Decode(recordPath, fpath)
if ok {
segments = append(segments, &Segment{
Fpath: fpath,
Start: pa.Start,
})
}
}
return nil
})
if err != nil {
return nil, err
}
if segments == nil {
return nil, errNoSegmentsFound
}
sort.Slice(segments, func(i, j int) bool {
return segments[i].Start.Before(segments[j].Start)
})
return segments, nil
}
internal/playback/segment_fmp4.go
+2 -1
View File
@@ -10,6 +10,7 @@ import (
"github.com/abema/go-mp4"
"github.com/bluenviron/mediacommon/pkg/formats/fmp4"
"github.com/bluenviron/mediamtx/internal/recordstore"
)
const (
@@ -466,7 +467,7 @@ func segmentFMP4SeekAndMuxParts(
}
if !atLeastOnePartWritten {
return 0, errNoSegmentsFound
return 0, recordstore.ErrNoSegmentsFound
}
return maxMuxerDTS, nil
internal/playback/server.go
+1 -3
View File
@@ -16,8 +16,6 @@ import (
"github.com/gin-gonic/gin"
)
var errNoSegmentsFound = errors.New("no recording segments found")
type serverAuthManager interface {
Authenticate(req *auth.Request) error
}
@@ -102,7 +100,7 @@ func (s *Server) safeFindPathConf(name string) (*conf.Path, error) {
s.mutex.RLock()
defer s.mutex.RUnlock()
_, pathConf, _, err := conf.FindPathConf(s.PathConfs, name)
pathConf, _, err := conf.FindPathConf(s.PathConfs, name)
return pathConf, err
}
internal/record/cleaner.go
-124
View File
@@ -1,124 +0,0 @@
package record
import (
"context"
"io/fs"
"os"
"path/filepath"
"time"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/logger"
)
var timeNow = time.Now
// CleanerEntry is a cleaner entry.
type CleanerEntry struct {
Path string
Format conf.RecordFormat
DeleteAfter time.Duration
}
// Cleaner removes expired recording segments from disk.
type Cleaner struct {
Entries []CleanerEntry
Parent logger.Writer
ctx context.Context
ctxCancel func()
done chan struct{}
}
// Initialize initializes a Cleaner.
func (c *Cleaner) Initialize() {
c.ctx, c.ctxCancel = context.WithCancel(context.Background())
c.done = make(chan struct{})
go c.run()
}
// Close closes the Cleaner.
func (c *Cleaner) Close() {
c.ctxCancel()
<-c.done
}
// Log implements logger.Writer.
func (c *Cleaner) Log(level logger.Level, format string, args ...interface{}) {
c.Parent.Log(level, "[record cleaner]"+format, args...)
}
func (c *Cleaner) run() {
defer close(c.done)
interval := 30 * 60 * time.Second
for _, e := range c.Entries {
if interval > (e.DeleteAfter / 2) {
interval = e.DeleteAfter / 2
}
}
c.doRun() //nolint:errcheck
for {
select {
case <-time.After(interval):
c.doRun()
case <-c.ctx.Done():
return
}
}
}
func (c *Cleaner) doRun() {
for _, e := range c.Entries {
c.doRunEntry(&e) //nolint:errcheck
}
}
func (c *Cleaner) doRunEntry(e *CleanerEntry) error {
entryPath := PathAddExtension(e.Path, e.Format)
// we have to convert to absolute paths
// otherwise, entryPath and fpath inside Walk() won't have common elements
entryPath, _ = filepath.Abs(entryPath)
commonPath := CommonPath(entryPath)
now := timeNow()
filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error { //nolint:errcheck
if err != nil {
return err
}
if !info.IsDir() {
var pa Path
ok := pa.Decode(entryPath, fpath)
if ok {
if now.Sub(pa.Start) > e.DeleteAfter {
c.Log(logger.Debug, "removing %s", fpath)
os.Remove(fpath)
}
}
}
return nil
})
filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error { //nolint:errcheck
if err != nil {
return err
}
if info.IsDir() {
os.Remove(fpath)
}
return nil
})
return nil
}
internal/record/cleaner_test.go
-52
View File
@@ -1,52 +0,0 @@
package record
import (
"os"
"path/filepath"
"testing"
"time"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/test"
"github.com/stretchr/testify/require"
)
func TestCleaner(t *testing.T) {
timeNow = func() time.Time {
return time.Date(2009, 0o5, 20, 22, 15, 25, 427000, time.Local)
}
dir, err := os.MkdirTemp("", "mediamtx-cleaner")
require.NoError(t, err)
defer os.RemoveAll(dir)
const specialChars = "_-+*?^$()[]{}|"
err = os.Mkdir(filepath.Join(dir, specialChars+"_mypath"), 0o755)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, specialChars+"_mypath", "2008-05-20_22-15-25-000125.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, specialChars+"_mypath", "2009-05-20_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
c := &Cleaner{
Entries: []CleanerEntry{{
Path: filepath.Join(dir, specialChars+"_%path/%Y-%m-%d_%H-%M-%S-%f"),
Format: conf.RecordFormatFMP4,
DeleteAfter: 10 * time.Second,
}},
Parent: test.NilLogger,
}
c.Initialize()
defer c.Close()
time.Sleep(500 * time.Millisecond)
_, err = os.Stat(filepath.Join(dir, specialChars+"_mypath", "2008-05-20_22-15-25-000125.mp4"))
require.Error(t, err)
_, err = os.Stat(filepath.Join(dir, specialChars+"_mypath", "2009-05-20_22-15-25-000427.mp4"))
require.NoError(t, err)
}
internal/record/record.go
-2
View File
@@ -1,2 +0,0 @@
// Package record contains the recording system.
package record
internal/recordcleaner/cleaner.go
+134
View File
@@ -0,0 +1,134 @@
// Package recordcleaner contains the recording cleaner.
package recordcleaner
import (
"context"
"os"
"time"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/recordstore"
)
var timeNow = time.Now
// Cleaner removes expired recording segments from disk.
type Cleaner struct {
PathConfs map[string]*conf.Path
Parent logger.Writer
ctx context.Context
ctxCancel func()
chReloadConf chan map[string]*conf.Path
done chan struct{}
}
// Initialize initializes a Cleaner.
func (c *Cleaner) Initialize() {
c.ctx, c.ctxCancel = context.WithCancel(context.Background())
c.chReloadConf = make(chan map[string]*conf.Path)
c.done = make(chan struct{})
go c.run()
}
// Close closes the Cleaner.
func (c *Cleaner) Close() {
c.ctxCancel()
<-c.done
}
// Log implements logger.Writer.
func (c *Cleaner) Log(level logger.Level, format string, args ...interface{}) {
c.Parent.Log(level, "[record cleaner]"+format, args...)
}
// ReloadPathConfs is called by core.Core.
func (c *Cleaner) ReloadPathConfs(pathConfs map[string]*conf.Path) {
select {
case c.chReloadConf <- pathConfs:
case <-c.ctx.Done():
}
}
func (c *Cleaner) run() {
defer close(c.done)
c.doRun() //nolint:errcheck
for {
select {
case <-time.After(c.cleanInterval()):
c.doRun()
case cnf := <-c.chReloadConf:
c.PathConfs = cnf
case <-c.ctx.Done():
return
}
}
}
func (c *Cleaner) atLeastOneRecordDeleteAfter() bool {
for _, e := range c.PathConfs {
if e.RecordDeleteAfter != 0 {
return true
}
}
return false
}
func (c *Cleaner) cleanInterval() time.Duration {
if !c.atLeastOneRecordDeleteAfter() {
return 365 * 24 * time.Hour
}
interval := 30 * 60 * time.Second
for _, e := range c.PathConfs {
if e.RecordDeleteAfter != 0 &&
interval > (time.Duration(e.RecordDeleteAfter)/2) {
interval = time.Duration(e.RecordDeleteAfter) / 2
}
}
return interval
}
func (c *Cleaner) doRun() {
now := timeNow()
pathNames := recordstore.FindAllPathsWithSegments(c.PathConfs)
for _, pathName := range pathNames {
c.processPath(now, pathName) //nolint:errcheck
}
}
func (c *Cleaner) processPath(now time.Time, pathName string) error {
pathConf, _, err := conf.FindPathConf(c.PathConfs, pathName)
if err != nil {
return err
}
if pathConf.RecordDeleteAfter == 0 {
return nil
}
segments, err := recordstore.FindSegments(pathConf, pathName)
if err != nil {
return err
}
for _, seg := range segments {
if now.Sub(seg.Start) > time.Duration(pathConf.RecordDeleteAfter) {
c.Log(logger.Debug, "removing %s", seg.Fpath)
os.Remove(seg.Fpath)
}
}
return nil
}
internal/recordcleaner/cleaner_test.go
+107
View File
@@ -0,0 +1,107 @@
package recordcleaner
import (
"os"
"path/filepath"
"regexp"
"testing"
"time"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/test"
"github.com/stretchr/testify/require"
)
func TestCleaner(t *testing.T) {
timeNow = func() time.Time {
return time.Date(2009, 5, 20, 22, 15, 25, 427000, time.Local)
}
dir, err := os.MkdirTemp("", "mediamtx-cleaner")
require.NoError(t, err)
defer os.RemoveAll(dir)
const specialChars = "_-+*?^$()[]{}|"
err = os.Mkdir(filepath.Join(dir, specialChars+"_mypath"), 0o755)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, specialChars+"_mypath", "2008-05-20_22-15-25-000125.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, specialChars+"_mypath", "2009-05-20_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
c := &Cleaner{
PathConfs: map[string]*conf.Path{
"~^.*$": {
Name: "~^.*$",
Regexp: regexp.MustCompile("^.*$"),
RecordPath: filepath.Join(dir, specialChars+"_%path/%Y-%m-%d_%H-%M-%S-%f"),
RecordFormat: conf.RecordFormatFMP4,
RecordDeleteAfter: conf.StringDuration(10 * time.Second),
},
},
Parent: test.NilLogger,
}
c.Initialize()
defer c.Close()
time.Sleep(500 * time.Millisecond)
_, err = os.Stat(filepath.Join(dir, specialChars+"_mypath", "2008-05-20_22-15-25-000125.mp4"))
require.Error(t, err)
_, err = os.Stat(filepath.Join(dir, specialChars+"_mypath", "2009-05-20_22-15-25-000427.mp4"))
require.NoError(t, err)
}
func TestCleanerMultipleEntriesSamePath(t *testing.T) {
timeNow = func() time.Time {
return time.Date(2009, 5, 20, 22, 15, 25, 427000, time.Local)
}
dir, err := os.MkdirTemp("", "mediamtx-cleaner")
require.NoError(t, err)
defer os.RemoveAll(dir)
err = os.Mkdir(filepath.Join(dir, "path1"), 0o755)
require.NoError(t, err)
err = os.Mkdir(filepath.Join(dir, "path2"), 0o755)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, "path1", "2009-05-19_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, "path2", "2009-05-19_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
c := &Cleaner{
PathConfs: map[string]*conf.Path{
"path1": {
Name: "path1",
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
RecordFormat: conf.RecordFormatFMP4,
RecordDeleteAfter: conf.StringDuration(10 * time.Second),
},
"path2": {
Name: "path2",
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
RecordFormat: conf.RecordFormatFMP4,
RecordDeleteAfter: conf.StringDuration(10 * 24 * time.Hour),
},
},
Parent: test.NilLogger,
}
c.Initialize()
defer c.Close()
time.Sleep(500 * time.Millisecond)
_, err = os.Stat(filepath.Join(dir, "path1", "2009-05-19_22-15-25-000427.mp4"))
require.Error(t, err)
_, err = os.Stat(filepath.Join(dir, "path2", "2009-05-19_22-15-25-000427.mp4"))
require.NoError(t, err)
}
internal/record/format.go → internal/recorder/format.go
+1 -1
View File
@@ -1,4 +1,4 @@
package record
package recorder
type format interface {
initialize()
internal/record/format_fmp4.go → internal/recorder/format_fmp4.go
+1 -1
View File
@@ -1,4 +1,4 @@
package record
package recorder
import (
"bytes"
internal/record/format_fmp4_part.go → internal/recorder/format_fmp4_part.go
+3 -2
View File
@@ -1,4 +1,4 @@
package record
package recorder
import (
"io"
@@ -10,6 +10,7 @@ import (
"github.com/bluenviron/mediacommon/pkg/formats/fmp4/seekablebuffer"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/recordstore"
)
func writePart(
@@ -54,7 +55,7 @@ func (p *formatFMP4Part) initialize() {
func (p *formatFMP4Part) close() error {
if p.s.fi == nil {
p.s.path = Path{Start: p.s.startNTP}.Encode(p.s.f.ai.pathFormat)
p.s.path = recordstore.Path{Start: p.s.startNTP}.Encode(p.s.f.ai.pathFormat)
p.s.f.ai.Log(logger.Debug, "creating segment %s", p.s.path)
err := os.MkdirAll(filepath.Dir(p.s.path), 0o755)
internal/record/format_fmp4_segment.go → internal/recorder/format_fmp4_segment.go
+1 -1
View File
@@ -1,4 +1,4 @@
package record
package recorder
import (
"io"
internal/record/format_fmp4_track.go → internal/recorder/format_fmp4_track.go
+1 -1
View File
@@ -1,4 +1,4 @@
package record
package recorder
import (
"github.com/bluenviron/mediacommon/pkg/formats/fmp4"
internal/record/format_mpegts.go → internal/recorder/format_mpegts.go
+1 -1
View File
@@ -1,4 +1,4 @@
package record
package recorder
import (
"bufio"
internal/record/format_mpegts_segment.go → internal/recorder/format_mpegts_segment.go
+3 -2
View File
@@ -1,4 +1,4 @@
package record
package recorder
import (
"os"
@@ -6,6 +6,7 @@ import (
"time"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/recordstore"
)
type formatMPEGTSSegment struct {
@@ -46,7 +47,7 @@ func (s *formatMPEGTSSegment) close() error {
func (s *formatMPEGTSSegment) Write(p []byte) (int, error) {
if s.fi == nil {
s.path = Path{Start: s.startNTP}.Encode(s.f.ai.pathFormat)
s.path = recordstore.Path{Start: s.startNTP}.Encode(s.f.ai.pathFormat)
s.f.ai.Log(logger.Debug, "creating segment %s", s.path)
err := os.MkdirAll(filepath.Dir(s.path), 0o755)
internal/record/agent_instance.go → internal/recorder/recoder_instance.go
+4 -3
View File
@@ -1,4 +1,4 @@
package record
package recorder
import (
"strings"
@@ -9,6 +9,7 @@ import (
"github.com/bluenviron/mediamtx/internal/asyncwriter"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/recordstore"
)
type sample struct {
@@ -18,7 +19,7 @@ type sample struct {
}
type agentInstance struct {
agent *Agent
agent *Recorder
pathFormat string
writer *asyncwriter.Writer
@@ -36,7 +37,7 @@ func (ai *agentInstance) Log(level logger.Level, format string, args ...interfac
func (ai *agentInstance) initialize() {
ai.pathFormat = ai.agent.PathFormat
ai.pathFormat = PathAddExtension(
ai.pathFormat = recordstore.PathAddExtension(
strings.ReplaceAll(ai.pathFormat, "%path", ai.agent.PathName),
ai.agent.Format,
)
internal/record/agent.go → internal/recorder/recorder.go
+10 -9
View File
@@ -1,4 +1,5 @@
package record
// Package recorder contains the recorder.
package recorder
import (
"time"
@@ -14,8 +15,8 @@ type OnSegmentCreateFunc = func(path string)
// OnSegmentCompleteFunc is the prototype of the function passed as OnSegmentComplete
type OnSegmentCompleteFunc = func(path string, duration time.Duration)
// Agent writes recordings to disk.
type Agent struct {
// Recorder writes recordings to disk.
type Recorder struct {
WriteQueueSize int
PathFormat string
Format conf.RecordFormat
@@ -35,8 +36,8 @@ type Agent struct {
done chan struct{}
}
// Initialize initializes Agent.
func (w *Agent) Initialize() {
// Initialize initializes Recorder.
func (w *Recorder) Initialize() {
if w.OnSegmentCreate == nil {
w.OnSegmentCreate = func(string) {
}
@@ -61,18 +62,18 @@ func (w *Agent) Initialize() {
}
// Log implements logger.Writer.
func (w *Agent) Log(level logger.Level, format string, args ...interface{}) {
w.Parent.Log(level, "[record] "+format, args...)
func (w *Recorder) Log(level logger.Level, format string, args ...interface{}) {
w.Parent.Log(level, "[recorder] "+format, args...)
}
// Close closes the agent.
func (w *Agent) Close() {
func (w *Recorder) Close() {
w.Log(logger.Info, "recording stopped")
close(w.terminate)
<-w.done
}
func (w *Agent) run() {
func (w *Recorder) run() {
defer close(w.done)
for {
internal/record/agent_test.go → internal/recorder/recorder_test.go
+9 -9
View File
@@ -1,4 +1,4 @@
package record
package recorder
import (
"os"
@@ -19,7 +19,7 @@ import (
"github.com/bluenviron/mediamtx/internal/unit"
)
func TestAgent(t *testing.T) {
func TestRecorder(t *testing.T) {
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
@@ -156,7 +156,7 @@ func TestAgent(t *testing.T) {
n := 0
w := &Agent{
w := &Recorder{
WriteQueueSize: 1024,
PathFormat: recordPath,
Format: f,
@@ -197,11 +197,11 @@ func TestAgent(t *testing.T) {
writeToStream(stream,
50*time.Second,
time.Date(2008, 0o5, 20, 22, 15, 25, 0, time.UTC))
time.Date(2008, 5, 20, 22, 15, 25, 0, time.UTC))
writeToStream(stream,
52*time.Second,
time.Date(2008, 0o5, 20, 22, 16, 25, 0, time.UTC))
time.Date(2008, 5, 20, 22, 16, 25, 0, time.UTC))
// simulate a write error
stream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
@@ -295,7 +295,7 @@ func TestAgent(t *testing.T) {
writeToStream(stream,
300*time.Second,
time.Date(2010, 0o5, 20, 22, 15, 25, 0, time.UTC))
time.Date(2010, 5, 20, 22, 15, 25, 0, time.UTC))
time.Sleep(50 * time.Millisecond)
@@ -310,7 +310,7 @@ func TestAgent(t *testing.T) {
}
}
func TestAgentFMP4NegativeDTS(t *testing.T) {
func TestRecorderFMP4NegativeDTS(t *testing.T) {
desc := &description.Session{Medias: []*description.Media{
{
Type: description.MediaTypeVideo,
@@ -350,7 +350,7 @@ func TestAgentFMP4NegativeDTS(t *testing.T) {
recordPath := filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f")
w := &Agent{
w := &Recorder{
WriteQueueSize: 1024,
PathFormat: recordPath,
Format: conf.RecordFormatFMP4,
@@ -366,7 +366,7 @@ func TestAgentFMP4NegativeDTS(t *testing.T) {
stream.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{
Base: unit.Base{
PTS: -50*time.Millisecond + (time.Duration(i) * 200 * time.Millisecond),
NTP: time.Date(2008, 0o5, 20, 22, 15, 25, 0, time.UTC),
NTP: time.Date(2008, 5, 20, 22, 15, 25, 0, time.UTC),
},
AU: [][]byte{
test.FormatH264.SPS,
internal/record/path.go → internal/recordstore/path.go
+2 -2
View File
@@ -1,4 +1,4 @@
package record
package recordstore
import (
"regexp"
@@ -23,7 +23,7 @@ func leadingZeros(v int, size int) string {
return out2 + out
}
// PathAddExtension adds the file extension to path.
// PathAddExtension adds the file extension to the path.
func PathAddExtension(path string, format conf.RecordFormat) string {
switch format {
case conf.RecordFormatMPEGTS:
internal/record/path_test.go → internal/recordstore/path_test.go
+1 -1
View File
@@ -1,4 +1,4 @@
package record
package recordstore
import (
"testing"
internal/recordstore/recordstore.go
+2
View File
@@ -0,0 +1,2 @@
// Package recordstore contains utilities to store/retrieve recordings to/from disk.
package recordstore
internal/recordstore/segment.go
+240
View File
@@ -0,0 +1,240 @@
package recordstore
import (
"errors"
"io/fs"
"path/filepath"
"sort"
"strings"
"time"
"github.com/bluenviron/mediamtx/internal/conf"
)
// ErrNoSegmentsFound is returned when no recording segments have been found.
var ErrNoSegmentsFound = errors.New("no recording segments found")
var errFound = errors.New("found")
// Segment is a recording segment.
type Segment struct {
Fpath string
Start time.Time
}
func fixedPathHasSegments(pathConf *conf.Path) bool {
recordPath := PathAddExtension(
strings.ReplaceAll(pathConf.RecordPath, "%path", pathConf.Name),
pathConf.RecordFormat,
)
// we have to convert to absolute paths
// otherwise, recordPath and fpath inside Walk() won't have common elements
recordPath, _ = filepath.Abs(recordPath)
commonPath := CommonPath(recordPath)
err := filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error {
if err != nil {
return err
}
if !info.IsDir() {
var pa Path
ok := pa.Decode(recordPath, fpath)
if ok {
return errFound
}
}
return nil
})
if err != nil && !errors.Is(err, errFound) {
return false
}
return errors.Is(err, errFound)
}
func regexpPathFindPathsWithSegments(pathConf *conf.Path) map[string]struct{} {
recordPath := PathAddExtension(
pathConf.RecordPath,
pathConf.RecordFormat,
)
// we have to convert to absolute paths
// otherwise, recordPath and fpath inside Walk() won't have common elements
recordPath, _ = filepath.Abs(recordPath)
commonPath := CommonPath(recordPath)
ret := make(map[string]struct{})
filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error { //nolint:errcheck
if err != nil {
return err
}
if !info.IsDir() {
var pa Path
ok := pa.Decode(recordPath, fpath)
if ok && pathConf.Regexp.FindStringSubmatch(pa.Path) != nil {
ret[pa.Path] = struct{}{}
}
}
return nil
})
return ret
}
// FindAllPathsWithSegments returns all paths that do have segments.
func FindAllPathsWithSegments(pathConfs map[string]*conf.Path) []string {
pathNames := make(map[string]struct{})
for _, pathConf := range pathConfs {
if pathConf.Regexp == nil {
if fixedPathHasSegments(pathConf) {
pathNames[pathConf.Name] = struct{}{}
}
} else {
for name := range regexpPathFindPathsWithSegments(pathConf) {
pathNames[name] = struct{}{}
}
}
}
out := make([]string, len(pathNames))
n := 0
for k := range pathNames {
out[n] = k
n++
}
sort.Strings(out)
return out
}
// FindSegments returns all segments of a path.
func FindSegments(
pathConf *conf.Path,
pathName string,
) ([]*Segment, error) {
recordPath := PathAddExtension(
strings.ReplaceAll(pathConf.RecordPath, "%path", pathName),
pathConf.RecordFormat,
)
// we have to convert to absolute paths
// otherwise, recordPath and fpath inside Walk() won't have common elements
recordPath, _ = filepath.Abs(recordPath)
commonPath := CommonPath(recordPath)
var segments []*Segment
err := filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error {
if err != nil {
return err
}
if !info.IsDir() {
var pa Path
ok := pa.Decode(recordPath, fpath)
if ok {
segments = append(segments, &Segment{
Fpath: fpath,
Start: pa.Start,
})
}
}
return nil
})
if err != nil {
return nil, err
}
if segments == nil {
return nil, ErrNoSegmentsFound
}
sort.Slice(segments, func(i, j int) bool {
return segments[i].Start.Before(segments[j].Start)
})
return segments, nil
}
// FindSegmentsInTimespan returns all segments in a certain timestamp.
func FindSegmentsInTimespan(
pathConf *conf.Path,
pathName string,
start time.Time,
duration time.Duration,
) ([]*Segment, error) {
recordPath := PathAddExtension(
strings.ReplaceAll(pathConf.RecordPath, "%path", pathName),
pathConf.RecordFormat,
)
// we have to convert to absolute paths
// otherwise, recordPath and fpath inside Walk() won't have common elements
recordPath, _ = filepath.Abs(recordPath)
commonPath := CommonPath(recordPath)
end := start.Add(duration)
var segments []*Segment
err := filepath.Walk(commonPath, func(fpath string, info fs.FileInfo, err error) error {
if err != nil {
return err
}
if !info.IsDir() {
var pa Path
ok := pa.Decode(recordPath, fpath)
// gather all segments that starts before the end of the playback
if ok && !end.Before(pa.Start) {
segments = append(segments, &Segment{
Fpath: fpath,
Start: pa.Start,
})
}
}
return nil
})
if err != nil {
return nil, err
}
if segments == nil {
return nil, ErrNoSegmentsFound
}
sort.Slice(segments, func(i, j int) bool {
return segments[i].Start.Before(segments[j].Start)
})
// find the segment that may contain the start of the playback and remove all previous ones
found := false
for i := 0; i < len(segments)-1; i++ {
if !start.Before(segments[i].Start) && start.Before(segments[i+1].Start) {
segments = segments[i:]
found = true
break
}
}
// otherwise, keep the last segment only and check if it may contain the start of the playback
if !found {
segments = segments[len(segments)-1:]
if segments[len(segments)-1].Start.After(start) {
return nil, ErrNoSegmentsFound
}
}
return segments, nil
}
internal/recordstore/segment_test.go
+123
View File
@@ -0,0 +1,123 @@
package recordstore
import (
"os"
"path/filepath"
"regexp"
"testing"
"time"
"github.com/bluenviron/mediamtx/internal/conf"
"github.com/stretchr/testify/require"
)
func TestFindAllPathsWithSegments(t *testing.T) {
dir, err := os.MkdirTemp("", "mediamtx-recordstore")
require.NoError(t, err)
defer os.RemoveAll(dir)
err = os.Mkdir(filepath.Join(dir, "path1"), 0o755)
require.NoError(t, err)
err = os.Mkdir(filepath.Join(dir, "path2"), 0o755)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, "path1", "2015-05-19_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, "path2", "2015-07-19_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
paths := FindAllPathsWithSegments(map[string]*conf.Path{
"~^.*$": {
Name: "~^.*$",
Regexp: regexp.MustCompile("^.*$"),
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
RecordFormat: conf.RecordFormatFMP4,
},
"path2": {
Name: "path2",
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
RecordFormat: conf.RecordFormatFMP4,
},
})
require.Equal(t, []string{"path1", "path2"}, paths)
}
func TestFindSegments(t *testing.T) {
dir, err := os.MkdirTemp("", "mediamtx-recordstore")
require.NoError(t, err)
defer os.RemoveAll(dir)
err = os.Mkdir(filepath.Join(dir, "path1"), 0o755)
require.NoError(t, err)
err = os.Mkdir(filepath.Join(dir, "path2"), 0o755)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, "path1", "2015-05-19_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, "path1", "2016-05-19_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
segments, err := FindSegments(
&conf.Path{
Name: "~^.*$",
Regexp: regexp.MustCompile("^.*$"),
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
RecordFormat: conf.RecordFormatFMP4,
},
"path1",
)
require.NoError(t, err)
require.Equal(t, []*Segment{
{
Fpath: filepath.Join(dir, "path1", "2015-05-19_22-15-25-000427.mp4"),
Start: time.Date(2015, 5, 19, 22, 15, 25, 427000, time.Local),
},
{
Fpath: filepath.Join(dir, "path1", "2016-05-19_22-15-25-000427.mp4"),
Start: time.Date(2016, 5, 19, 22, 15, 25, 427000, time.Local),
},
}, segments)
}
func TestFindSegmentsInTimespan(t *testing.T) {
dir, err := os.MkdirTemp("", "mediamtx-recordstore")
require.NoError(t, err)
defer os.RemoveAll(dir)
err = os.Mkdir(filepath.Join(dir, "path1"), 0o755)
require.NoError(t, err)
err = os.Mkdir(filepath.Join(dir, "path2"), 0o755)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, "path1", "2015-05-19_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
err = os.WriteFile(filepath.Join(dir, "path1", "2016-05-19_22-15-25-000427.mp4"), []byte{1}, 0o644)
require.NoError(t, err)
segments, err := FindSegmentsInTimespan(
&conf.Path{
Name: "~^.*$",
Regexp: regexp.MustCompile("^.*$"),
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"),
RecordFormat: conf.RecordFormatFMP4,
},
"path1",
time.Date(2015, 5, 19, 22, 18, 25, 427000, time.Local),
60*time.Minute,
)
require.NoError(t, err)
require.Equal(t, []*Segment{
{
Fpath: filepath.Join(dir, "path1", "2015-05-19_22-15-25-000427.mp4"),
Start: time.Date(2015, 5, 19, 22, 15, 25, 427000, time.Local),
},
}, segments)
}