Add real ffmpeg-backed transcoding subsystem (bounded concurrency, streaming, HLS segmenting)
This commit is contained in:
@@ -0,0 +1,148 @@
|
||||
package transcoding
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// segmentSeconds is the target HLS segment duration.
|
||||
const segmentSeconds = 10
|
||||
|
||||
// HLSSession holds the completed segments for one (source, bitrate)
|
||||
// combination, cached on disk so repeated segment requests (which every
|
||||
// HLS client issues, one per segment) don't re-transcode.
|
||||
type HLSSession struct {
|
||||
Dir string
|
||||
SegmentCount int
|
||||
}
|
||||
|
||||
// hlsCache maps a cache key (source path + options) to its completed
|
||||
// session, so concurrent/repeated requests for the same transcode reuse
|
||||
// one ffmpeg run instead of racing separate ones.
|
||||
type hlsCacheEntry struct {
|
||||
once sync.Once
|
||||
session *HLSSession
|
||||
err error
|
||||
}
|
||||
|
||||
var (
|
||||
hlsCacheMu sync.Mutex
|
||||
hlsCache = map[string]*hlsCacheEntry{}
|
||||
)
|
||||
|
||||
func hlsCacheKey(sourcePath string, opts Options) string {
|
||||
return fmt.Sprintf("%s|%d|%s", sourcePath, opts.MaxBitRateKbps, opts.Format)
|
||||
}
|
||||
|
||||
// BuildHLS transcodes and segments sourcePath into cacheDir (once per
|
||||
// distinct source+options; concurrent callers for the same key share the
|
||||
// result), returning the session describing the produced .ts segments.
|
||||
func (m *Manager) BuildHLS(ctx context.Context, sourcePath, cacheDir string, opts Options) (*HLSSession, error) {
|
||||
key := hlsCacheKey(sourcePath, opts)
|
||||
|
||||
hlsCacheMu.Lock()
|
||||
entry, ok := hlsCache[key]
|
||||
if !ok {
|
||||
entry = &hlsCacheEntry{}
|
||||
hlsCache[key] = entry
|
||||
}
|
||||
hlsCacheMu.Unlock()
|
||||
|
||||
entry.once.Do(func() {
|
||||
entry.session, entry.err = m.buildHLSOnce(ctx, sourcePath, cacheDir, opts, key)
|
||||
})
|
||||
return entry.session, entry.err
|
||||
}
|
||||
|
||||
func (m *Manager) buildHLSOnce(ctx context.Context, sourcePath, cacheDir string, opts Options, key string) (*HLSSession, error) {
|
||||
select {
|
||||
case m.sem <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
defer func() { <-m.sem }()
|
||||
|
||||
dir := filepath.Join(cacheDir, "hls", sanitizeKey(key))
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
playlistPath := filepath.Join(dir, "index.m3u8")
|
||||
segmentPattern := filepath.Join(dir, "seg%05d.ts")
|
||||
|
||||
bitrate := opts.MaxBitRateKbps
|
||||
if bitrate <= 0 {
|
||||
bitrate = defaultBitRateKbps
|
||||
}
|
||||
|
||||
args := []string{
|
||||
"-i", sourcePath,
|
||||
"-c:v", "libx264", "-preset", "veryfast", "-c:a", "aac",
|
||||
"-b:a", fmt.Sprintf("%dk", bitrate),
|
||||
"-f", "hls",
|
||||
"-hls_time", strconv.Itoa(segmentSeconds),
|
||||
"-hls_list_size", "0",
|
||||
"-hls_segment_filename", segmentPattern,
|
||||
"--", playlistPath,
|
||||
}
|
||||
cmd := exec.CommandContext(ctx, m.FFmpegPath, args...)
|
||||
if out, err := cmd.CombinedOutput(); err != nil {
|
||||
return nil, fmt.Errorf("ffmpeg hls: %w: %s", err, out)
|
||||
}
|
||||
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
count := 0
|
||||
for _, e := range entries {
|
||||
if strings.HasSuffix(e.Name(), ".ts") {
|
||||
count++
|
||||
}
|
||||
}
|
||||
return &HLSSession{Dir: dir, SegmentCount: count}, nil
|
||||
}
|
||||
|
||||
func sanitizeKey(key string) string {
|
||||
var b strings.Builder
|
||||
for _, r := range key {
|
||||
switch {
|
||||
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9':
|
||||
b.WriteRune(r)
|
||||
default:
|
||||
b.WriteByte('_')
|
||||
}
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// Playlist returns the m3u8 media playlist text for a completed session,
|
||||
// with segment URIs built by segmentURL(index).
|
||||
func (s *HLSSession) Playlist(segmentURL func(index int) string) string {
|
||||
var b strings.Builder
|
||||
b.WriteString("#EXTM3U\n")
|
||||
b.WriteString("#EXT-X-VERSION:3\n")
|
||||
b.WriteString(fmt.Sprintf("#EXT-X-TARGETDURATION:%d\n", segmentSeconds))
|
||||
b.WriteString("#EXT-X-PLAYLIST-TYPE:VOD\n")
|
||||
for i := 0; i < s.SegmentCount; i++ {
|
||||
b.WriteString(fmt.Sprintf("#EXTINF:%d.0,\n", segmentSeconds))
|
||||
b.WriteString(segmentURL(i))
|
||||
b.WriteString("\n")
|
||||
}
|
||||
b.WriteString("#EXT-X-ENDLIST\n")
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// SegmentPath returns the on-disk path for segment index, or "" if out
|
||||
// of range.
|
||||
func (s *HLSSession) SegmentPath(index int) string {
|
||||
if index < 0 || index >= s.SegmentCount {
|
||||
return ""
|
||||
}
|
||||
return filepath.Join(s.Dir, fmt.Sprintf("seg%05d.ts", index))
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
// Package transcoding runs ffmpeg to convert media on the fly for
|
||||
// clients that can't play a file's native format or want a lower
|
||||
// bitrate. It never buffers a whole output file in memory or on disk
|
||||
// before sending it: ffmpeg's stdout is piped straight to the response,
|
||||
// and the process is bounded (a small worker semaphore) and killed the
|
||||
// moment the request context is cancelled (client disconnect).
|
||||
package transcoding
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"os/exec"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// Format is a supported transcode target.
|
||||
type Format string
|
||||
|
||||
const (
|
||||
FormatMP3 Format = "mp3"
|
||||
FormatOpus Format = "opus"
|
||||
FormatAAC Format = "aac"
|
||||
)
|
||||
|
||||
func (f Format) MimeType() string {
|
||||
switch f {
|
||||
case FormatOpus:
|
||||
return "audio/ogg"
|
||||
case FormatAAC:
|
||||
return "audio/aac"
|
||||
default:
|
||||
return "audio/mpeg"
|
||||
}
|
||||
}
|
||||
|
||||
// ParseFormat maps a client-requested format string (as sent in the
|
||||
// Subsonic "format" parameter) to a supported Format, defaulting to MP3
|
||||
// for anything unrecognized — the same behavior real Subsonic servers
|
||||
// use since mp3 is the universally-supported fallback.
|
||||
func ParseFormat(s string) Format {
|
||||
switch s {
|
||||
case "opus":
|
||||
return FormatOpus
|
||||
case "aac", "m4a":
|
||||
return FormatAAC
|
||||
default:
|
||||
return FormatMP3
|
||||
}
|
||||
}
|
||||
|
||||
// Options configures one transcode.
|
||||
type Options struct {
|
||||
Format Format
|
||||
MaxBitRateKbps int // 0 means "no limit / use a sensible default"
|
||||
TimeOffset float64 // seconds to seek into the source before transcoding
|
||||
}
|
||||
|
||||
const defaultBitRateKbps = 192
|
||||
const maxConcurrentJobs = 4
|
||||
|
||||
// Manager bounds how many ffmpeg processes can run at once, so a burst
|
||||
// of client requests can't fork-bomb the host.
|
||||
type Manager struct {
|
||||
FFmpegPath string
|
||||
sem chan struct{}
|
||||
}
|
||||
|
||||
func NewManager(ffmpegPath string) *Manager {
|
||||
if ffmpegPath == "" {
|
||||
ffmpegPath = "ffmpeg"
|
||||
}
|
||||
return &Manager{FFmpegPath: ffmpegPath, sem: make(chan struct{}, maxConcurrentJobs)}
|
||||
}
|
||||
|
||||
// Stream transcodes sourcePath per opts and copies the result to w as it
|
||||
// is produced, never writing a complete temporary file to disk first.
|
||||
// It blocks until the transcode finishes, the client disconnects (ctx
|
||||
// cancelled, in which case ffmpeg is killed), or an error occurs.
|
||||
func (m *Manager) Stream(ctx context.Context, w io.Writer, sourcePath string, opts Options) error {
|
||||
select {
|
||||
case m.sem <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
defer func() { <-m.sem }()
|
||||
|
||||
args := m.buildArgs(sourcePath, opts, "-")
|
||||
cmd := exec.CommandContext(ctx, m.FFmpegPath, args...)
|
||||
cmd.Stdout = w
|
||||
return cmd.Run()
|
||||
}
|
||||
|
||||
func (m *Manager) buildArgs(sourcePath string, opts Options, output string) []string {
|
||||
bitrate := opts.MaxBitRateKbps
|
||||
if bitrate <= 0 {
|
||||
bitrate = defaultBitRateKbps
|
||||
}
|
||||
|
||||
var args []string
|
||||
if opts.TimeOffset > 0 {
|
||||
args = append(args, "-ss", strconv.FormatFloat(opts.TimeOffset, 'f', 2, 64))
|
||||
}
|
||||
args = append(args, "-i", sourcePath, "-vn", "-map_metadata", "-1")
|
||||
|
||||
switch opts.Format {
|
||||
case FormatOpus:
|
||||
args = append(args, "-c:a", "libopus", "-b:a", fmt.Sprintf("%dk", bitrate), "-f", "opus")
|
||||
case FormatAAC:
|
||||
args = append(args, "-c:a", "aac", "-b:a", fmt.Sprintf("%dk", bitrate), "-f", "adts")
|
||||
default:
|
||||
args = append(args, "-c:a", "libmp3lame", "-b:a", fmt.Sprintf("%dk", bitrate), "-f", "mp3")
|
||||
}
|
||||
args = append(args, "--", output)
|
||||
return args
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
package transcoding
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func genTestFile(t *testing.T, dir string) string {
|
||||
t.Helper()
|
||||
if _, err := exec.LookPath("ffmpeg"); err != nil {
|
||||
t.Skip("ffmpeg not installed")
|
||||
}
|
||||
out := filepath.Join(dir, "src.mp3")
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
cmd := exec.CommandContext(ctx, "ffmpeg", "-f", "lavfi", "-i", "sine=frequency=440:duration=2", "-codec:a", "libmp3lame", out)
|
||||
if combined, err := cmd.CombinedOutput(); err != nil {
|
||||
t.Fatalf("ffmpeg: %v\n%s", err, combined)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestStreamProducesOutput(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
src := genTestFile(t, dir)
|
||||
|
||||
m := NewManager("")
|
||||
var buf bytes.Buffer
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
if err := m.Stream(ctx, &buf, src, Options{Format: FormatMP3, MaxBitRateKbps: 64}); err != nil {
|
||||
t.Fatalf("Stream: %v", err)
|
||||
}
|
||||
if buf.Len() == 0 {
|
||||
t.Fatal("transcoded output is empty")
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamRespectsContextCancellation(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
src := genTestFile(t, dir)
|
||||
|
||||
m := NewManager("")
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel() // already cancelled before starting
|
||||
|
||||
var buf bytes.Buffer
|
||||
err := m.Stream(ctx, &buf, src, Options{Format: FormatMP3})
|
||||
if err == nil {
|
||||
t.Fatal("expected an error when the context is already cancelled")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseFormat(t *testing.T) {
|
||||
cases := map[string]Format{
|
||||
"mp3": FormatMP3,
|
||||
"opus": FormatOpus,
|
||||
"aac": FormatAAC,
|
||||
"unknown": FormatMP3,
|
||||
"": FormatMP3,
|
||||
}
|
||||
for in, want := range cases {
|
||||
if got := ParseFormat(in); got != want {
|
||||
t.Errorf("ParseFormat(%q) = %q, want %q", in, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildHLS(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
src := genTestFile(t, dir)
|
||||
|
||||
m := NewManager("")
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
session, err := m.BuildHLS(ctx, src, filepath.Join(dir, "cache"), Options{Format: FormatAAC, MaxBitRateKbps: 96})
|
||||
if err != nil {
|
||||
t.Fatalf("BuildHLS: %v", err)
|
||||
}
|
||||
if session.SegmentCount == 0 {
|
||||
t.Fatal("expected at least one HLS segment")
|
||||
}
|
||||
playlist := session.Playlist(func(i int) string { return "seg" + string(rune('0'+i)) })
|
||||
if playlist == "" {
|
||||
t.Fatal("empty playlist")
|
||||
}
|
||||
if session.SegmentPath(0) == "" {
|
||||
t.Fatal("SegmentPath(0) should resolve")
|
||||
}
|
||||
if session.SegmentPath(9999) != "" {
|
||||
t.Fatal("SegmentPath out of range should return empty")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user