From 34e4b9f3d573a8342471bda51ac3eb299a115baa Mon Sep 17 00:00:00 2001 From: Varun Patil Date: Thu, 10 Nov 2022 04:27:29 -0800 Subject: [PATCH] ffprobe --- config.go | 7 ++++ main.go | 29 ++++++++++--- manager.go | 118 ++++++++++++++++++++++++++++++++++++++++++++++------- stream.go | 5 ++- 4 files changed, 138 insertions(+), 21 deletions(-) create mode 100644 config.go diff --git a/config.go b/config.go new file mode 100644 index 00000000..277417b7 --- /dev/null +++ b/config.go @@ -0,0 +1,7 @@ +package main + +type Config struct { + ffmpeg string + ffprobe string + chunkSize float64 +} diff --git a/main.go b/main.go index e7f80cbb..e82356e9 100644 --- a/main.go +++ b/main.go @@ -8,13 +8,18 @@ import ( ) type Handler struct { + c *Config managers map[string]*Manager mutex sync.RWMutex close chan string } -func NewHandler() *Handler { - h := &Handler{managers: make(map[string]*Manager), close: make(chan string)} +func NewHandler(c *Config) *Handler { + h := &Handler{ + c: c, + managers: make(map[string]*Manager), + close: make(chan string), + } go h.watchClose() return h } @@ -36,7 +41,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } streamid := parts[0] - path := "/" + strings.Join(parts[1:len(parts)-2], "/") + path := "/" + strings.Join(parts[1:len(parts)-1], "/") chunk := parts[len(parts)-1] log.Println("Serving", path, streamid, chunk) @@ -50,6 +55,11 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { if manager == nil { manager = h.createManager(path, streamid) } + + if manager == nil { + w.WriteHeader(http.StatusInternalServerError) + return + } manager.ServeHTTP(w, r, chunk) } @@ -62,7 +72,12 @@ func (h *Handler) getManager(streamid string) *Manager { func (h *Handler) createManager(path string, streamid string) *Manager { h.mutex.Lock() defer h.mutex.Unlock() - manager := NewManager(path, streamid, h.close) + manager, err := NewManager(h.c, path, streamid, h.close) + if err != nil { + log.Println("Error creating manager", err) + return nil + } + h.managers[streamid] = manager return manager } @@ -92,7 +107,11 @@ func (h *Handler) Close() { func main() { log.Println("Starting VOD server") - h := NewHandler() + h := NewHandler(&Config{ + ffmpeg: "ffmpeg", + ffprobe: "ffprobe", + chunkSize: 4.0, + }) http.Handle("/", h) http.ListenAndServe(":47788", nil) diff --git a/manager.go b/manager.go index 37e6dd86..a93d5ec9 100644 --- a/manager.go +++ b/manager.go @@ -1,40 +1,65 @@ package main import ( + "bytes" + "context" + "encoding/json" "fmt" + "log" "math" "net/http" + "os/exec" "sort" + "time" ) type Manager struct { + c *Config + path string id string close chan string - duration float64 + probe *ProbeVideoData numChunks int - chunkSize float64 streams map[string]*Stream } -func NewManager(path string, id string, close chan string) *Manager { - m := &Manager{path: path, id: id, close: close} +type ProbeVideoData struct { + Width int + Height int + Duration time.Duration +} + +func NewManager(c *Config, path string, id string, close chan string) (*Manager, error) { + m := &Manager{c: c, path: path, id: id, close: close} m.streams = make(map[string]*Stream) - m.chunkSize = 4 - m.duration = 300 + if err := m.ffprobe(); err != nil { + return nil, err + } - m.numChunks = int(math.Ceil(m.duration / m.chunkSize)) + log.Println("Video duration:", m.probe) - m.streams["360p.m3u8"] = &Stream{m: m, quality: "360p", height: 360, width: 640, bitrate: 945000} - m.streams["480p.m3u8"] = &Stream{m: m, quality: "480p", height: 480, width: 640, bitrate: 1365000} - m.streams["720p.m3u8"] = &Stream{m: m, quality: "720p", height: 720, width: 1280, bitrate: 3045000} - m.streams["1080p.m3u8"] = &Stream{m: m, quality: "1080p", height: 1080, width: 1920, bitrate: 6045000} - m.streams["1440p.m3u8"] = &Stream{m: m, quality: "1440p", height: 1440, width: 2560, bitrate: 9045000} - m.streams["2160p.m3u8"] = &Stream{m: m, quality: "2160p", height: 2160, width: 3840, bitrate: 14045000} - return m + m.numChunks = int(math.Ceil(m.probe.Duration.Seconds() / c.chunkSize)) + + // Possible streams + m.streams["360p.m3u8"] = &Stream{c: c, m: m, quality: "360p", height: 360, width: 640, bitrate: 945000} + m.streams["480p.m3u8"] = &Stream{c: c, m: m, quality: "480p", height: 480, width: 640, bitrate: 1365000} + m.streams["720p.m3u8"] = &Stream{c: c, m: m, quality: "720p", height: 720, width: 1280, bitrate: 3045000} + m.streams["1080p.m3u8"] = &Stream{c: c, m: m, quality: "1080p", height: 1080, width: 1920, bitrate: 6045000} + m.streams["1440p.m3u8"] = &Stream{c: c, m: m, quality: "1440p", height: 1440, width: 2560, bitrate: 9045000} + m.streams["2160p.m3u8"] = &Stream{c: c, m: m, quality: "2160p", height: 2160, width: 3840, bitrate: 14045000} + + // Only keep streams that are smaller than the video + for k, stream := range m.streams { + if stream.height > m.probe.Height || stream.width > m.probe.Width { + delete(m.streams, k) + } + } + + return m, nil } func (m *Manager) ServeHTTP(w http.ResponseWriter, r *http.Request, chunk string) error { @@ -74,3 +99,68 @@ func (m *Manager) ServeIndex(w http.ResponseWriter, r *http.Request) error { func WriteM3U8ContentType(w http.ResponseWriter) { w.Header().Set("Content-Type", "application/x-mpegURL") } + +func (m *Manager) ffprobe() error { + args := []string{ + // Hide debug information + "-v", "error", + + // video + "-show_entries", "format=duration", + "-show_entries", "stream=duration,width,height", + "-select_streams", "v", // Video stream only, we're not interested in audio + + "-of", "json", + m.path, + } + + ctx, _ := context.WithDeadline(context.TODO(), time.Now().Add(5*time.Second)) + cmd := exec.CommandContext(ctx, m.c.ffprobe, args...) + + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + + err := cmd.Run() + if err != nil { + log.Println(stderr.String()) + return err + } + + out := struct { + Streams []struct { + Width int `json:"width"` + Height int `json:"height"` + Duration string `json:"duration"` + } `json:"streams"` + Format struct { + Duration string `json:"duration"` + } `json:"format"` + }{} + + if err := json.Unmarshal(stdout.Bytes(), &out); err != nil { + return err + } + + var duration time.Duration + if out.Streams[0].Duration != "" { + duration, err = time.ParseDuration(out.Streams[0].Duration + "s") + if err != nil { + return err + } + } + if out.Format.Duration != "" { + duration, err = time.ParseDuration(out.Format.Duration + "s") + if err != nil { + return err + } + } + + m.probe = &ProbeVideoData{ + Width: out.Streams[0].Width, + Height: out.Streams[0].Height, + Duration: duration, + } + + return nil +} diff --git a/stream.go b/stream.go index 88d0299d..fda3d4d2 100644 --- a/stream.go +++ b/stream.go @@ -6,6 +6,7 @@ import ( ) type Stream struct { + c *Config m *Manager quality string height int @@ -19,10 +20,10 @@ func (s *Stream) ServeList(w http.ResponseWriter, r *http.Request) error { w.Write([]byte("#EXT-X-VERSION:4\n")) w.Write([]byte("#EXT-X-MEDIA-SEQUENCE:0\n")) w.Write([]byte("#EXT-X-PLAYLIST-TYPE:VOD\n")) - w.Write([]byte(fmt.Sprintf("#EXT-X-TARGETDURATION:%.3f\n", s.m.chunkSize))) + w.Write([]byte(fmt.Sprintf("#EXT-X-TARGETDURATION:%.3f\n", s.c.chunkSize))) for i := 0; i < s.m.numChunks; i++ { - w.Write([]byte(fmt.Sprintf("#EXTINF:%.3f, nodesc\n", s.m.chunkSize))) + w.Write([]byte(fmt.Sprintf("#EXTINF:%.3f, nodesc\n", s.c.chunkSize))) w.Write([]byte(fmt.Sprintf("%s-%06d.ts\n", s.quality, i))) }