Use pointers to store streams

transcoder/src/filestream.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"log"
 	"math"
+	"os"
 	"os/exec"
 	"strconv"
 	"strings"
@@ -12,10 +13,19 @@ import (
 
 type FileStream struct {
 	Path        string
+	Out         string
 	Keyframes   []float64
 	CanTransmux bool
 	Info        *MediaInfo
-	streams     map[Quality]TranscodeStream
+	streams     map[Quality]*TranscodeStream
+}
+
+func GetOutPath() string {
+	out := os.Getenv("CACHE_ROOT")
+	if out == "" {
+		return "/cache"
+	}
+	return out
 }
 
 func NewFileStream(path string) (*FileStream, error) {
@@ -42,16 +52,17 @@ func NewFileStream(path string) (*FileStream, error) {
 
 	return &FileStream{
 		Path:        path,
+		Out:         fmt.Sprintf("%s/%s", GetOutPath(), info.info.Sha),
 		Keyframes:   keyframes,
 		CanTransmux: can_transmux,
 		Info:        info.info,
-		streams:     make(map[Quality]TranscodeStream),
+		streams:     make(map[Quality]*TranscodeStream),
 	}, nil
 }
 
 func GetKeyframes(path string) ([]float64, bool, error) {
-	// run ffprobe to return all IFrames, IFrames are points where we can split the video in segments.
 	log.Printf("Starting ffprobe for keyframes analysis for %s", path)
+	// run ffprobe to return all IFrames, IFrames are points where we can split the video in segments.
 	start := time.Now()
 	out, err := exec.Command(
 		"ffprobe",
@@ -173,19 +184,7 @@ func (fs *FileStream) GetMaster() string {
 	return master
 }
 
-func (fs *FileStream) GetIndex(quality Quality, client string) (string, error) {
-	index := `#EXTM3U
-#EXT-X-VERSION:3
-#EXT-X-PLAYLIST-TYPE:VOD
-#EXT-X-ALLOW-CACHE:YES
-#EXT-X-TARGETDURATION:4
-#EXT-X-MEDIA-SEQUENCE:0
-`
-
-	for segment := 1; segment < len(fs.Keyframes); segment++ {
-		index += fmt.Sprintf("#EXTINF:%.6f\n", fs.Keyframes[segment]-fs.Keyframes[segment-1])
-		index += fmt.Sprintf("segment-%d.ts\n", segment)
-	}
-	index += `#EXT-X-ENDLIST`
-	return index, nil
+func (fs *FileStream) GetVideoIndex(quality Quality, client string) (string, error) {
+	stream := fs.streams[quality]
+	return stream.GetIndex(client)
 }

transcoder/src/transcoder.go

@@ -8,7 +8,7 @@ import (
 
 type Transcoder struct {
 	// All file streams currently running, index is file path
-	streams map[string]FileStream
+	streams map[string]*FileStream
 	// Streams that are staring up
 	preparing map[string]chan *FileStream
 	mutex     sync.RWMutex
@@ -16,7 +16,7 @@ type Transcoder struct {
 
 func NewTranscoder() *Transcoder {
 	return &Transcoder{
-		streams:   make(map[string]FileStream),
+		streams:   make(map[string]*FileStream),
 		preparing: make(map[string]chan *FileStream),
 	}
 }
@@ -28,11 +28,10 @@ func (t *Transcoder) getFileStream(path string) (*FileStream, error) {
 	t.mutex.RUnlock()
 
 	if preparing {
-		pstream := <-channel
-		if pstream == nil {
+		stream = <-channel
+		if stream == nil {
 			return nil, errors.New("could not transcode file. Try again later")
 		}
-		stream = *pstream
 	} else if !ok {
 		t.mutex.Lock()
 		channel = make(chan *FileStream, 1)
@@ -40,7 +39,7 @@ func (t *Transcoder) getFileStream(path string) (*FileStream, error) {
 		t.cleanUnused()
 		t.mutex.Unlock()
 
-		newstream, err := NewFileStream(path)
+		stream, err := NewFileStream(path)
 		log.Printf("Stream created for %s", path)
 		if err != nil {
 			t.mutex.Lock()
@@ -52,14 +51,13 @@ func (t *Transcoder) getFileStream(path string) (*FileStream, error) {
 		}
 
 		t.mutex.Lock()
-		t.streams[path] = *newstream
-		stream = t.streams[path]
+		t.streams[path] = stream
 		delete(t.preparing, path)
 		t.mutex.Unlock()
 
-		channel <- &stream
+		channel <- stream
 	}
-	return &stream, nil
+	return stream, nil
 }
 
 // This method assume the lock is already taken.
@@ -86,7 +84,7 @@ func (t *Transcoder) GetVideoIndex(path string, quality Quality, client string)
 	if err != nil {
 		return "", err
 	}
-	return stream.GetIndex(quality, client)
+	return stream.GetVideoIndex(quality, client)
 }
 
 func (t *Transcoder) GetAudioIndex(path string, audio string, client string) (string, error) {