Tuner v1: HLS pipeline with playback/ingest modes
FFmpeg -> MediaMTX -> HLS relay with runtime-configurable modes: - PLAYBACK_MODE: standard (mpegts), fmp4 (HEVC), llhls - INGEST_MODE: rtmp, rtsp - INGEST_AUDIO_MODE: aac, copy - Hourly playlist auto-refresh from upstream URL - Fullscreen video UI with collapsible sidebar, upstream-down banner - Same-origin HLS proxy for HTTPS deployments Tagged as the last HLS-based version before the mpegts.js (v2) rework.
This commit is contained in:
@@ -5,6 +5,7 @@ import (
|
||||
"log"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"tuner/m3u"
|
||||
"tuner/models"
|
||||
@@ -13,12 +14,34 @@ import (
|
||||
|
||||
// App holds shared application state for all handlers.
|
||||
type App struct {
|
||||
mu sync.RWMutex
|
||||
Channels []models.Channel
|
||||
Status models.StreamStatus
|
||||
PlaylistPath string
|
||||
MediaMTX *process.MediaMTXManager
|
||||
FFmpeg *process.FFmpegManager
|
||||
mu sync.RWMutex
|
||||
Channels []models.Channel
|
||||
Status models.StreamStatus
|
||||
PlaylistPath string
|
||||
PlaylistURL string // if set, reload fetches from this URL before parsing
|
||||
PlaybackMode string // "llhls" or "standard"
|
||||
SwitchStartedAt time.Time
|
||||
CurrentStreamURL string
|
||||
PlaylistVersion int64
|
||||
MediaMTX *process.MediaMTXManager
|
||||
FFmpeg *process.FFmpegManager
|
||||
}
|
||||
|
||||
// HandleConfig returns runtime client configuration as JSON.
|
||||
// GET /api/config
|
||||
func (a *App) HandleConfig(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
mode := a.PlaybackMode
|
||||
if mode == "" {
|
||||
mode = "standard"
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]string{"playback_mode": mode})
|
||||
}
|
||||
|
||||
// HandleChannels lists, searches, and filters channels.
|
||||
@@ -89,6 +112,9 @@ func (a *App) HandleSetSource(w http.ResponseWriter, r *http.Request) {
|
||||
a.mu.Lock()
|
||||
a.Status.ChannelName = ""
|
||||
a.Status.Transitioning = false
|
||||
a.Status.UpstreamDown = false
|
||||
a.SwitchStartedAt = time.Time{}
|
||||
a.CurrentStreamURL = ""
|
||||
a.mu.Unlock()
|
||||
}
|
||||
|
||||
@@ -141,6 +167,9 @@ func (a *App) HandleSetChannel(w http.ResponseWriter, r *http.Request) {
|
||||
a.Status.ChannelName = found.Name
|
||||
a.Status.Live = true
|
||||
a.Status.Transitioning = true
|
||||
a.Status.UpstreamDown = false
|
||||
a.SwitchStartedAt = time.Now()
|
||||
a.CurrentStreamURL = found.StreamURL
|
||||
a.mu.Unlock()
|
||||
|
||||
log.Printf("[admin] channel set to %s (%s)", found.Name, found.ID)
|
||||
@@ -149,7 +178,42 @@ func (a *App) HandleSetChannel(w http.ResponseWriter, r *http.Request) {
|
||||
json.NewEncoder(w).Encode(map[string]string{"status": "ok", "channel": found.Name})
|
||||
}
|
||||
|
||||
// HandleReloadPlaylist re-parses the M3U playlist from disk.
|
||||
// ReloadPlaylist refreshes the playlist from its configured source, reparses it,
|
||||
// increments the playlist version, and remaps the active channel name when the
|
||||
// current stream URL still exists under a new display name.
|
||||
func (a *App) ReloadPlaylist() (int, error) {
|
||||
if a.PlaylistURL != "" {
|
||||
log.Printf("[playlist] re-fetching playlist from %s", a.PlaylistURL)
|
||||
if err := FetchPlaylist(a.PlaylistURL, a.PlaylistPath); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
}
|
||||
|
||||
channels, err := m3u.ParseFile(a.PlaylistPath)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
a.Channels = channels
|
||||
a.PlaylistVersion++
|
||||
a.Status.PlaylistVersion = a.PlaylistVersion
|
||||
|
||||
if a.CurrentStreamURL != "" {
|
||||
for _, ch := range channels {
|
||||
if ch.StreamURL == a.CurrentStreamURL {
|
||||
a.Status.ChannelName = ch.Name
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return len(channels), nil
|
||||
}
|
||||
|
||||
// HandleReloadPlaylist re-fetches (if a URL is configured) and re-parses the M3U playlist.
|
||||
// POST /api/admin/playlist/reload
|
||||
func (a *App) HandleReloadPlaylist(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
@@ -157,20 +221,16 @@ func (a *App) HandleReloadPlaylist(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
channels, err := m3u.ParseFile(a.PlaylistPath)
|
||||
count, err := a.ReloadPlaylist()
|
||||
if err != nil {
|
||||
http.Error(w, "failed to reload playlist: "+err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
a.mu.Lock()
|
||||
a.Channels = channels
|
||||
a.mu.Unlock()
|
||||
|
||||
log.Printf("[admin] playlist reloaded: %d channels", len(channels))
|
||||
log.Printf("[admin] playlist reloaded: %d channels", count)
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]interface{}{"status": "ok", "channels": len(channels)})
|
||||
json.NewEncoder(w).Encode(map[string]interface{}{"status": "ok", "channels": count})
|
||||
}
|
||||
|
||||
// HandleProcessStatus returns the status of managed processes.
|
||||
@@ -189,4 +249,3 @@ func (a *App) HandleProcessStatus(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(status)
|
||||
}
|
||||
|
||||
|
||||
35
backend/api/playlist.go
Normal file
35
backend/api/playlist.go
Normal file
@@ -0,0 +1,35 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
)
|
||||
|
||||
// FetchPlaylist downloads a playlist from url and writes it to destPath.
|
||||
func FetchPlaylist(url, destPath string) error {
|
||||
log.Printf("[playlist] fetching from %s", url)
|
||||
resp, err := http.Get(url) //nolint:gosec
|
||||
if err != nil {
|
||||
return fmt.Errorf("fetch playlist: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("fetch playlist: server returned %s", resp.Status)
|
||||
}
|
||||
if err := os.MkdirAll("/data", 0o755); err != nil {
|
||||
return fmt.Errorf("mkdir /data: %w", err)
|
||||
}
|
||||
f, err := os.Create(destPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create playlist file: %w", err)
|
||||
}
|
||||
defer f.Close()
|
||||
if _, err := io.Copy(f, resp.Body); err != nil {
|
||||
return fmt.Errorf("write playlist file: %w", err)
|
||||
}
|
||||
log.Printf("[playlist] saved to %s", destPath)
|
||||
return nil
|
||||
}
|
||||
@@ -3,8 +3,11 @@ package api
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
const upstreamDownAfter = 10 * time.Second
|
||||
|
||||
// HandleStatus returns the current stream status as JSON.
|
||||
// GET /api/status
|
||||
func (a *App) HandleStatus(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -20,12 +23,26 @@ func (a *App) HandleStatus(w http.ResponseWriter, r *http.Request) {
|
||||
// Attach live stream health metrics when FFmpeg is running
|
||||
status.Health = a.FFmpeg.Health()
|
||||
|
||||
// Auto-clear transitioning once FFmpeg is producing health data
|
||||
// Auto-clear transitioning once FFmpeg is producing health data.
|
||||
// If no progress arrives for a while after a channel switch, flag upstream_down
|
||||
// so all watchers see an "Upstream down?" warning via polling.
|
||||
if status.Transitioning && status.Health != nil && status.Health.Speed != "" {
|
||||
a.mu.Lock()
|
||||
a.Status.Transitioning = false
|
||||
a.Status.UpstreamDown = false
|
||||
a.mu.Unlock()
|
||||
status.Transitioning = false
|
||||
status.UpstreamDown = false
|
||||
} else if status.Transitioning {
|
||||
a.mu.RLock()
|
||||
switchStartedAt := a.SwitchStartedAt
|
||||
a.mu.RUnlock()
|
||||
if !switchStartedAt.IsZero() && time.Since(switchStartedAt) >= upstreamDownAfter {
|
||||
a.mu.Lock()
|
||||
a.Status.UpstreamDown = true
|
||||
a.mu.Unlock()
|
||||
status.UpstreamDown = true
|
||||
}
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
|
||||
158
backend/main.go
158
backend/main.go
@@ -2,10 +2,15 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/http/httputil"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -21,22 +26,144 @@ func getEnv(key, fallback string) string {
|
||||
return fallback
|
||||
}
|
||||
|
||||
// normalizePlaybackMode validates the PLAYBACK_MODE value. Accepts "standard",
|
||||
// "fmp4", or "llhls" (case-insensitive); anything else falls back to "standard".
|
||||
func normalizePlaybackMode(mode string) string {
|
||||
switch strings.ToLower(strings.TrimSpace(mode)) {
|
||||
case "llhls", "lowlatency", "low-latency":
|
||||
return "llhls"
|
||||
case "fmp4", "mp4", "fragmented-mp4", "fragmented_mp4":
|
||||
return "fmp4"
|
||||
default:
|
||||
return "standard"
|
||||
}
|
||||
}
|
||||
|
||||
func normalizeIngestMode(mode string) string {
|
||||
switch strings.ToLower(strings.TrimSpace(mode)) {
|
||||
case "rtsp":
|
||||
return "rtsp"
|
||||
default:
|
||||
return "rtmp"
|
||||
}
|
||||
}
|
||||
|
||||
func normalizeIngestAudioMode(mode string) string {
|
||||
switch strings.ToLower(strings.TrimSpace(mode)) {
|
||||
case "copy":
|
||||
return "copy"
|
||||
default:
|
||||
return "aac"
|
||||
}
|
||||
}
|
||||
|
||||
// mediamtxEnvForMode returns the MTX_* environment variable overrides for the
|
||||
// given playback mode.
|
||||
func mediamtxEnvForMode(mode string) []string {
|
||||
if mode == "llhls" {
|
||||
// Low-Latency HLS: minimal latency, more sensitive to upstream jitter.
|
||||
return []string{
|
||||
"MTX_HLSVARIANT=lowLatency",
|
||||
"MTX_HLSSEGMENTCOUNT=7",
|
||||
"MTX_HLSSEGMENTDURATION=1s",
|
||||
"MTX_HLSPARTDURATION=200ms",
|
||||
}
|
||||
}
|
||||
if mode == "fmp4" {
|
||||
// Fragmented MP4 HLS: smooth standard HLS with HEVC/H.265 support.
|
||||
return []string{
|
||||
"MTX_HLSVARIANT=fmp4",
|
||||
"MTX_HLSSEGMENTCOUNT=7",
|
||||
"MTX_HLSSEGMENTDURATION=2s",
|
||||
}
|
||||
}
|
||||
// Standard HLS (mpegts): smooth, resilient, universally compatible (including iOS Safari).
|
||||
return []string{
|
||||
"MTX_HLSVARIANT=mpegts",
|
||||
"MTX_HLSSEGMENTCOUNT=7",
|
||||
"MTX_HLSSEGMENTDURATION=2s",
|
||||
}
|
||||
}
|
||||
|
||||
func parseRefreshInterval(value string) time.Duration {
|
||||
v := strings.ToLower(strings.TrimSpace(value))
|
||||
if v == "" || v == "0" || v == "off" || v == "false" || v == "disabled" {
|
||||
return 0
|
||||
}
|
||||
d, err := time.ParseDuration(v)
|
||||
if err != nil {
|
||||
log.Printf("[startup] warning: invalid PLAYLIST_AUTO_REFRESH_INTERVAL=%q, using 1h", value)
|
||||
return time.Hour
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// copyFile copies src to dst, creating dst if it doesn't exist.
|
||||
func copyFile(src, dst string) error {
|
||||
in, err := os.Open(src)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open source: %w", err)
|
||||
}
|
||||
defer in.Close()
|
||||
if err := os.MkdirAll("/data", 0o755); err != nil {
|
||||
return fmt.Errorf("mkdir /data: %w", err)
|
||||
}
|
||||
out, err := os.Create(dst)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create dest: %w", err)
|
||||
}
|
||||
defer out.Close()
|
||||
if _, err := io.Copy(out, in); err != nil {
|
||||
return fmt.Errorf("copy: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
playlistPath := getEnv("PLAYLIST_PATH", "/data/playlist.m3u")
|
||||
mediamtxBin := getEnv("MEDIAMTX_PATH", "mediamtx")
|
||||
mediamtxCfg := getEnv("MEDIAMTX_CONFIG", "mediamtx.yml")
|
||||
port := getEnv("RESTREAMER_PORT", "8080")
|
||||
playlistURL := getEnv("PLAYLIST_URL", "")
|
||||
useLocal := getEnv("PLAYLIST_USE_LOCAL", "false") == "true"
|
||||
playbackMode := normalizePlaybackMode(getEnv("PLAYBACK_MODE", "standard"))
|
||||
ingestMode := normalizeIngestMode(getEnv("INGEST_MODE", "rtmp"))
|
||||
ingestAudioMode := normalizeIngestAudioMode(getEnv("INGEST_AUDIO_MODE", "aac"))
|
||||
playlistRefreshInterval := parseRefreshInterval(getEnv("PLAYLIST_AUTO_REFRESH_INTERVAL", "1h"))
|
||||
|
||||
localSrcPath := getEnv("PLAYLIST_LOCAL_SRC", "/data/playlist.m3u.local")
|
||||
|
||||
// Fetch playlist from URL unless PLAYLIST_USE_LOCAL=true
|
||||
if !useLocal && playlistURL != "" {
|
||||
if err := api.FetchPlaylist(playlistURL, playlistPath); err != nil {
|
||||
log.Printf("[startup] warning: could not fetch playlist from URL: %v", err)
|
||||
}
|
||||
} else if useLocal {
|
||||
log.Printf("[startup] PLAYLIST_USE_LOCAL=true, copying %s -> %s", localSrcPath, playlistPath)
|
||||
if err := copyFile(localSrcPath, playlistPath); err != nil {
|
||||
log.Printf("[startup] warning: could not copy local playlist: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Initialize process managers
|
||||
mtxManager := process.NewMediaMTXManager(mediamtxBin, mediamtxCfg)
|
||||
ffmpegManager := process.NewFFmpegManager()
|
||||
mtxManager.SetEnv(mediamtxEnvForMode(playbackMode))
|
||||
ffmpegManager := process.NewFFmpegManager(ingestMode, ingestAudioMode)
|
||||
log.Printf("[startup] playback mode: %s", playbackMode)
|
||||
log.Printf("[startup] ingest mode: %s", ingestMode)
|
||||
log.Printf("[startup] ingest audio mode: %s", ingestAudioMode)
|
||||
|
||||
// Build shared app state
|
||||
app := &api.App{
|
||||
PlaylistPath: playlistPath,
|
||||
PlaybackMode: playbackMode,
|
||||
MediaMTX: mtxManager,
|
||||
FFmpeg: ffmpegManager,
|
||||
}
|
||||
// Only expose the URL for reloads when we're in URL mode
|
||||
if !useLocal && playlistURL != "" {
|
||||
app.PlaylistURL = playlistURL
|
||||
}
|
||||
|
||||
// Load playlist (warn but don't crash if missing)
|
||||
channels, err := m3u.ParseFile(playlistPath)
|
||||
@@ -44,9 +171,29 @@ func main() {
|
||||
log.Printf("[startup] warning: could not load playlist %s: %v", playlistPath, err)
|
||||
} else {
|
||||
app.Channels = channels
|
||||
app.PlaylistVersion = 1
|
||||
app.Status.PlaylistVersion = app.PlaylistVersion
|
||||
log.Printf("[startup] loaded %d channels from %s", len(channels), playlistPath)
|
||||
}
|
||||
|
||||
if playlistRefreshInterval > 0 {
|
||||
log.Printf("[startup] playlist auto-refresh enabled: %s", playlistRefreshInterval)
|
||||
go func() {
|
||||
ticker := time.NewTicker(playlistRefreshInterval)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
count, err := app.ReloadPlaylist()
|
||||
if err != nil {
|
||||
log.Printf("[playlist] auto-refresh failed: %v", err)
|
||||
continue
|
||||
}
|
||||
log.Printf("[playlist] auto-refreshed: %d channels", count)
|
||||
}
|
||||
}()
|
||||
} else {
|
||||
log.Printf("[startup] playlist auto-refresh disabled")
|
||||
}
|
||||
|
||||
// Start MediaMTX (warn but don't crash if binary not found)
|
||||
if err := mtxManager.Start(); err != nil {
|
||||
log.Printf("[startup] warning: could not start mediamtx: %v", err)
|
||||
@@ -56,6 +203,7 @@ func main() {
|
||||
mux := http.NewServeMux()
|
||||
|
||||
// API routes
|
||||
mux.HandleFunc("/api/config", app.HandleConfig)
|
||||
mux.HandleFunc("/api/status", app.HandleStatus)
|
||||
mux.HandleFunc("/api/admin/channels", app.HandleChannels)
|
||||
mux.HandleFunc("/api/admin/groups", app.HandleGroups)
|
||||
@@ -64,6 +212,14 @@ func main() {
|
||||
mux.HandleFunc("/api/admin/playlist/reload", app.HandleReloadPlaylist)
|
||||
mux.HandleFunc("/api/admin/process/status", app.HandleProcessStatus)
|
||||
|
||||
// Proxy HLS through the main app origin so HTTPS deployments avoid mixed content.
|
||||
hlsURL, err := url.Parse("http://127.0.0.1:8888")
|
||||
if err != nil {
|
||||
log.Printf("[startup] warning: invalid HLS proxy URL: %v", err)
|
||||
} else {
|
||||
mux.Handle("/live/", httputil.NewSingleHostReverseProxy(hlsURL))
|
||||
}
|
||||
|
||||
// Serve frontend static files (falls back to index.html for SPA routing)
|
||||
frontendDir := getEnv("FRONTEND_DIR", "frontend/dist")
|
||||
if info, err := os.Stat(frontendDir); err == nil && info.IsDir() {
|
||||
|
||||
@@ -22,11 +22,13 @@ type StreamHealth struct {
|
||||
|
||||
// StreamStatus represents the current streaming state.
|
||||
type StreamStatus struct {
|
||||
Source string `json:"source"` // "obs" or "iptv"
|
||||
ChannelName string `json:"channel_name"` // current IPTV channel name (empty if OBS)
|
||||
Live bool `json:"live"`
|
||||
Transitioning bool `json:"transitioning"` // true while switching to a new channel
|
||||
Health *StreamHealth `json:"health,omitempty"`
|
||||
Source string `json:"source"` // "obs" or "iptv"
|
||||
ChannelName string `json:"channel_name"` // current IPTV channel name (empty if OBS)
|
||||
Live bool `json:"live"`
|
||||
Transitioning bool `json:"transitioning"` // true while switching to a new channel
|
||||
UpstreamDown bool `json:"upstream_down"` // true when no data is detected from the selected upstream
|
||||
PlaylistVersion int64 `json:"playlist_version"` // increments when playlist reloads
|
||||
Health *StreamHealth `json:"health,omitempty"`
|
||||
}
|
||||
|
||||
// ProcessInfo represents the status of a managed child process.
|
||||
|
||||
@@ -15,17 +15,25 @@ import (
|
||||
|
||||
// FFmpegManager manages an FFmpeg child process for IPTV relay.
|
||||
type FFmpegManager struct {
|
||||
mu sync.Mutex
|
||||
cmd *exec.Cmd
|
||||
startTime time.Time
|
||||
lastError string
|
||||
streamURL string
|
||||
health models.StreamHealth
|
||||
mu sync.Mutex
|
||||
cmd *exec.Cmd
|
||||
ingestMode string
|
||||
audioMode string
|
||||
startTime time.Time
|
||||
lastError string
|
||||
streamURL string
|
||||
health models.StreamHealth
|
||||
}
|
||||
|
||||
// NewFFmpegManager creates a new FFmpeg process manager.
|
||||
func NewFFmpegManager() *FFmpegManager {
|
||||
return &FFmpegManager{}
|
||||
func NewFFmpegManager(ingestMode, audioMode string) *FFmpegManager {
|
||||
if ingestMode != "rtsp" {
|
||||
ingestMode = "rtmp"
|
||||
}
|
||||
if audioMode != "copy" {
|
||||
audioMode = "aac"
|
||||
}
|
||||
return &FFmpegManager{ingestMode: ingestMode, audioMode: audioMode}
|
||||
}
|
||||
|
||||
// Start spawns an FFmpeg process to pull from streamURL and push to MediaMTX.
|
||||
@@ -37,17 +45,39 @@ func (f *FFmpegManager) Start(streamURL string) error {
|
||||
// Kill existing process if running
|
||||
f.stopLocked()
|
||||
|
||||
f.cmd = exec.Command("ffmpeg",
|
||||
args := []string{
|
||||
"-loglevel", "warning",
|
||||
"-progress", "pipe:1",
|
||||
"-reconnect", "1",
|
||||
"-reconnect_streamed", "1",
|
||||
"-reconnect_delay_max", "5",
|
||||
"-i", streamURL,
|
||||
"-c", "copy",
|
||||
"-f", "flv",
|
||||
"rtmp://localhost:1935/live/stream",
|
||||
)
|
||||
}
|
||||
if f.ingestMode == "rtsp" {
|
||||
if f.audioMode == "copy" {
|
||||
args = append(args, "-c", "copy")
|
||||
} else {
|
||||
args = append(args,
|
||||
"-c:v", "copy",
|
||||
"-c:a", "aac",
|
||||
"-ar", "48000",
|
||||
"-b:a", "160k",
|
||||
)
|
||||
}
|
||||
args = append(args,
|
||||
"-f", "rtsp",
|
||||
"-rtsp_transport", "tcp",
|
||||
"rtsp://localhost:8554/live/stream",
|
||||
)
|
||||
} else {
|
||||
args = append(args,
|
||||
"-c", "copy",
|
||||
"-f", "flv",
|
||||
"rtmp://localhost:1935/live/stream",
|
||||
)
|
||||
}
|
||||
|
||||
f.cmd = exec.Command("ffmpeg", args...)
|
||||
f.streamURL = streamURL
|
||||
f.lastError = ""
|
||||
f.health = models.StreamHealth{}
|
||||
@@ -68,7 +98,7 @@ func (f *FFmpegManager) Start(streamURL string) error {
|
||||
}
|
||||
|
||||
f.startTime = time.Now()
|
||||
log.Printf("[ffmpeg] started with pid %d, source: %s", f.cmd.Process.Pid, streamURL)
|
||||
log.Printf("[ffmpeg] started with pid %d, ingest=%s, audio=%s, source: %s", f.cmd.Process.Pid, f.ingestMode, f.audioMode, streamURL)
|
||||
|
||||
// Pass cmd reference so goroutines can detect process replacement
|
||||
cmd := f.cmd
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"os/exec"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -18,6 +19,7 @@ type MediaMTXManager struct {
|
||||
cmd *exec.Cmd
|
||||
binaryPath string
|
||||
configPath string
|
||||
extraEnv []string
|
||||
startTime time.Time
|
||||
lastError string
|
||||
stopCh chan struct{}
|
||||
@@ -32,6 +34,14 @@ func NewMediaMTXManager(binaryPath, configPath string) *MediaMTXManager {
|
||||
}
|
||||
}
|
||||
|
||||
// SetEnv sets additional environment variables (e.g. MTX_HLSVARIANT) that are
|
||||
// passed to MediaMTX. Must be called before Start.
|
||||
func (m *MediaMTXManager) SetEnv(env []string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.extraEnv = env
|
||||
}
|
||||
|
||||
// Start launches the MediaMTX process. It auto-restarts on unexpected exit.
|
||||
func (m *MediaMTXManager) Start() error {
|
||||
m.mu.Lock()
|
||||
@@ -52,6 +62,9 @@ func (m *MediaMTXManager) startLocked() error {
|
||||
}
|
||||
|
||||
m.cmd = exec.Command(m.binaryPath, m.configPath)
|
||||
if len(m.extraEnv) > 0 {
|
||||
m.cmd.Env = append(os.Environ(), m.extraEnv...)
|
||||
}
|
||||
m.lastError = ""
|
||||
m.stopped = false
|
||||
m.stopCh = make(chan struct{})
|
||||
|
||||
Reference in New Issue
Block a user