Alessandro Ros
1 year ago
committed by
GitHub
7 changed files with 783 additions and 504 deletions
@ -0,0 +1,179 @@
@@ -0,0 +1,179 @@
|
||||
package playback |
||||
|
||||
import ( |
||||
"errors" |
||||
"fmt" |
||||
"io" |
||||
"net" |
||||
"net/http" |
||||
"os" |
||||
"strconv" |
||||
"time" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/logger" |
||||
"github.com/gin-gonic/gin" |
||||
) |
||||
|
||||
var errStopIteration = errors.New("stop iteration") |
||||
|
||||
func parseDuration(raw string) (time.Duration, error) { |
||||
// seconds
|
||||
if secs, err := strconv.ParseFloat(raw, 64); err == nil { |
||||
return time.Duration(secs * float64(time.Second)), nil |
||||
} |
||||
|
||||
// deprecated, golang format
|
||||
return time.ParseDuration(raw) |
||||
} |
||||
|
||||
func seekAndMux( |
||||
recordFormat conf.RecordFormat, |
||||
segments []*Segment, |
||||
start time.Time, |
||||
duration time.Duration, |
||||
w io.Writer, |
||||
) error { |
||||
if recordFormat == conf.RecordFormatFMP4 { |
||||
minTime := start.Sub(segments[0].Start) |
||||
maxTime := minTime + duration |
||||
var init []byte |
||||
var elapsed time.Duration |
||||
|
||||
err := func() error { |
||||
f, err := os.Open(segments[0].Fpath) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
defer f.Close() |
||||
|
||||
init, err = fmp4ReadInit(f) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
elapsed, err = fmp4SeekAndMuxParts(f, init, minTime, maxTime, w) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
return nil |
||||
}() |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
prevInit := init |
||||
prevEnd := start.Add(elapsed) |
||||
duration -= elapsed |
||||
overallElapsed := elapsed |
||||
|
||||
for _, seg := range segments[1:] { |
||||
err := func() error { |
||||
f, err := os.Open(seg.Fpath) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
defer f.Close() |
||||
|
||||
init, err := fmp4ReadInit(f) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
if !fmp4CanBeConcatenated(prevInit, prevEnd, init, seg.Start) { |
||||
return errStopIteration |
||||
} |
||||
|
||||
elapsed, err = fmp4MuxParts(f, overallElapsed, duration, w) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
return nil |
||||
}() |
||||
if err != nil { |
||||
if errors.Is(err, errStopIteration) { |
||||
return nil |
||||
} |
||||
|
||||
return err |
||||
} |
||||
|
||||
prevEnd = seg.Start.Add(elapsed) |
||||
duration -= elapsed |
||||
overallElapsed += elapsed |
||||
} |
||||
|
||||
return nil |
||||
} |
||||
|
||||
return fmt.Errorf("MPEG-TS format is not supported yet") |
||||
} |
||||
|
||||
func (p *Server) onGet(ctx *gin.Context) { |
||||
pathName := ctx.Query("path") |
||||
|
||||
if !p.doAuth(ctx, pathName) { |
||||
return |
||||
} |
||||
|
||||
start, err := time.Parse(time.RFC3339, ctx.Query("start")) |
||||
if err != nil { |
||||
p.writeError(ctx, http.StatusBadRequest, fmt.Errorf("invalid start: %w", err)) |
||||
return |
||||
} |
||||
|
||||
duration, err := parseDuration(ctx.Query("duration")) |
||||
if err != nil { |
||||
p.writeError(ctx, http.StatusBadRequest, fmt.Errorf("invalid duration: %w", err)) |
||||
return |
||||
} |
||||
|
||||
format := ctx.Query("format") |
||||
if format != "" && format != "fmp4" { |
||||
p.writeError(ctx, http.StatusBadRequest, fmt.Errorf("invalid format: %s", format)) |
||||
return |
||||
} |
||||
|
||||
pathConf, err := p.safeFindPathConf(pathName) |
||||
if err != nil { |
||||
p.writeError(ctx, http.StatusBadRequest, err) |
||||
return |
||||
} |
||||
|
||||
segments, err := findSegmentsInTimespan(pathConf, pathName, start, duration) |
||||
if err != nil { |
||||
if errors.Is(err, errNoSegmentsFound) { |
||||
p.writeError(ctx, http.StatusNotFound, err) |
||||
} else { |
||||
p.writeError(ctx, http.StatusBadRequest, err) |
||||
} |
||||
return |
||||
} |
||||
|
||||
ww := &writerWrapper{ctx: ctx} |
||||
|
||||
err = seekAndMux(pathConf.RecordFormat, segments, start, duration, ww) |
||||
if err != nil { |
||||
// user aborted the download
|
||||
var neterr *net.OpError |
||||
if errors.As(err, &neterr) { |
||||
return |
||||
} |
||||
|
||||
// nothing has been written yet; send back JSON
|
||||
if !ww.written { |
||||
if errors.Is(err, errNoSegmentsFound) { |
||||
p.writeError(ctx, http.StatusNotFound, err) |
||||
} else { |
||||
p.writeError(ctx, http.StatusBadRequest, err) |
||||
} |
||||
return |
||||
} |
||||
|
||||
// something has already been written: abort and write logs only
|
||||
p.Log(logger.Error, err.Error()) |
||||
return |
||||
} |
||||
} |
||||
@ -0,0 +1,120 @@
@@ -0,0 +1,120 @@
|
||||
package playback |
||||
|
||||
import ( |
||||
"encoding/json" |
||||
"errors" |
||||
"fmt" |
||||
"io" |
||||
"net/http" |
||||
"os" |
||||
"time" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/gin-gonic/gin" |
||||
) |
||||
|
||||
type listEntryDuration time.Duration |
||||
|
||||
func (d listEntryDuration) MarshalJSON() ([]byte, error) { |
||||
return json.Marshal(time.Duration(d).Seconds()) |
||||
} |
||||
|
||||
type listEntry struct { |
||||
Start time.Time `json:"start"` |
||||
Duration listEntryDuration `json:"duration"` |
||||
} |
||||
|
||||
func computeDurationAndConcatenate(recordFormat conf.RecordFormat, segments []*Segment) ([]listEntry, error) { |
||||
if recordFormat == conf.RecordFormatFMP4 { |
||||
out := []listEntry{} |
||||
var prevInit []byte |
||||
|
||||
for _, seg := range segments { |
||||
err := func() error { |
||||
f, err := os.Open(seg.Fpath) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
defer f.Close() |
||||
|
||||
init, err := fmp4ReadInit(f) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
_, err = f.Seek(0, io.SeekStart) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
duration, err := fmp4ReadDuration(f) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
if len(out) != 0 && fmp4CanBeConcatenated( |
||||
prevInit, |
||||
out[len(out)-1].Start.Add(time.Duration(out[len(out)-1].Duration)), |
||||
init, |
||||
seg.Start) { |
||||
prevStart := out[len(out)-1].Start |
||||
curEnd := seg.Start.Add(duration) |
||||
out[len(out)-1].Duration = listEntryDuration(curEnd.Sub(prevStart)) |
||||
} else { |
||||
out = append(out, listEntry{ |
||||
Start: seg.Start, |
||||
Duration: listEntryDuration(duration), |
||||
}) |
||||
} |
||||
|
||||
prevInit = init |
||||
|
||||
return nil |
||||
}() |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
} |
||||
|
||||
return out, nil |
||||
} |
||||
|
||||
return nil, fmt.Errorf("MPEG-TS format is not supported yet") |
||||
} |
||||
|
||||
func (p *Server) onList(ctx *gin.Context) { |
||||
pathName := ctx.Query("path") |
||||
|
||||
if !p.doAuth(ctx, pathName) { |
||||
return |
||||
} |
||||
|
||||
pathConf, err := p.safeFindPathConf(pathName) |
||||
if err != nil { |
||||
p.writeError(ctx, http.StatusBadRequest, err) |
||||
return |
||||
} |
||||
|
||||
if !pathConf.Playback { |
||||
p.writeError(ctx, http.StatusBadRequest, fmt.Errorf("playback is disabled on path '%s'", pathName)) |
||||
return |
||||
} |
||||
|
||||
segments, err := FindSegments(pathConf, pathName) |
||||
if err != nil { |
||||
if errors.Is(err, errNoSegmentsFound) { |
||||
p.writeError(ctx, http.StatusNotFound, err) |
||||
} else { |
||||
p.writeError(ctx, http.StatusBadRequest, err) |
||||
} |
||||
return |
||||
} |
||||
|
||||
out, err := computeDurationAndConcatenate(pathConf.RecordFormat, segments) |
||||
if err != nil { |
||||
p.writeError(ctx, http.StatusInternalServerError, err) |
||||
return |
||||
} |
||||
|
||||
ctx.JSON(http.StatusOK, out) |
||||
} |
||||
@ -0,0 +1,134 @@
@@ -0,0 +1,134 @@
|
||||
package playback |
||||
|
||||
import ( |
||||
"encoding/json" |
||||
"net/http" |
||||
"net/url" |
||||
"os" |
||||
"path/filepath" |
||||
"testing" |
||||
"time" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/test" |
||||
"github.com/stretchr/testify/require" |
||||
) |
||||
|
||||
func TestOnList(t *testing.T) { |
||||
dir, err := os.MkdirTemp("", "mediamtx-playback") |
||||
require.NoError(t, err) |
||||
defer os.RemoveAll(dir) |
||||
|
||||
err = os.Mkdir(filepath.Join(dir, "mypath"), 0o755) |
||||
require.NoError(t, err) |
||||
|
||||
writeSegment1(t, filepath.Join(dir, "mypath", "2008-11-07_11-22-00-500000.mp4")) |
||||
writeSegment2(t, filepath.Join(dir, "mypath", "2008-11-07_11-23-02-500000.mp4")) |
||||
writeSegment2(t, filepath.Join(dir, "mypath", "2009-11-07_11-23-02-500000.mp4")) |
||||
|
||||
s := &Server{ |
||||
Address: "127.0.0.1:9996", |
||||
ReadTimeout: conf.StringDuration(10 * time.Second), |
||||
PathConfs: map[string]*conf.Path{ |
||||
"mypath": { |
||||
Playback: true, |
||||
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"), |
||||
}, |
||||
}, |
||||
AuthManager: authManager, |
||||
Parent: &test.NilLogger{}, |
||||
} |
||||
err = s.Initialize() |
||||
require.NoError(t, err) |
||||
defer s.Close() |
||||
|
||||
u, err := url.Parse("http://myuser:mypass@localhost:9996/list") |
||||
require.NoError(t, err) |
||||
|
||||
v := url.Values{} |
||||
v.Set("path", "mypath") |
||||
u.RawQuery = v.Encode() |
||||
|
||||
req, err := http.NewRequest(http.MethodGet, u.String(), nil) |
||||
require.NoError(t, err) |
||||
|
||||
res, err := http.DefaultClient.Do(req) |
||||
require.NoError(t, err) |
||||
defer res.Body.Close() |
||||
|
||||
require.Equal(t, http.StatusOK, res.StatusCode) |
||||
|
||||
var out interface{} |
||||
err = json.NewDecoder(res.Body).Decode(&out) |
||||
require.NoError(t, err) |
||||
|
||||
require.Equal(t, []interface{}{ |
||||
map[string]interface{}{ |
||||
"duration": float64(64), |
||||
"start": time.Date(2008, 11, 0o7, 11, 22, 0, 500000000, time.Local).Format(time.RFC3339Nano), |
||||
}, |
||||
map[string]interface{}{ |
||||
"duration": float64(2), |
||||
"start": time.Date(2009, 11, 0o7, 11, 23, 2, 500000000, time.Local).Format(time.RFC3339Nano), |
||||
}, |
||||
}, out) |
||||
} |
||||
|
||||
func TestOnListDifferentInit(t *testing.T) { |
||||
dir, err := os.MkdirTemp("", "mediamtx-playback") |
||||
require.NoError(t, err) |
||||
defer os.RemoveAll(dir) |
||||
|
||||
err = os.Mkdir(filepath.Join(dir, "mypath"), 0o755) |
||||
require.NoError(t, err) |
||||
|
||||
writeSegment1(t, filepath.Join(dir, "mypath", "2008-11-07_11-22-00-500000.mp4")) |
||||
writeSegment3(t, filepath.Join(dir, "mypath", "2008-11-07_11-23-02-500000.mp4")) |
||||
|
||||
s := &Server{ |
||||
Address: "127.0.0.1:9996", |
||||
ReadTimeout: conf.StringDuration(10 * time.Second), |
||||
PathConfs: map[string]*conf.Path{ |
||||
"mypath": { |
||||
Playback: true, |
||||
RecordPath: filepath.Join(dir, "%path/%Y-%m-%d_%H-%M-%S-%f"), |
||||
}, |
||||
}, |
||||
AuthManager: authManager, |
||||
Parent: &test.NilLogger{}, |
||||
} |
||||
err = s.Initialize() |
||||
require.NoError(t, err) |
||||
defer s.Close() |
||||
|
||||
u, err := url.Parse("http://myuser:mypass@localhost:9996/list") |
||||
require.NoError(t, err) |
||||
|
||||
v := url.Values{} |
||||
v.Set("path", "mypath") |
||||
u.RawQuery = v.Encode() |
||||
|
||||
req, err := http.NewRequest(http.MethodGet, u.String(), nil) |
||||
require.NoError(t, err) |
||||
|
||||
res, err := http.DefaultClient.Do(req) |
||||
require.NoError(t, err) |
||||
defer res.Body.Close() |
||||
|
||||
require.Equal(t, http.StatusOK, res.StatusCode) |
||||
|
||||
var out interface{} |
||||
err = json.NewDecoder(res.Body).Decode(&out) |
||||
require.NoError(t, err) |
||||
|
||||
require.Equal(t, []interface{}{ |
||||
map[string]interface{}{ |
||||
"duration": float64(62), |
||||
"start": time.Date(2008, 11, 0o7, 11, 22, 0, 500000000, time.Local).Format(time.RFC3339Nano), |
||||
}, |
||||
map[string]interface{}{ |
||||
"duration": float64(1), |
||||
"start": time.Date(2008, 11, 0o7, 11, 23, 2, 500000000, time.Local).Format(time.RFC3339Nano), |
||||
}, |
||||
}, out) |
||||
} |
||||
Loading…
Reference in new issue