Improvements with concurrent encodings & stopping videos when leaving…

… page

NOTES.txt

@@ -0,0 +1,3 @@
+
+
+https://bitbucket.org/walterebert/ffmpeg-hls/src/f14ccab90f1884ecee9bbccd025af87a3b2f837a/build.sh?at=default&fileviewer=file-view-default

README.md

@@ -15,5 +15,4 @@ Running it
 
 License
 -------
-See LICENSE.txt
-
+See LICENSE.txt

cache/.gitignore

@@ -0,0 +1,5 @@
+# Ignore everything in this directory
+*
+# Except this file
+!.gitignore
+!README.md

http_command.go

@@ -0,0 +1,110 @@
+package main
+
+import (
+	"sync"
+	//"net/http"
+	"bufio"
+	"crypto/sha1"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"syscall"
+)
+
+type Empty struct{}
+
+type HttpCommandHandler struct {
+	tokenChannel    chan Empty
+	cacheDir        string
+	inProgress      map[string]string
+	inProgressMutex *sync.RWMutex
+	// path string
+}
+
+func NewHttpCommandHandler(workerCount int, cacheDir string) *HttpCommandHandler {
+	ch := &HttpCommandHandler{make(chan Empty, workerCount), cacheDir, make(map[string]string), new(sync.RWMutex)}
+	for i := workerCount; i > 0; i-- {
+		ch.tokenChannel <- Empty{}
+	}
+	go ch.start()
+	return ch
+}
+
+func (s *HttpCommandHandler) start() {
+
+}
+
+func (s *HttpCommandHandler) calculateKey(cmd string, args []string) string {
+	h := sha1.New()
+	h.Write([]byte(cmd))
+	for _, v := range args {
+		h.Write([]byte(v))
+	}
+	sum := h.Sum(nil)
+	return fmt.Sprintf("%x", sum)
+}
+
+func (s *HttpCommandHandler) ServeCommand(cmdPath string, args []string, w io.Writer) error {
+	key := s.calculateKey(cmdPath, args)
+	token := <-s.tokenChannel
+	//log.Printf("Token: %v",key)
+	defer func() {
+		s.tokenChannel <- token
+		//log.Printf("Released token")
+	}()
+	cachePath := filepath.Join("cache", s.cacheDir, key)
+	mkerr := os.MkdirAll(filepath.Join("cache", s.cacheDir), 0777)
+	if mkerr != nil {
+		log.Printf("Could not create cache dir %v: %v", filepath.Join("cache", s.cacheDir), mkerr)
+		return mkerr
+	}
+	if file, err := os.Open(cachePath); err == nil {
+		defer file.Close()
+		_, err = io.Copy(w, file)
+		if err != nil {
+			log.Printf("Error copying file to client: %v", err)
+			return err
+		}
+		return nil
+	}
+	cacheFile, ferr := os.Create(cachePath)
+	if ferr != nil {
+		log.Printf("Could not create cache file %v: %v", cacheFile, ferr)
+		return ferr
+	}
+	defer cacheFile.Close()
+	log.Printf("Executing %v %v", cmdPath, args)
+	cmd := exec.Command(cmdPath, args...)
+	stdout, err := cmd.StdoutPipe()
+	defer stdout.Close()
+	if err != nil {
+		log.Printf("Error opening stdout of command: %v", err)
+		return err
+	}
+	err = cmd.Start()
+	if err != nil {
+		log.Printf("Error starting command: %v", err)
+		return err
+	}
+	filew := bufio.NewWriter(cacheFile)
+	multiw := io.MultiWriter(filew, w)
+	_, err = io.Copy(multiw, stdout)
+	if err != nil {
+		log.Printf("Error copying data to client: %v", err)
+		cacheFile.Close()
+		os.Remove(cachePath)
+		// Ask the process to exit
+		cmd.Process.Signal(syscall.SIGKILL)
+		cmd.Process.Wait()
+		return err
+	}
+	cmd.Wait()
+	filew.Flush()
+	log.Printf("Streaming done\n");
+	return nil
+	//s.inProgressMutex.Lock()
+	//s.inProgressMutex.Unlock()
+}

http_frame.go

@@ -3,22 +3,21 @@ package main
 import (
 	"log"
 	"net/http"
-	"os/exec"
 	"path"
 )
 
 type FrameHandler struct {
-	root string
+	root       string
+	cmdHandler *HttpCommandHandler
 }
 
 func NewFrameHandler(root string) *FrameHandler {
-	return &FrameHandler{root}
+	return &FrameHandler{root, NewHttpCommandHandler(2, "frames")}
 }
 
 func (s *FrameHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	path := path.Join(s.root, r.URL.Path)
-	cmd := exec.Command("tools/ffmpeg", "-loglevel", "error", "-ss", "00:00:30", "-i", path, "-vf", "scale=320:-1", "-frames:v", "1", "-f", "image2", "-")
-	if err := ServeCommand(cmd, w); err != nil {
+	if err := s.cmdHandler.ServeCommand("tools/ffmpeg", []string{"-timelimit","10","-loglevel", "error", "-ss", "00:00:30", "-i", path, "-vf", "scale=320:-1", "-frames:v", "1", "-f", "image2", "-"}, w); err != nil {
 		log.Printf("Error serving screenshot: %v", err)
 	}
 }

http_playlists.go

@@ -16,7 +16,7 @@ func UrlEncoded(str string) (string, error) {
 	return u.String(), nil
 }
 
-const hlsSegmentLength = 5.0 // 5 Seconds
+const hlsSegmentLength = 5.0 // 10 Seconds
 
 type PlaylistHandler struct {
 	root string
@@ -49,7 +49,7 @@ func (s *PlaylistHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	fmt.Fprint(w, "#EXT-X-VERSION:3\n")
 	fmt.Fprint(w, "#EXT-X-MEDIA-SEQUENCE:0\n")
 	fmt.Fprint(w, "#EXT-X-ALLOW-CACHE:YES\n")
-	fmt.Fprint(w, "#EXT-X-TARGETDURATION:5\n")
+	fmt.Fprint(w, "#EXT-X-TARGETDURATION:"+fmt.Sprintf("%.f",hlsSegmentLength)+"\n")
 	fmt.Fprint(w, "#EXT-X-PLAYLIST-TYPE:VOD\n")
 
 	leftover := duration

http_stream.go

@@ -2,29 +2,35 @@ package main
 
 import (
 	"fmt"
+	"log"
 	"net/http"
-	"os/exec"
 	"path"
 	"strconv"
 	"strings"
 )
 
 type StreamHandler struct {
-	root string
+	root       string
+	cmdHandler *HttpCommandHandler
 }
 
 func NewStreamHandler(root string) *StreamHandler {
-	return &StreamHandler{root}
+	return &StreamHandler{root, NewHttpCommandHandler(1, "segments")}
 }
 
 func (s *StreamHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	filePath := path.Join(s.root, r.URL.Path[0:strings.LastIndex(r.URL.Path, "/")])
 	idx, _ := strconv.ParseInt(r.URL.Path[strings.LastIndex(r.URL.Path, "/")+1:strings.LastIndex(r.URL.Path, ".")], 0, 64)
 	startTime := idx * hlsSegmentLength
 	debug.Printf("Streaming second %v of %v", startTime, filePath)
-
 	w.Header()["Access-Control-Allow-Origin"] = []string{"*"}
-
-	cmd := exec.Command("tools/ffmpeg", "-ss", fmt.Sprintf("%v", startTime), "-t", "5", "-i", filePath, "-vcodec", "libx264", "-strict", "experimental", "-acodec", "aac", "-pix_fmt", "yuv420p", "-r", "25", "-profile:v", "baseline", "-b:v", "2000k", "-maxrate", "2500k", "-f", "mpegts", "-")
-	ServeCommand(cmd, w)
+	// -ss = offset
+	// -t = duration
+	// -i = input
+	// -vcodec = video codec
+	// -acodec = audio codec
+	// --timelimit
+	if err := s.cmdHandler.ServeCommand("tools/ffmpeg", []string{"-timelimit","30","-ss", fmt.Sprintf("%v.00", startTime), "-t", fmt.Sprintf("%v.00", hlsSegmentLength), "-i", filePath, "-strict", "-2", "-vcodec", "libx264", "-acodec", "aac", "-pix_fmt", "yuv420p", "-r", "25", "-f", "mpegts", "-force_key_frames", "00:00:00.00", "-x264opts", "keyint=25:min-keyint=25:scenecut=-1", "-"}, w); err != nil {
+		log.Printf("Error streaming file %v and segment %v", filePath, idx)
+	}
 }

http_util.go

@@ -2,37 +2,9 @@ package main
 
 import (
 	"encoding/json"
-	"io"
-	"log"
 	"net/http"
-	"os/exec"
-	"syscall"
 )
 
-func ServeCommand(cmd *exec.Cmd, w io.Writer) error {
-	stdout, err := cmd.StdoutPipe()
-	defer stdout.Close()
-	if err != nil {
-		log.Printf("Error opening stdout of command: %v", err)
-		return err
-	}
-	err = cmd.Start()
-	if err != nil {
-		log.Printf("Error starting command: %v", err)
-		return err
-	}
-	_, err = io.Copy(w, stdout)
-	if err != nil {
-		log.Printf("Error copying data to client: %v", err)
-		// Ask the process to exit
-		cmd.Process.Signal(syscall.SIGKILL)
-		cmd.Process.Wait()
-		return err
-	}
-	cmd.Wait()
-	return nil
-}
-
 func ServeJson(status int, data interface{}, w http.ResponseWriter) {
 	js, err := json.Marshal(data)
 	if err != nil {

main.go

@@ -15,13 +15,12 @@ func main() {
 		contentDir = flag.Arg(0)
 	}
 
-	http.Handle("/", http.RedirectHandler("/ui/",302))
+	http.Handle("/", http.RedirectHandler("/ui/", 302))
 	http.Handle("/ui/assets/", http.StripPrefix("/ui/", http.FileServer(http.Dir(uiDirectory))))
 	http.Handle("/ui/", NewDebugHandlerWrapper(http.StripPrefix("/ui/", NewSingleFileServer(indexHtml))))
 	http.Handle("/list/", NewDebugHandlerWrapper(http.StripPrefix("/list/", NewListHandler(contentDir))))
 	http.Handle("/frame/", NewDebugHandlerWrapper(http.StripPrefix("/frame/", NewFrameHandler(contentDir))))
 	http.Handle("/playlist/", NewDebugHandlerWrapper(http.StripPrefix("/playlist/", NewPlaylistHandler(contentDir))))
 	http.Handle("/segments/", NewDebugHandlerWrapper(http.StripPrefix("/segments/", NewStreamHandler(contentDir))))
 	http.ListenAndServe(":8080", nil)
-
 }

ui/assets/css/app.css

@@ -78,7 +78,7 @@ body, html {
       display: flex; }
 
 .player {
-  min-height: 100vh;
+  height: 100vh;
   background: black;
   display: flex;
   flex-direction: column;

ui/assets/gulpfile.js

@@ -13,8 +13,10 @@ gulp.task('sass', function () {
 
 gulp.task('babel', function () {
 	gulp.src('./jsx/*.jsx')
-		.pipe(plumber())	
-		.pipe(babel())
+		.pipe(plumber())
+		.pipe(babel({
+            presets: ["react"]
+        }))
 		.pipe(gulp.dest('./js/'));
 });