add mp4 download

CHANGELOG_PENDING.md

@@ -9,6 +9,7 @@
 #### Broadcaster
 
 - \#1877 Refresh TicketParams for the active session before expiry (@kyriediculous)
+- \#1879 Add mp4 download of recorded stream (@darkdarkdragon)
 
 #### Orchestrator
 

core/playlistmanager.go

@@ -61,6 +61,10 @@ type jsonSeg struct {
 	discontinuity bool
 }
 
+func (js *jsonSeg) GetDiscontinuity() bool {
+	return js.discontinuity
+}
+
 type JsonPlaylist struct {
 	name       string
 	DurationMs uint64               `json:"duration_ms,omitempty"` // total duration of the saved sagments

go.mod

@@ -33,7 +33,7 @@ require (
 	github.com/jackpal/go-nat-pmp v1.0.1 // indirect
 	github.com/jaypipes/ghw v0.7.0
 	github.com/karalabe/usb v0.0.0-20190919080040-51dc0efba356 // indirect
-	github.com/livepeer/lpms v0.0.0-20210505021749-6e5c36facc44
+	github.com/livepeer/lpms v0.0.0-20210518190748-f7f249116025
 	github.com/livepeer/m3u8 v0.11.1
 	github.com/mattn/go-colorable v0.1.2 // indirect
 	github.com/mattn/go-sqlite3 v1.11.0

go.sum

@@ -223,8 +223,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/livepeer/joy4 v0.1.2-0.20191121080656-b2fea45cbded h1:ZQlvR5RB4nfT+cOQee+WqmaDOgGtP2oDMhcVvR4L0yA=
 github.com/livepeer/joy4 v0.1.2-0.20191121080656-b2fea45cbded/go.mod h1:xkDdm+akniYxVT9KW1Y2Y7Hso6aW+rZObz3nrA9yTHw=
-github.com/livepeer/lpms v0.0.0-20210505021749-6e5c36facc44 h1:mplEHprPN5ke24B9JEbDWkkNwH2vm7OSDDjbRSkHisk=
-github.com/livepeer/lpms v0.0.0-20210505021749-6e5c36facc44/go.mod h1:POdMzwnvPmf6UgRkaXGP/ZI7akrzHhzjCNygZej3gzc=
+github.com/livepeer/lpms v0.0.0-20210518190748-f7f249116025 h1:AsTBqU1zrRR4uM5hRv7WwscvSVB9eHFmDqaN5U6zxkY=
+github.com/livepeer/lpms v0.0.0-20210518190748-f7f249116025/go.mod h1:POdMzwnvPmf6UgRkaXGP/ZI7akrzHhzjCNygZej3gzc=
 github.com/livepeer/m3u8 v0.11.1 h1:VkUJzfNTyjy9mqsgp5JPvouwna8wGZMvd/gAfT5FinU=
 github.com/livepeer/m3u8 v0.11.1/go.mod h1:IUqAtwWPAG2CblfQa4SVzTQoDcEMPyfNOaBSxqHMS04=
 github.com/mattn/go-colorable v0.1.2 h1:/bC9yWikZXAL9uJdulbSfyVNIR3n3trXl+v8+1sx8mU=

install_ffmpeg.sh

@@ -124,7 +124,7 @@ if [ ! -e "$HOME/ffmpeg/libavcodec/libavcodec.a" ]; then
     --disable-encoders --disable-decoders --disable-filters --disable-bsfs \
     --disable-postproc --disable-lzma \
     --enable-gnutls --enable-libx264 --enable-gpl \
-    --enable-protocol=https,http,rtmp,file \
+    --enable-protocol=https,http,rtmp,file,pipe \
     --enable-muxer=mpegts,hls,segment,mp4,null --enable-demuxer=flv,mpegts,mp4,mov \
     --enable-bsf=h264_mp4toannexb,aac_adtstoasc,h264_metadata,h264_redundant_pps \
     --enable-parser=aac,aac_latm,h264 \

server/mediaserver.go

@@ -16,6 +16,7 @@ import (
 	"net/http"
 	"net/textproto"
 	"net/url"
+	"os"
 	"path"
 	"regexp"
 	"runtime"
@@ -1037,6 +1038,98 @@ func getPlaylistsFromStore(ctx context.Context, sess drivers.OSSession, manifest
 	return filesMap, jsonFiles, latestPlaylistTime, nil
 }
 
+func (s *LivepeerServer) streamMP4(w http.ResponseWriter, r *http.Request, jpl *core.JsonPlaylist, manifestID, track string) {
+	w.Header().Set("Access-Control-Allow-Origin", "*")
+	contentType, _ := common.TypeByExtension(".mp4")
+	w.Header().Set("Content-Type", contentType)
+	w.Header().Set("Connection", "keep-alive")
+	var sourceBytesSent, resultBytesSent int64
+
+	or, ow, err := os.Pipe()
+	if err != nil {
+		glog.Errorf("Error creating pipe manifestID=%s err=%v", manifestID, err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	tc := ffmpeg.NewTranscoder()
+	done := make(chan struct{})
+	go func() {
+		var err2 error
+		resultBytesSent, err2 = io.Copy(w, or)
+		if err2 != nil {
+			glog.Errorf("Error transmuxing to mp4 request=%s manifestID=%s err=%v", r.URL.String(), manifestID, err2)
+		}
+		or.Close()
+		done <- struct{}{}
+	}()
+	defer func() {
+		tc.StopTranscoder()
+		ow.Close()
+		<-done
+	}()
+	oname := fmt.Sprintf("pipe:%d", ow.Fd())
+	out := []ffmpeg.TranscodeOptions{
+		{
+			Oname: oname,
+			VideoEncoder: ffmpeg.ComponentOptions{
+				Name: "copy",
+			},
+			AudioEncoder: ffmpeg.ComponentOptions{
+				Name: "copy",
+			},
+			Profile: ffmpeg.VideoProfile{Format: ffmpeg.FormatNone},
+			Muxer: ffmpeg.ComponentOptions{
+				Name: "mp4",
+				// main option is 'frag_keyframe' which tells ffmpeg to create fragmented MP4 (which we need to be able to stream generatd file)
+				// other options is not mandatory but they will slightly improve generated MP4 file
+				Opts: map[string]string{"movflags": "frag_keyframe+negative_cts_offsets+omit_tfhd_offset+disable_chpl+default_base_moof"},
+			},
+		},
+	}
+	for _, seg := range jpl.Segments[track] {
+		if seg.GetDiscontinuity() {
+			tc.Discontinuity()
+		}
+
+		ir, iw, err := os.Pipe()
+		if err != nil {
+			glog.Errorf("Error creating pipe manifestID=%s err=%v", manifestID, err)
+			return
+		}
+		fname := fmt.Sprintf("pipe:%d", ir.Fd())
+
+		in := &ffmpeg.TranscodeOptionsIn{Fname: fname, Transmuxing: true}
+		go func() {
+			defer iw.Close()
+			glog.V(common.VERBOSE).Infof("Adding manifestID=%s track=%s uri=%s to mp4", manifestID, track, seg.URI)
+			resp, err := http.Get(seg.URI)
+			if err != nil {
+				glog.Errorf("Error getting HTTP uri=%s manifestID=%s err=%v", seg.URI, manifestID, err)
+				return
+			}
+			defer resp.Body.Close()
+			if resp.StatusCode != 200 {
+				glog.Errorf("Non-200 response for status=%v uri=%s manifestID=%s request=%s", resp.Status, seg.URI, manifestID, r.URL.String())
+				return
+			}
+			wn, err := io.Copy(iw, resp.Body)
+			if err != nil {
+				glog.Errorf("Error transmuxing to mp4 request=%s uri=%s manifestID=%s err=%v", r.URL.String(), seg.URI, manifestID, err)
+			}
+			sourceBytesSent += wn
+		}()
+
+		_, err = tc.Transcode(in, out)
+		ir.Close()
+		if err != nil {
+			glog.Errorf("Error transmuxing to mp4 request=%s uri=%s manifestID=%s err=%v", r.URL.String(), seg.URI, manifestID, err)
+			return
+		}
+	}
+	glog.Infof("Completed mp4 request=%s manifestID=%s sourceBytes=%d destBytes=%d", r.URL.String(),
+		manifestID, sourceBytesSent, resultBytesSent)
+}
+
 // HandleRecordings handle requests to /recordings/ endpoint
 func (s *LivepeerServer) HandleRecordings(w http.ResponseWriter, r *http.Request) {
 	if r.Method != "GET" {
@@ -1045,7 +1138,7 @@ func (s *LivepeerServer) HandleRecordings(w http.ResponseWriter, r *http.Request
 		return
 	}
 	ext := path.Ext(r.URL.Path)
-	if ext != ".m3u8" && ext != ".ts" {
+	if ext != ".m3u8" && ext != ".ts" && ext != ".mp4" {
 		glog.Errorf(`/recordings request wrong extension=%s url=%s host=%s`, ext, r.URL, r.Host)
 		w.WriteHeader(http.StatusBadRequest)
 		return
@@ -1154,7 +1247,7 @@ func (s *LivepeerServer) HandleRecordings(w http.ResponseWriter, r *http.Request
 		w.WriteHeader(http.StatusNotFound)
 		return
 	}
-	if time.Since(latestPlaylistTime) > 24*time.Hour && !finalizeSet {
+	if time.Since(latestPlaylistTime) > 24*time.Hour && !finalizeSet && ext == ".m3u8" {
 		finalize = true
 	}
 
@@ -1210,6 +1303,14 @@ func (s *LivepeerServer) HandleRecordings(w http.ResponseWriter, r *http.Request
 			}
 		}
 	}
+	if ext == ".mp4" {
+		if segs, has := mainJspl.Segments[track]; !has || len(segs) == 0 {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		s.streamMP4(w, r, mainJspl, manifestID, track)
+		return
+	}
 
 	masterPList := m3u8.NewMasterPlaylist()
 	mediaLists := make(map[string]*m3u8.MediaPlaylist)