From 73a300afd0b6d37ebec09f3a5bdfbe861145eb2b Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Sun, 8 Sep 2024 20:33:18 +0200 Subject: [PATCH] fix cleaning of recordings in case of multiple recordDeleteAfter values (#3557) (#3741) --- internal/api/api.go | 39 ++- internal/api/api_test.go | 1 + internal/api/record.go | 137 ---------- internal/conf/path.go | 14 +- internal/core/core.go | 51 +--- internal/core/path.go | 25 +- internal/core/path_manager.go | 34 ++- internal/playback/on_get.go | 9 +- internal/playback/on_get_test.go | 3 + internal/playback/on_list.go | 10 +- internal/playback/on_list_test.go | 2 + internal/playback/segment.go | 140 ---------- internal/playback/segment_fmp4.go | 3 +- internal/playback/server.go | 4 +- internal/record/cleaner.go | 124 --------- internal/record/cleaner_test.go | 52 ---- internal/record/record.go | 2 - internal/recordcleaner/cleaner.go | 134 ++++++++++ internal/recordcleaner/cleaner_test.go | 107 ++++++++ internal/{record => recorder}/format.go | 2 +- internal/{record => recorder}/format_fmp4.go | 2 +- .../{record => recorder}/format_fmp4_part.go | 5 +- .../format_fmp4_segment.go | 2 +- .../{record => recorder}/format_fmp4_track.go | 2 +- .../{record => recorder}/format_mpegts.go | 2 +- .../format_mpegts_segment.go | 5 +- .../recoder_instance.go} | 7 +- .../{record/agent.go => recorder/recorder.go} | 19 +- .../recorder_test.go} | 18 +- internal/{record => recordstore}/path.go | 4 +- internal/{record => recordstore}/path_test.go | 2 +- internal/recordstore/recordstore.go | 2 + internal/recordstore/segment.go | 240 ++++++++++++++++++ internal/recordstore/segment_test.go | 123 +++++++++ 34 files changed, 738 insertions(+), 588 deletions(-) delete mode 100644 internal/api/record.go delete mode 100644 internal/playback/segment.go delete mode 100644 internal/record/cleaner.go delete mode 100644 internal/record/cleaner_test.go delete mode 100644 internal/record/record.go create mode 100644 internal/recordcleaner/cleaner.go create mode 100644 internal/recordcleaner/cleaner_test.go rename internal/{record => recorder}/format.go (74%) rename internal/{record => recorder}/format_fmp4.go (99%) rename internal/{record => recorder}/format_fmp4_part.go (93%) rename internal/{record => recorder}/format_fmp4_segment.go (99%) rename internal/{record => recorder}/format_fmp4_track.go (98%) rename internal/{record => recorder}/format_mpegts.go (99%) rename internal/{record => recorder}/format_mpegts_segment.go (88%) rename internal/{record/agent_instance.go => recorder/recoder_instance.go} (91%) rename internal/{record/agent.go => recorder/recorder.go} (82%) rename internal/{record/agent_test.go => recorder/recorder_test.go} (96%) rename internal/{record => recordstore}/path.go (98%) rename internal/{record => recordstore}/path_test.go (97%) create mode 100644 internal/recordstore/recordstore.go create mode 100644 internal/recordstore/segment.go create mode 100644 internal/recordstore/segment_test.go diff --git a/internal/api/api.go b/internal/api/api.go index c0324624..b3c77d6a 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -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) diff --git a/internal/api/api_test.go b/internal/api/api_test.go index a2a6a5ec..d7d026e0 100644 --- a/internal/api/api_test.go +++ b/internal/api/api_test.go @@ -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{ diff --git a/internal/api/record.go b/internal/api/record.go deleted file mode 100644 index 82a1e7b5..00000000 --- a/internal/api/record.go +++ /dev/null @@ -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 -} diff --git a/internal/conf/path.go b/internal/conf/path.go index d6735dab..0ae1e0bc 100644 --- a/internal/conf/path.go +++ b/internal/conf/path.go @@ -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. diff --git a/internal/core/core.go b/internal/core/core.go index baf643cc..d55a8d8e 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -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 || diff --git a/internal/core/path.go b/internal/core/path.go index 2119927c..254122db 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -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) { diff --git a/internal/core/path_manager.go b/internal/core/path_manager.go index 623ac6a7..a3495864 100644 --- a/internal/core/path_manager.go +++ b/internal/core/path_manager.go @@ -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) } diff --git a/internal/playback/on_get.go b/internal/playback/on_get.go index 77ab0117..5676441c 100644 --- a/internal/playback/on_get.go +++ b/internal/playback/on_get.go @@ -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) diff --git a/internal/playback/on_get_test.go b/internal/playback/on_get_test.go index 86acb41c..528fc724 100644 --- a/internal/playback/on_get_test.go +++ b/internal/playback/on_get_test.go @@ -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"), }, }, diff --git a/internal/playback/on_list.go b/internal/playback/on_list.go index dd2eb639..219c171a 100644 --- a/internal/playback/on_list.go +++ b/internal/playback/on_list.go @@ -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) diff --git a/internal/playback/on_list_test.go b/internal/playback/on_list_test.go index 9283e51d..a05843c6 100644 --- a/internal/playback/on_list_test.go +++ b/internal/playback/on_list_test.go @@ -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"), }, }, diff --git a/internal/playback/segment.go b/internal/playback/segment.go deleted file mode 100644 index 5bd10574..00000000 --- a/internal/playback/segment.go +++ /dev/null @@ -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 -} diff --git a/internal/playback/segment_fmp4.go b/internal/playback/segment_fmp4.go index e1619f8f..d92cbc6f 100644 --- a/internal/playback/segment_fmp4.go +++ b/internal/playback/segment_fmp4.go @@ -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 diff --git a/internal/playback/server.go b/internal/playback/server.go index 52086169..dca99feb 100644 --- a/internal/playback/server.go +++ b/internal/playback/server.go @@ -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 } diff --git a/internal/record/cleaner.go b/internal/record/cleaner.go deleted file mode 100644 index f285be42..00000000 --- a/internal/record/cleaner.go +++ /dev/null @@ -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 -} diff --git a/internal/record/cleaner_test.go b/internal/record/cleaner_test.go deleted file mode 100644 index dbdea55e..00000000 --- a/internal/record/cleaner_test.go +++ /dev/null @@ -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) -} diff --git a/internal/record/record.go b/internal/record/record.go deleted file mode 100644 index 9d5b89ab..00000000 --- a/internal/record/record.go +++ /dev/null @@ -1,2 +0,0 @@ -// Package record contains the recording system. -package record diff --git a/internal/recordcleaner/cleaner.go b/internal/recordcleaner/cleaner.go new file mode 100644 index 00000000..76dffe36 --- /dev/null +++ b/internal/recordcleaner/cleaner.go @@ -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 +} diff --git a/internal/recordcleaner/cleaner_test.go b/internal/recordcleaner/cleaner_test.go new file mode 100644 index 00000000..fd5e4ac0 --- /dev/null +++ b/internal/recordcleaner/cleaner_test.go @@ -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) +} diff --git a/internal/record/format.go b/internal/recorder/format.go similarity index 74% rename from internal/record/format.go rename to internal/recorder/format.go index 0ccc55cd..6f816f2a 100644 --- a/internal/record/format.go +++ b/internal/recorder/format.go @@ -1,4 +1,4 @@ -package record +package recorder type format interface { initialize() diff --git a/internal/record/format_fmp4.go b/internal/recorder/format_fmp4.go similarity index 99% rename from internal/record/format_fmp4.go rename to internal/recorder/format_fmp4.go index 462f24fd..156aea32 100644 --- a/internal/record/format_fmp4.go +++ b/internal/recorder/format_fmp4.go @@ -1,4 +1,4 @@ -package record +package recorder import ( "bytes" diff --git a/internal/record/format_fmp4_part.go b/internal/recorder/format_fmp4_part.go similarity index 93% rename from internal/record/format_fmp4_part.go rename to internal/recorder/format_fmp4_part.go index c8871d3d..b8ed532d 100644 --- a/internal/record/format_fmp4_part.go +++ b/internal/recorder/format_fmp4_part.go @@ -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) diff --git a/internal/record/format_fmp4_segment.go b/internal/recorder/format_fmp4_segment.go similarity index 99% rename from internal/record/format_fmp4_segment.go rename to internal/recorder/format_fmp4_segment.go index 476e5183..5c1d1519 100644 --- a/internal/record/format_fmp4_segment.go +++ b/internal/recorder/format_fmp4_segment.go @@ -1,4 +1,4 @@ -package record +package recorder import ( "io" diff --git a/internal/record/format_fmp4_track.go b/internal/recorder/format_fmp4_track.go similarity index 98% rename from internal/record/format_fmp4_track.go rename to internal/recorder/format_fmp4_track.go index 7107fa75..537fd829 100644 --- a/internal/record/format_fmp4_track.go +++ b/internal/recorder/format_fmp4_track.go @@ -1,4 +1,4 @@ -package record +package recorder import ( "github.com/bluenviron/mediacommon/pkg/formats/fmp4" diff --git a/internal/record/format_mpegts.go b/internal/recorder/format_mpegts.go similarity index 99% rename from internal/record/format_mpegts.go rename to internal/recorder/format_mpegts.go index ced04950..c70764e0 100644 --- a/internal/record/format_mpegts.go +++ b/internal/recorder/format_mpegts.go @@ -1,4 +1,4 @@ -package record +package recorder import ( "bufio" diff --git a/internal/record/format_mpegts_segment.go b/internal/recorder/format_mpegts_segment.go similarity index 88% rename from internal/record/format_mpegts_segment.go rename to internal/recorder/format_mpegts_segment.go index cf3dc705..8b7ff619 100644 --- a/internal/record/format_mpegts_segment.go +++ b/internal/recorder/format_mpegts_segment.go @@ -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) diff --git a/internal/record/agent_instance.go b/internal/recorder/recoder_instance.go similarity index 91% rename from internal/record/agent_instance.go rename to internal/recorder/recoder_instance.go index 37af8403..a24e5b96 100644 --- a/internal/record/agent_instance.go +++ b/internal/recorder/recoder_instance.go @@ -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, ) diff --git a/internal/record/agent.go b/internal/recorder/recorder.go similarity index 82% rename from internal/record/agent.go rename to internal/recorder/recorder.go index f7189f67..0700b2fe 100644 --- a/internal/record/agent.go +++ b/internal/recorder/recorder.go @@ -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 { diff --git a/internal/record/agent_test.go b/internal/recorder/recorder_test.go similarity index 96% rename from internal/record/agent_test.go rename to internal/recorder/recorder_test.go index fb342e9e..cb408974 100644 --- a/internal/record/agent_test.go +++ b/internal/recorder/recorder_test.go @@ -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, diff --git a/internal/record/path.go b/internal/recordstore/path.go similarity index 98% rename from internal/record/path.go rename to internal/recordstore/path.go index fd58de69..d8aad3b9 100644 --- a/internal/record/path.go +++ b/internal/recordstore/path.go @@ -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: diff --git a/internal/record/path_test.go b/internal/recordstore/path_test.go similarity index 97% rename from internal/record/path_test.go rename to internal/recordstore/path_test.go index 80064797..f32611e0 100644 --- a/internal/record/path_test.go +++ b/internal/recordstore/path_test.go @@ -1,4 +1,4 @@ -package record +package recordstore import ( "testing" diff --git a/internal/recordstore/recordstore.go b/internal/recordstore/recordstore.go new file mode 100644 index 00000000..90c42094 --- /dev/null +++ b/internal/recordstore/recordstore.go @@ -0,0 +1,2 @@ +// Package recordstore contains utilities to store/retrieve recordings to/from disk. +package recordstore diff --git a/internal/recordstore/segment.go b/internal/recordstore/segment.go new file mode 100644 index 00000000..0be38ca9 --- /dev/null +++ b/internal/recordstore/segment.go @@ -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 +} diff --git a/internal/recordstore/segment_test.go b/internal/recordstore/segment_test.go new file mode 100644 index 00000000..28279b80 --- /dev/null +++ b/internal/recordstore/segment_test.go @@ -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) +}