commit 61ffe36e6efb13fc404698f0769f345d2807a2a6
parent ff9281350d5e4901cc7da5aad607b45bb84e58f9
Author: Oliver Lowe <o@olowe.co>
Date: Tue, 4 Jun 2024 15:33:57 +1000
cmd/hlsserve: import our poor man's livestream server
More documentation to come
Diffstat:
2 files changed, 261 insertions(+), 0 deletions(-)
diff --git a/m3u8/cmd/hlsproxy/hlsproxy.go b/m3u8/cmd/hlsproxy/hlsproxy.go
@@ -0,0 +1,91 @@
+package main
+
+import (
+ "fmt"
+ "log"
+ "net/http"
+ "net/url"
+ "os"
+ "path"
+ "time"
+
+ "github.com/untangledco/streaming/m3u8"
+)
+
+const usage string = "usage: hlsproxy url"
+
+func init() {
+ log.SetFlags(0)
+ log.SetPrefix("hlssrv: ")
+}
+
+func servePlaylist(p *m3u8.Playlist) http.HandlerFunc {
+ return func(w http.ResponseWriter, req *http.Request) {
+ if req.Method != http.MethodGet {
+ http.Error(w, "nope", http.StatusMethodNotAllowed)
+ }
+ log.Println(req.Method, req.URL)
+ w.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
+ if err := m3u8.Encode(w, p); err != nil {
+ log.Println("encode playlist:", err)
+ }
+ }
+}
+
+func injectBreak(plist *m3u8.Playlist, dur time.Duration, schedule int) error {
+ // ticks := (dur * time.Second) / 90000
+ var money []m3u8.Segment
+ for i, seg := range plist.Segments {
+ if i%schedule != 0 || i == 0 {
+ money = append(money, seg)
+ continue
+ }
+ adbreak := m3u8.Segment{
+ URI: "https://test-streams.mux.dev/test_001/stream_1000k_48k_640x360_000.ts",
+ Duration: dur,
+ Discontinuity: true,
+ }
+ seg.Discontinuity = true
+ money = append(money, adbreak, seg)
+ }
+ plist.Segments = money
+ return nil
+}
+
+func main() {
+ if len(os.Args) != 2 {
+ fmt.Fprintln(os.Stderr, usage)
+ os.Exit(127)
+ }
+ link, err := url.Parse(os.Args[1])
+ if err != nil {
+ log.Fatal(err)
+ }
+ resp, err := http.Get(link.String())
+ if err != nil {
+ log.Fatal("get playlist:", err)
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ log.Fatal("get playlist: remote status:", resp.Status)
+ }
+ source, err := m3u8.ParsePlaylist(resp.Body)
+ if err != nil {
+ log.Fatal("parse playlist:", err)
+ }
+ link.Path = path.Dir(link.Path) + "/"
+ for i := range source.Media {
+ source.Media[i].URI = link.String() + source.Media[i].URI
+ }
+ for i := range source.Variants {
+ source.Variants[i].URI = link.String() + source.Variants[i].URI
+ }
+ for i := range source.Segments {
+ source.Segments[i].URI = link.String() + source.Segments[i].URI
+ }
+ if err := injectBreak(source, 10*time.Second, 8); err != nil {
+ log.Fatalf("inject ad break: %v", err)
+ }
+ http.HandleFunc("/", servePlaylist(source))
+ log.Fatal(http.ListenAndServe(":8080", nil))
+}
diff --git a/m3u8/cmd/hlsserve/hlsserve.go b/m3u8/cmd/hlsserve/hlsserve.go
@@ -0,0 +1,170 @@
+package main
+
+import (
+ "bytes"
+ "fmt"
+ "io"
+ "log"
+ "net"
+ "net/http"
+ "os"
+ "path"
+ "path/filepath"
+ "strings"
+ "time"
+
+ "github.com/untangledco/streaming/m3u8"
+)
+
+func init() {
+ log.SetFlags(0)
+ log.SetPrefix("hlsserve: ")
+ var err error
+ cacheDir, err = os.UserCacheDir()
+ if err != nil {
+ log.Fatalln("user cache dir:", err)
+ }
+ cacheDir = filepath.Join(cacheDir, "hlsserve")
+}
+
+const tsSize int = 188
+
+// rule of thumb for UDP transport
+const maxTSBytes int = 7 * tsSize
+
+const segmentDuration = 4 * time.Second
+
+var cacheDir string
+var sequence int
+
+func removeOld(dir string, maxAge time.Duration) error {
+ ents, err := os.ReadDir(dir)
+ if err != nil {
+ return err
+ }
+ for _, dent := range ents {
+ if !strings.HasSuffix(dent.Name(), ".ts") {
+ continue
+ }
+ stat, err := dent.Info()
+ if err != nil {
+ return err
+ }
+ if time.Since(stat.ModTime()) > maxAge {
+ if err := os.Remove(filepath.Join(dir, dent.Name())); err != nil {
+ return err
+ }
+ sequence++
+ }
+ }
+ return nil
+}
+
+func dumpMedia(dir string, r io.Reader, ch <-chan time.Time) error {
+ var segment int
+ segments := &bytes.Buffer{}
+ for {
+ select {
+ case <-ch:
+ s := fmt.Sprintf("%04d.ts", segment)
+ fname := path.Join(dir, s)
+ if err := os.WriteFile(fname, segments.Bytes(), 0644); err != nil {
+ return err
+ }
+ segments.Reset()
+ segment++
+ default:
+ _, err := io.CopyN(segments, r, int64(maxTSBytes))
+ if err != nil {
+ return err
+ }
+ }
+ }
+ fmt.Fprintln(os.Stderr, "shouldn't really be returning here...?")
+ return nil
+}
+
+func makePlaylist(dir string) (*m3u8.Playlist, error) {
+ names, err := filepath.Glob(filepath.Join(dir, "*.ts"))
+ if err != nil {
+ return nil, fmt.Errorf("find segments: %w", err)
+ }
+ playlist := &m3u8.Playlist{
+ Version: 7,
+ TargetDuration: segmentDuration,
+ Sequence: sequence,
+ }
+ for _, name := range names {
+ seg := m3u8.Segment{URI: path.Base(name), Duration: playlist.TargetDuration}
+ playlist.Segments = append(playlist.Segments, seg)
+ }
+ return playlist, nil
+}
+
+const usage string = "usage: hlsserve dir"
+
+func servePlaylist(dir string) http.HandlerFunc {
+ return func(w http.ResponseWriter, req *http.Request) {
+ playlist, err := makePlaylist(dir)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ w.Header().Set("Content-Type", m3u8.MimeType)
+ if err := m3u8.Encode(w, playlist); err != nil {
+ log.Printf("encode playlist: %v", err)
+ }
+ }
+}
+
+func setCache(seconds int, next http.Handler) http.HandlerFunc {
+ return func(w http.ResponseWriter, req *http.Request) {
+ w.Header().Set("Cache-Control", fmt.Sprintf("max-age=%d", seconds))
+ next.ServeHTTP(w, req)
+ }
+}
+
+func main() {
+ if len(os.Args) > 2 {
+ fmt.Fprintln(os.Stderr, usage)
+ os.Exit(2)
+ } else if len(os.Args) == 2 {
+ cacheDir = os.Args[1]
+ }
+ if err := os.MkdirAll(cacheDir, 0755); err != nil {
+ log.Fatal(err)
+ }
+
+ ln, err := net.Listen("tcp", ":9000")
+ if err != nil {
+ log.Fatal(err)
+ }
+ conn, err := ln.Accept()
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ go func() {
+ ticker := time.NewTicker(segmentDuration)
+ if err := dumpMedia(cacheDir, conn, ticker.C); err != nil {
+ log.Fatalln("dump media:", err)
+ }
+ }()
+
+ go func() {
+ ticker := time.NewTicker(segmentDuration)
+ for {
+ select {
+ case <-ticker.C:
+ if err := removeOld(cacheDir, 3*8*segmentDuration); err != nil {
+ log.Println("remove old segments:", err)
+ }
+ }
+ }
+ }()
+
+ http.Handle("/playlist.m3u8", servePlaylist(cacheDir))
+ fsys := http.FileServer(http.FS(os.DirFS(cacheDir)))
+ http.Handle("/", setCache(60, fsys))
+ log.Fatal(http.ListenAndServe(":8000", nil))
+}
