From 723ae1f1acfc74767e63c7acb3097b942e1ed579 Mon Sep 17 00:00:00 2001 From: Scott Register Date: Fri, 12 Jun 2026 19:33:06 -0700 Subject: [PATCH] v2: mpegts.js + single-upstream fan-out hub (no FFmpeg/HLS) Replaces the FFmpeg -> MediaMTX -> HLS pipeline with a raw MPEG-TS fan-out architecture: - Go stream.Hub holds ONE upstream connection per active channel and broadcasts the raw TS bytes to all connected browsers via /ts. The IPTV provider only ever sees one IP (the server), no matter how many viewers are watching (shared single-channel model). - Frontend uses mpegts.js to play the raw TS in-browser via MSE, fixing HEVC/H.265 4K streams that HLS could not package. - Dropped FFmpeg, MediaMTX, RTMP/RTSP, HLS proxy, playback/ingest modes, the OBS source toggle, and FFmpeg health stats. - UI look preserved (fullscreen video, collapsible sidebar, status bar, upstream-down banner, hourly playlist auto-refresh). Tradeoff: loses iOS Safari (no MSE/mpegts.js); fixes desktop/Android playback of both H.264 and H.265. --- Dockerfile | 18 +- backend/api/admin.go | 85 +-------- backend/api/status.go | 27 +-- backend/api/ts.go | 45 +++++ backend/main.go | 105 +---------- backend/models/types.go | 37 +--- backend/process/ffmpeg.go | 218 ---------------------- backend/process/mediamtx.go | 189 ------------------- backend/stream/hub.go | 242 +++++++++++++++++++++++++ docker-compose.yml | 8 - frontend/bun.lock | 10 +- frontend/package.json | 2 +- frontend/src/App.tsx | 31 +--- frontend/src/api/client.ts | 18 +- frontend/src/components/AdminPanel.tsx | 31 +--- frontend/src/components/Player.tsx | 198 ++++++++++---------- frontend/src/components/StatusBar.tsx | 39 +--- frontend/src/types/index.ts | 28 --- 18 files changed, 432 insertions(+), 899 deletions(-) create mode 100644 backend/api/ts.go delete mode 100644 backend/process/ffmpeg.go delete mode 100644 backend/process/mediamtx.go create mode 100644 backend/stream/hub.go diff --git a/Dockerfile b/Dockerfile index cb70e8f..c9925da 100644 --- a/Dockerfile +++ b/Dockerfile @@ -12,33 +12,19 @@ RUN bun install && bun run build # Stage 3: Runtime FROM alpine:3.20 -RUN apk add --no-cache ffmpeg ca-certificates wget tar \ - && wget -qO /tmp/mediamtx.tar.gz \ - https://github.com/bluenviron/mediamtx/releases/download/v1.12.2/mediamtx_v1.12.2_linux_amd64.tar.gz \ - && tar -xzf /tmp/mediamtx.tar.gz -C /usr/local/bin mediamtx \ - && rm /tmp/mediamtx.tar.gz \ - && apk del wget tar +RUN apk add --no-cache ca-certificates WORKDIR /app COPY --from=backend /out/tuner /app/tuner COPY --from=frontend /src/dist /app/frontend/dist -COPY mediamtx.yml /app/mediamtx.yml ENV RESTREAMER_PORT=8080 ENV PLAYLIST_PATH=/data/playlist.m3u ENV PLAYLIST_URL=https://pia.cx/m3u/630xhAmY/smp424GL ENV PLAYLIST_USE_LOCAL=false ENV PLAYLIST_AUTO_REFRESH_INTERVAL=1h -# Playback mode: "standard" (MPEG-TS, iOS-friendly), "fmp4" (HEVC/H.265), or "llhls" -ENV PLAYBACK_MODE=standard -# Ingest mode: "rtmp" (default) or "rtsp" (better for HEVC/H.265) -ENV INGEST_MODE=rtmp -# RTSP ingest audio mode: "aac" (transcode audio, recommended) or "copy" -ENV INGEST_AUDIO_MODE=aac -ENV MEDIAMTX_PATH=/usr/local/bin/mediamtx -ENV MEDIAMTX_CONFIG=/app/mediamtx.yml ENV FRONTEND_DIR=/app/frontend/dist -EXPOSE 1935 8080 8888 +EXPOSE 8080 ENTRYPOINT ["/app/tuner"] diff --git a/backend/api/admin.go b/backend/api/admin.go index 2c21750..210740c 100644 --- a/backend/api/admin.go +++ b/backend/api/admin.go @@ -9,7 +9,7 @@ import ( "tuner/m3u" "tuner/models" - "tuner/process" + "tuner/stream" ) // App holds shared application state for all handlers. @@ -19,12 +19,10 @@ type App struct { 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 + Hub *stream.Hub } // HandleConfig returns runtime client configuration as JSON. @@ -35,13 +33,8 @@ func (a *App) HandleConfig(w http.ResponseWriter, r *http.Request) { 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}) + json.NewEncoder(w).Encode(map[string]string{"player": "mpegts"}) } // HandleChannels lists, searches, and filters channels. @@ -79,52 +72,7 @@ func (a *App) HandleGroups(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(groups) } -// HandleSetSource sets the active source (obs or iptv). -// POST /api/admin/source {"source": "obs"|"iptv"} -func (a *App) HandleSetSource(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) - return - } - - var req struct { - Source string `json:"source"` - } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - http.Error(w, "invalid json", http.StatusBadRequest) - return - } - - if req.Source != "obs" && req.Source != "iptv" { - http.Error(w, `source must be "obs" or "iptv"`, http.StatusBadRequest) - return - } - - a.mu.Lock() - a.Status.Source = req.Source - a.mu.Unlock() - - // If switching to OBS, stop FFmpeg relay - if req.Source == "obs" { - if err := a.FFmpeg.Stop(); err != nil { - log.Printf("[admin] error stopping ffmpeg: %v", err) - } - a.mu.Lock() - a.Status.ChannelName = "" - a.Status.Transitioning = false - a.Status.UpstreamDown = false - a.SwitchStartedAt = time.Time{} - a.CurrentStreamURL = "" - a.mu.Unlock() - } - - log.Printf("[admin] source set to %s", req.Source) - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]string{"status": "ok", "source": req.Source}) -} - -// HandleSetChannel selects an IPTV channel and starts FFmpeg relay. +// HandleSetChannel selects an IPTV channel and points the hub at its upstream. // POST /api/admin/channel {"channel_id": "..."} func (a *App) HandleSetChannel(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { @@ -156,14 +104,10 @@ func (a *App) HandleSetChannel(w http.ResponseWriter, r *http.Request) { return } - // Start FFmpeg with the channel's stream URL - if err := a.FFmpeg.Start(found.StreamURL); err != nil { - http.Error(w, "failed to start stream: "+err.Error(), http.StatusInternalServerError) - return - } + // Point the shared fan-out hub at this channel's upstream URL. + a.Hub.SetChannel(found.StreamURL) a.mu.Lock() - a.Status.Source = "iptv" a.Status.ChannelName = found.Name a.Status.Live = true a.Status.Transitioning = true @@ -232,20 +176,3 @@ func (a *App) HandleReloadPlaylist(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]interface{}{"status": "ok", "channels": count}) } - -// HandleProcessStatus returns the status of managed processes. -// GET /api/admin/process/status -func (a *App) HandleProcessStatus(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) - return - } - - status := models.SystemStatus{ - MediaMTX: a.MediaMTX.Status(), - FFmpeg: a.FFmpeg.Status(), - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(status) -} diff --git a/backend/api/status.go b/backend/api/status.go index 7c3aab5..fc12f12 100644 --- a/backend/api/status.go +++ b/backend/api/status.go @@ -20,29 +20,22 @@ func (a *App) HandleStatus(w http.ResponseWriter, r *http.Request) { status := a.Status a.mu.RUnlock() - // Attach live stream health metrics when FFmpeg is running - status.Health = a.FFmpeg.Health() - - // 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 != "" { + // Detect upstream health from the hub. Once the hub receives data, the + // channel is considered live; if no data arrives within the threshold + // after a switch, flag upstream_down so all watchers see the warning. + gotData, since := a.Hub.GotData() + if status.Transitioning && gotData { 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 - } + } else if status.Transitioning && !gotData && since >= upstreamDownAfter { + a.mu.Lock() + a.Status.UpstreamDown = true + a.mu.Unlock() + status.UpstreamDown = true } w.Header().Set("Content-Type", "application/json") diff --git a/backend/api/ts.go b/backend/api/ts.go new file mode 100644 index 0000000..4b1da35 --- /dev/null +++ b/backend/api/ts.go @@ -0,0 +1,45 @@ +package api + +import ( + "net/http" +) + +// HandleTS streams the shared channel's raw MPEG-TS bytes to a viewer. +// Every connected browser subscribes to the same upstream via the hub, so the +// server holds only one connection to the IPTV source regardless of how many +// clients are watching. +// GET /ts +func (a *App) HandleTS(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "streaming unsupported", http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "video/mp2t") + w.Header().Set("Cache-Control", "no-cache, no-store") + w.Header().Set("Connection", "keep-alive") + w.WriteHeader(http.StatusOK) + flusher.Flush() + + sub, unsub := a.Hub.Subscribe() + defer unsub() + + ctx := r.Context() + for { + select { + case <-ctx.Done(): + return + case <-sub.Done(): + return + case chunk, ok := <-sub.Chan(): + if !ok { + return + } + if _, err := w.Write(chunk); err != nil { + return + } + flusher.Flush() + } + } +} diff --git a/backend/main.go b/backend/main.go index 648292f..896d40d 100644 --- a/backend/main.go +++ b/backend/main.go @@ -6,8 +6,6 @@ import ( "io" "log" "net/http" - "net/http/httputil" - "net/url" "os" "os/signal" "strings" @@ -16,7 +14,7 @@ import ( "tuner/api" "tuner/m3u" - "tuner/process" + "tuner/stream" ) func getEnv(key, fallback string) string { @@ -26,65 +24,6 @@ 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" { @@ -121,14 +60,9 @@ func copyFile(src, dst string) error { 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") @@ -145,20 +79,13 @@ func main() { } } - // Initialize process managers - mtxManager := process.NewMediaMTXManager(mediamtxBin, mediamtxCfg) - 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) + // Single shared upstream fan-out hub. + hub := stream.NewHub() // Build shared app state app := &api.App{ PlaylistPath: playlistPath, - PlaybackMode: playbackMode, - MediaMTX: mtxManager, - FFmpeg: ffmpegManager, + Hub: hub, } // Only expose the URL for reloads when we're in URL mode if !useLocal && playlistURL != "" { @@ -194,11 +121,6 @@ func main() { 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) - } - // Register routes mux := http.NewServeMux() @@ -207,18 +129,11 @@ func main() { mux.HandleFunc("/api/status", app.HandleStatus) mux.HandleFunc("/api/admin/channels", app.HandleChannels) mux.HandleFunc("/api/admin/groups", app.HandleGroups) - mux.HandleFunc("/api/admin/source", app.HandleSetSource) mux.HandleFunc("/api/admin/channel", app.HandleSetChannel) 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)) - } + // Raw MPEG-TS fan-out endpoint: every viewer streams the shared channel here. + mux.HandleFunc("/ts", app.HandleTS) // Serve frontend static files (falls back to index.html for SPA routing) frontendDir := getEnv("FRONTEND_DIR", "frontend/dist") @@ -257,13 +172,7 @@ func main() { sig := <-sigCh log.Printf("[shutdown] received %v, shutting down...", sig) - // Stop child processes - if err := ffmpegManager.Stop(); err != nil { - log.Printf("[shutdown] error stopping ffmpeg: %v", err) - } - if err := mtxManager.Stop(); err != nil { - log.Printf("[shutdown] error stopping mediamtx: %v", err) - } + hub.Stop() ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() diff --git a/backend/models/types.go b/backend/models/types.go index 8b5ca23..754f37c 100644 --- a/backend/models/types.go +++ b/backend/models/types.go @@ -9,38 +9,11 @@ type Channel struct { StreamURL string `json:"stream_url"` } -// StreamHealth holds real-time FFmpeg progress metrics for diagnosing issues. -// Speed < 1x indicates the upstream IPTV source can't deliver fast enough. -// Drop frames > 0 with speed ~1x suggests local processing issues. -type StreamHealth struct { - Speed string `json:"speed"` // "1.00x" — upstream delivery rate - FPS string `json:"fps"` // frames per second being processed - Bitrate string `json:"bitrate"` // output bitrate - DropFrames string `json:"drop_frames"` // frames dropped by FFmpeg - DupFrames string `json:"dup_frames"` // frames duplicated by FFmpeg -} - // 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 - 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. -type ProcessInfo struct { - Running bool `json:"running"` - PID int `json:"pid"` - Uptime string `json:"uptime"` - Error string `json:"error,omitempty"` -} - -// SystemStatus holds status for all managed processes. -type SystemStatus struct { - MediaMTX ProcessInfo `json:"mediamtx"` - FFmpeg ProcessInfo `json:"ffmpeg"` + ChannelName string `json:"channel_name"` // current IPTV channel name + Live bool `json:"live"` // true when a channel is selected + 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 } diff --git a/backend/process/ffmpeg.go b/backend/process/ffmpeg.go deleted file mode 100644 index 5af8056..0000000 --- a/backend/process/ffmpeg.go +++ /dev/null @@ -1,218 +0,0 @@ -package process - -import ( - "bufio" - "fmt" - "io" - "log" - "os/exec" - "strings" - "sync" - "time" - - "tuner/models" -) - -// FFmpegManager manages an FFmpeg child process for IPTV relay. -type FFmpegManager struct { - 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(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. -// It kills any existing FFmpeg process first. -func (f *FFmpegManager) Start(streamURL string) error { - f.mu.Lock() - defer f.mu.Unlock() - - // Kill existing process if running - f.stopLocked() - - args := []string{ - "-loglevel", "warning", - "-progress", "pipe:1", - "-reconnect", "1", - "-reconnect_streamed", "1", - "-reconnect_delay_max", "5", - "-i", streamURL, - } - 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{} - - // Capture stdout for progress stats, stderr for errors - stdout, err := f.cmd.StdoutPipe() - if err != nil { - return fmt.Errorf("stdout pipe: %w", err) - } - stderr, err := f.cmd.StderrPipe() - if err != nil { - return fmt.Errorf("stderr pipe: %w", err) - } - - if err := f.cmd.Start(); err != nil { - f.lastError = err.Error() - return fmt.Errorf("start ffmpeg: %w", err) - } - - f.startTime = time.Now() - 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 - go f.parseProgress(stdout) - go f.logStderr(stderr) - go f.monitor(cmd) - - return nil -} - -// parseProgress reads FFmpeg -progress output and updates health stats. -func (f *FFmpegManager) parseProgress(r io.Reader) { - scanner := bufio.NewScanner(r) - for scanner.Scan() { - line := scanner.Text() - k, v, ok := strings.Cut(line, "=") - if !ok { - continue - } - f.mu.Lock() - switch k { - case "speed": - f.health.Speed = v - case "fps": - f.health.FPS = v - case "bitrate": - f.health.Bitrate = v - case "drop_frames": - f.health.DropFrames = v - case "dup_frames": - f.health.DupFrames = v - } - f.mu.Unlock() - } -} - -func (f *FFmpegManager) logStderr(r io.Reader) { - scanner := bufio.NewScanner(r) - for scanner.Scan() { - log.Printf("[ffmpeg] %s", scanner.Text()) - } -} - -func (f *FFmpegManager) monitor(cmd *exec.Cmd) { - err := cmd.Wait() - - f.mu.Lock() - defer f.mu.Unlock() - - // If the process was replaced by a new Start() call, don't touch state - if f.cmd != cmd { - return - } - - if err != nil { - f.lastError = err.Error() - log.Printf("[ffmpeg] exited: %v", err) - } else { - log.Println("[ffmpeg] exited (status 0)") - } - - f.cmd = nil -} - -// Stop kills the FFmpeg process. -func (f *FFmpegManager) Stop() error { - f.mu.Lock() - defer f.mu.Unlock() - return f.stopLocked() -} - -func (f *FFmpegManager) stopLocked() error { - if f.cmd == nil || f.cmd.Process == nil { - return nil - } - - if err := f.cmd.Process.Kill(); err != nil { - return fmt.Errorf("kill ffmpeg: %w", err) - } - - // Wait for the process to fully exit to avoid zombies - _ = f.cmd.Wait() - f.cmd = nil - log.Println("[ffmpeg] killed") - return nil -} - -// Health returns the latest stream health stats from FFmpeg progress. -func (f *FFmpegManager) Health() *models.StreamHealth { - f.mu.Lock() - defer f.mu.Unlock() - - if f.cmd == nil || f.health.Speed == "" { - return nil - } - - h := f.health // copy - return &h -} - -// Status returns the current FFmpeg process status. -func (f *FFmpegManager) Status() models.ProcessInfo { - f.mu.Lock() - defer f.mu.Unlock() - - info := models.ProcessInfo{ - Error: f.lastError, - } - - if f.cmd != nil && f.cmd.Process != nil { - info.Running = true - info.PID = f.cmd.Process.Pid - info.Uptime = time.Since(f.startTime).Truncate(time.Second).String() - } - - return info -} diff --git a/backend/process/mediamtx.go b/backend/process/mediamtx.go deleted file mode 100644 index 23fe3eb..0000000 --- a/backend/process/mediamtx.go +++ /dev/null @@ -1,189 +0,0 @@ -package process - -import ( - "bufio" - "fmt" - "io" - "log" - "os" - "os/exec" - "sync" - "time" - - "tuner/models" -) - -// MediaMTXManager manages the MediaMTX child process. -type MediaMTXManager struct { - mu sync.Mutex - cmd *exec.Cmd - binaryPath string - configPath string - extraEnv []string - startTime time.Time - lastError string - stopCh chan struct{} - stopped bool -} - -// NewMediaMTXManager creates a new manager for the MediaMTX process. -func NewMediaMTXManager(binaryPath, configPath string) *MediaMTXManager { - return &MediaMTXManager{ - binaryPath: binaryPath, - configPath: configPath, - } -} - -// 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() - defer m.mu.Unlock() - - if m.cmd != nil && m.cmd.Process != nil { - return fmt.Errorf("mediamtx already running (pid %d)", m.cmd.Process.Pid) - } - - return m.startLocked() -} - -func (m *MediaMTXManager) startLocked() error { - _, err := exec.LookPath(m.binaryPath) - if err != nil { - m.lastError = fmt.Sprintf("binary not found: %s", m.binaryPath) - return fmt.Errorf("mediamtx binary not found: %s", m.binaryPath) - } - - 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{}) - - // Capture stdout/stderr for logging - stdout, _ := m.cmd.StdoutPipe() - stderr, _ := m.cmd.StderrPipe() - - if err := m.cmd.Start(); err != nil { - m.lastError = err.Error() - return fmt.Errorf("start mediamtx: %w", err) - } - - m.startTime = time.Now() - log.Printf("[mediamtx] started with pid %d", m.cmd.Process.Pid) - - go m.logOutput("stdout", stdout) - go m.logOutput("stderr", stderr) - - // Monitor in background and auto-restart on unexpected exit - go m.monitor() - - return nil -} - -func (m *MediaMTXManager) logOutput(name string, r io.Reader) { - scanner := bufio.NewScanner(r) - for scanner.Scan() { - log.Printf("[mediamtx:%s] %s", name, scanner.Text()) - } -} - -func (m *MediaMTXManager) monitor() { - err := m.cmd.Wait() - - m.mu.Lock() - defer m.mu.Unlock() - - // Check if this was a deliberate stop - select { - case <-m.stopCh: - log.Println("[mediamtx] stopped") - return - default: - } - - if err != nil { - m.lastError = err.Error() - log.Printf("[mediamtx] exited unexpectedly: %v — restarting in 2s", err) - } else { - log.Println("[mediamtx] exited unexpectedly (status 0) — restarting in 2s") - } - - m.cmd = nil - - // Auto-restart after a brief delay (without lock held) - go func() { - time.Sleep(2 * time.Second) - m.mu.Lock() - defer m.mu.Unlock() - if m.stopped { - return - } - if m.cmd != nil { - return // already restarted - } - if err := m.startLocked(); err != nil { - log.Printf("[mediamtx] auto-restart failed: %v", err) - } - }() -} - -// Stop kills the MediaMTX process. -func (m *MediaMTXManager) Stop() error { - m.mu.Lock() - defer m.mu.Unlock() - - m.stopped = true - - if m.cmd == nil || m.cmd.Process == nil { - return nil - } - - if m.stopCh != nil { - close(m.stopCh) - } - - if err := m.cmd.Process.Kill(); err != nil { - return fmt.Errorf("kill mediamtx: %w", err) - } - - m.cmd = nil - log.Println("[mediamtx] killed") - return nil -} - -// Restart stops and restarts the MediaMTX process. -func (m *MediaMTXManager) Restart() error { - if err := m.Stop(); err != nil { - return err - } - time.Sleep(500 * time.Millisecond) - return m.Start() -} - -// Status returns the current process status. -func (m *MediaMTXManager) Status() models.ProcessInfo { - m.mu.Lock() - defer m.mu.Unlock() - - info := models.ProcessInfo{ - Error: m.lastError, - } - - if m.cmd != nil && m.cmd.Process != nil { - info.Running = true - info.PID = m.cmd.Process.Pid - info.Uptime = time.Since(m.startTime).Truncate(time.Second).String() - } - - return info -} diff --git a/backend/stream/hub.go b/backend/stream/hub.go new file mode 100644 index 0000000..f9fb322 --- /dev/null +++ b/backend/stream/hub.go @@ -0,0 +1,242 @@ +// Package stream implements a single-upstream, many-client fan-out hub for +// raw MPEG-TS streams. Only one connection is ever opened to the upstream IPTV +// source at a time; all connected browsers receive a copy of the same bytes. +package stream + +import ( + "context" + "io" + "log" + "net/http" + "sync" + "time" +) + +// chunkSize is the read buffer size for pulling from the upstream source. +const chunkSize = 64 * 1024 + +// subscriberBuffer is how many chunks may queue per client before it is +// considered too slow and dropped (prevents one slow client stalling others). +const subscriberBuffer = 256 + +// Subscriber receives copies of upstream TS chunks. +type subscriber struct { + ch chan []byte + closed chan struct{} +} + +// Hub manages a single shared upstream connection and fans its bytes out to +// any number of subscribers. Switching channels tears down the current +// upstream and starts a new one. +type Hub struct { + mu sync.Mutex + + streamURL string + cancel context.CancelFunc + subscribers map[*subscriber]struct{} + running bool + + // gotData is closed once the first upstream bytes arrive after a switch; + // used to detect upstream-down conditions. + gotData bool + startedAt time.Time + + client *http.Client +} + +// NewHub creates an empty hub. +func NewHub() *Hub { + return &Hub{ + subscribers: make(map[*subscriber]struct{}), + client: &http.Client{ + // No overall timeout: this is a long-lived stream. Per-attempt + // dial/response-header timeouts are set on the transport. + Transport: &http.Transport{ + ResponseHeaderTimeout: 15 * time.Second, + IdleConnTimeout: 30 * time.Second, + }, + }, + } +} + +// SetChannel switches the shared upstream to streamURL. If the same URL is +// already active, it is a no-op. An empty URL stops streaming. +func (h *Hub) SetChannel(streamURL string) { + h.mu.Lock() + defer h.mu.Unlock() + + if streamURL == h.streamURL && h.running { + return + } + + // Tear down existing upstream + h.stopLocked() + + h.streamURL = streamURL + if streamURL == "" { + return + } + + ctx, cancel := context.WithCancel(context.Background()) + h.cancel = cancel + h.running = true + h.gotData = false + h.startedAt = time.Now() + + go h.run(ctx, streamURL) + log.Printf("[hub] switched upstream to %s", streamURL) +} + +// Stop closes the upstream connection and clears the active channel. +func (h *Hub) Stop() { + h.mu.Lock() + defer h.mu.Unlock() + h.streamURL = "" + h.stopLocked() +} + +func (h *Hub) stopLocked() { + if h.cancel != nil { + h.cancel() + h.cancel = nil + } + h.running = false +} + +// GotData reports whether upstream bytes have been received since the last +// channel switch, plus how long ago the switch occurred. +func (h *Hub) GotData() (bool, time.Duration) { + h.mu.Lock() + defer h.mu.Unlock() + if !h.running { + return false, 0 + } + return h.gotData, time.Since(h.startedAt) +} + +// Subscribe registers a new client. Returns a subscriber and an unsubscribe +// function the caller must invoke when done. +func (h *Hub) Subscribe() (*subscriber, func()) { + sub := &subscriber{ + ch: make(chan []byte, subscriberBuffer), + closed: make(chan struct{}), + } + + h.mu.Lock() + h.subscribers[sub] = struct{}{} + count := len(h.subscribers) + h.mu.Unlock() + + log.Printf("[hub] client subscribed (%d total)", count) + + var once sync.Once + unsub := func() { + once.Do(func() { + h.mu.Lock() + delete(h.subscribers, sub) + remaining := len(h.subscribers) + h.mu.Unlock() + close(sub.closed) + log.Printf("[hub] client unsubscribed (%d remaining)", remaining) + }) + } + return sub, unsub +} + +// Chan exposes the subscriber's chunk channel. +func (s *subscriber) Chan() <-chan []byte { return s.ch } + +// Closed signals the subscriber has been removed by the hub. +func (s *subscriber) Done() <-chan struct{} { return s.closed } + +// broadcast sends a chunk to all subscribers, dropping any that are too slow. +func (h *Hub) broadcast(chunk []byte) { + h.mu.Lock() + if !h.gotData { + h.gotData = true + } + subs := make([]*subscriber, 0, len(h.subscribers)) + for s := range h.subscribers { + subs = append(subs, s) + } + h.mu.Unlock() + + for _, s := range subs { + select { + case s.ch <- chunk: + default: + // Slow client: drop this chunk for them rather than blocking everyone. + } + } +} + +// run maintains the upstream connection, reconnecting on failure until the +// context is cancelled (channel switch/stop). +func (h *Hub) run(ctx context.Context, streamURL string) { + backoff := time.Second + for { + if ctx.Err() != nil { + return + } + err := h.pull(ctx, streamURL) + if ctx.Err() != nil { + return + } + if err != nil { + log.Printf("[hub] upstream error (%s), retrying in %s: %v", streamURL, backoff, err) + } + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } + if backoff < 10*time.Second { + backoff *= 2 + } + } +} + +// pull opens a single upstream connection and copies bytes to subscribers +// until the stream ends or the context is cancelled. +func (h *Hub) pull(ctx context.Context, streamURL string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, streamURL, nil) + if err != nil { + return err + } + req.Header.Set("User-Agent", "VLC/3.0.20 LibVLC/3.0.20") + + resp, err := h.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return &httpError{status: resp.StatusCode} + } + + buf := make([]byte, chunkSize) + for { + n, err := resp.Body.Read(buf) + if n > 0 { + chunk := make([]byte, n) + copy(chunk, buf[:n]) + h.broadcast(chunk) + } + if err != nil { + if err == io.EOF { + return nil + } + return err + } + if ctx.Err() != nil { + return ctx.Err() + } + } +} + +type httpError struct{ status int } + +func (e *httpError) Error() string { + return "upstream returned HTTP status " + http.StatusText(e.status) +} diff --git a/docker-compose.yml b/docker-compose.yml index c29f19e..52f82f5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -2,9 +2,7 @@ services: tuner: build: . ports: - - "1935:1935" - "8080:8080" - - "8888:8888" volumes: # real1.m3u is mounted read-only as the local source. # It is only used when PLAYLIST_USE_LOCAL=true. @@ -18,10 +16,4 @@ services: PLAYLIST_USE_LOCAL: "false" # Set to "0" or "off" to disable automatic playlist refresh. PLAYLIST_AUTO_REFRESH_INTERVAL: "1h" - # Playback mode: "standard" (MPEG-TS, iOS-friendly), "fmp4" (HEVC/H.265), or "llhls" - PLAYBACK_MODE: "standard" - # Ingest mode: "rtmp" (default) or "rtsp" (better for HEVC/H.265) - INGEST_MODE: "rtmp" - # RTSP ingest audio mode: "aac" (transcode audio, recommended) or "copy" - INGEST_AUDIO_MODE: "aac" restart: unless-stopped diff --git a/frontend/bun.lock b/frontend/bun.lock index 93be81a..060b727 100644 --- a/frontend/bun.lock +++ b/frontend/bun.lock @@ -5,7 +5,7 @@ "": { "name": "frontend", "dependencies": { - "hls.js": "^1.6.15", + "mpegts.js": "^1.8.0", "react": "^19.2.0", "react-dom": "^19.2.0", }, @@ -286,6 +286,8 @@ "electron-to-chromium": ["electron-to-chromium@1.5.286", "", {}, "sha512-9tfDXhJ4RKFNerfjdCcZfufu49vg620741MNs26a9+bhLThdB+plgMeou98CAaHu/WATj2iHOOHTp1hWtABj2A=="], + "es6-promise": ["es6-promise@4.2.8", "", {}, "sha512-HJDGx5daxeIvxdBxvG2cb9g4tEvwIk3i8+nhX0yGrYmZUzbkdg8QbDevheDB8gd0//uPj4c1EQua8Q+MViT0/w=="], + "esbuild": ["esbuild@0.27.3", "", { "optionalDependencies": { "@esbuild/aix-ppc64": "0.27.3", "@esbuild/android-arm": "0.27.3", "@esbuild/android-arm64": "0.27.3", "@esbuild/android-x64": "0.27.3", "@esbuild/darwin-arm64": "0.27.3", "@esbuild/darwin-x64": "0.27.3", "@esbuild/freebsd-arm64": "0.27.3", "@esbuild/freebsd-x64": "0.27.3", "@esbuild/linux-arm": "0.27.3", "@esbuild/linux-arm64": "0.27.3", "@esbuild/linux-ia32": "0.27.3", "@esbuild/linux-loong64": "0.27.3", "@esbuild/linux-mips64el": "0.27.3", "@esbuild/linux-ppc64": "0.27.3", "@esbuild/linux-riscv64": "0.27.3", "@esbuild/linux-s390x": "0.27.3", "@esbuild/linux-x64": "0.27.3", "@esbuild/netbsd-arm64": "0.27.3", "@esbuild/netbsd-x64": "0.27.3", "@esbuild/openbsd-arm64": "0.27.3", "@esbuild/openbsd-x64": "0.27.3", "@esbuild/openharmony-arm64": "0.27.3", "@esbuild/sunos-x64": "0.27.3", "@esbuild/win32-arm64": "0.27.3", "@esbuild/win32-ia32": "0.27.3", "@esbuild/win32-x64": "0.27.3" }, "bin": { "esbuild": "bin/esbuild" } }, "sha512-8VwMnyGCONIs6cWue2IdpHxHnAjzxnw2Zr7MkVxB2vjmQ2ivqGFb4LEG3SMnv0Gb2F/G/2yA8zUaiL1gywDCCg=="], "escalade": ["escalade@3.2.0", "", {}, "sha512-WUj2qlxaQtO4g6Pq5c29GTcWGDyd8itL8zTlipgECz3JesAiiOKotd8JU6otB3PACgG6xkJUyVhboMS+bje/jA=="], @@ -342,8 +344,6 @@ "hermes-parser": ["hermes-parser@0.25.1", "", { "dependencies": { "hermes-estree": "0.25.1" } }, "sha512-6pEjquH3rqaI6cYAXYPcz9MS4rY6R4ngRgrgfDshRptUZIc3lw0MCIJIGDj9++mfySOuPTHB4nrSW99BCvOPIA=="], - "hls.js": ["hls.js@1.6.15", "", {}, "sha512-E3a5VwgXimGHwpRGV+WxRTKeSp2DW5DI5MWv34ulL3t5UNmyJWCQ1KmLEHbYzcfThfXG8amBL+fCYPneGHC4VA=="], - "ignore": ["ignore@5.3.2", "", {}, "sha512-hsBTNUqQTDwkWtcdYI2i06Y/nUBEsNEDJKjWdigLvegy8kDuJAS8uRlpkkcQpyEXL0Z/pjDy5HBmMjRCJ2gq+g=="], "import-fresh": ["import-fresh@3.3.1", "", { "dependencies": { "parent-module": "^1.0.0", "resolve-from": "^4.0.0" } }, "sha512-TR3KfrTZTYLPB6jUjfx6MF9WcWrHL9su5TObK4ZkYgBdWKPOFoSoQIdEuTuR82pmtxH2spWG9h6etwfr1pLBqQ=="], @@ -382,6 +382,8 @@ "minimatch": ["minimatch@3.1.2", "", { "dependencies": { "brace-expansion": "^1.1.7" } }, "sha512-J7p63hRiAjw1NDEww1W7i37+ByIrOWO5XQQAzZ3VOcL0PNybwpfmV/N05zFAzwQ9USyEcX6t3UO+K5aqBQOIHw=="], + "mpegts.js": ["mpegts.js@1.8.0", "", { "dependencies": { "es6-promise": "^4.2.5", "webworkify-webpack": "github:xqq/webworkify-webpack" } }, "sha512-ZtujqtmTjWgcDDkoOnLvrOKUTO/MKgLHM432zGDI8oPaJ0S+ebPxg1nEpDpLw6I7KmV/GZgUIrfbWi3qqEircg=="], + "ms": ["ms@2.1.3", "", {}, "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="], "nanoid": ["nanoid@3.3.11", "", { "bin": { "nanoid": "bin/nanoid.cjs" } }, "sha512-N8SpfPUnUp1bK+PMYW8qSWdl9U+wwNWI4QKxOYDy9JAro3WMX7p2OeVRF9v+347pnakNevPmiHhNmZ2HbFA76w=="], @@ -454,6 +456,8 @@ "vite": ["vite@7.3.1", "", { "dependencies": { "esbuild": "^0.27.0", "fdir": "^6.5.0", "picomatch": "^4.0.3", "postcss": "^8.5.6", "rollup": "^4.43.0", "tinyglobby": "^0.2.15" }, "optionalDependencies": { "fsevents": "~2.3.3" }, "peerDependencies": { "@types/node": "^20.19.0 || >=22.12.0", "jiti": ">=1.21.0", "less": "^4.0.0", "lightningcss": "^1.21.0", "sass": "^1.70.0", "sass-embedded": "^1.70.0", "stylus": ">=0.54.8", "sugarss": "^5.0.0", "terser": "^5.16.0", "tsx": "^4.8.1", "yaml": "^2.4.2" }, "optionalPeers": ["@types/node", "jiti", "less", "lightningcss", "sass", "sass-embedded", "stylus", "sugarss", "terser", "tsx", "yaml"], "bin": { "vite": "bin/vite.js" } }, "sha512-w+N7Hifpc3gRjZ63vYBXA56dvvRlNWRczTdmCBBa+CotUzAPf5b7YMdMR/8CQoeYE5LX3W4wj6RYTgonm1b9DA=="], + "webworkify-webpack": ["webworkify-webpack@github:xqq/webworkify-webpack#24d1e71", {}, "xqq-webworkify-webpack-24d1e71", "sha512-G1LdOhY/Ute4RrohjqCH/KsME92eCS5jE9MuiXO2e6oKg1EkVbHf6YUvFsp2C/EQfLJc88oafJOqdZZVhNkd5A=="], + "which": ["which@2.0.2", "", { "dependencies": { "isexe": "^2.0.0" }, "bin": { "node-which": "./bin/node-which" } }, "sha512-BLI3Tl1TW3Pvl70l3yq3Y64i+awpwXqsGBYWkkqMtnbXgrMD+yj7rhW0kuEDxzJaYXGjEW5ogapKNMEKNMjibA=="], "word-wrap": ["word-wrap@1.2.5", "", {}, "sha512-BN22B5eaMMI9UMtjrGd5g5eCYPpCPDUy0FJXbYsaT5zYxjFOckS53SQDE3pWkVoWpHXVb3BrYcEN4Twa55B5cA=="], diff --git a/frontend/package.json b/frontend/package.json index 13f84c5..63a4057 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -10,7 +10,7 @@ "preview": "vite preview" }, "dependencies": { - "hls.js": "^1.6.15", + "mpegts.js": "^1.8.0", "react": "^19.2.0", "react-dom": "^19.2.0" }, diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 2de0db4..c5a01a3 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1,5 +1,5 @@ import { useState, useEffect, useCallback, useRef } from "react"; -import type { StreamStatus, Channel, PlaybackMode } from "./types"; +import type { StreamStatus, Channel } from "./types"; import * as api from "./api/client"; import Player from "./components/Player"; import StatusBar from "./components/StatusBar"; @@ -17,23 +17,9 @@ export default function App() { const [status, setStatus] = useState(null); const [channels, setChannels] = useState([]); const [sidebarOpen, setSidebarOpen] = useState(false); - const [playbackMode, setPlaybackMode] = useState("standard"); const [toggleVisible, setToggleVisible] = useState(false); const hideTimer = useRef | null>(null); - useEffect(() => { - api - .getConfig() - .then((c) => { - if (c.playback_mode === "llhls" || c.playback_mode === "fmp4") { - setPlaybackMode(c.playback_mode); - } else { - setPlaybackMode("standard"); - } - }) - .catch(() => {}); - }, []); - const fetchStatus = useCallback(async () => { try { const s = await api.getStatus(); @@ -56,7 +42,7 @@ export default function App() { fetchStatus(); fetchChannels(); // Poll faster while live so passive viewers detect channel switches promptly. - // Transitioning gets the fastest rate so the UI clears quickly after FFmpeg starts. + // Transitioning gets the fastest rate so the UI clears quickly after a switch. const rate = status?.transitioning ? 1000 : status?.live ? 2000 : 5000; const interval = setInterval(fetchStatus, rate); return () => clearInterval(interval); @@ -103,10 +89,6 @@ export default function App() { } }, [showToggle]); - const handleSourceChanged = () => { - fetchStatus(); - }; - const handlePlaylistReloaded = () => { fetchChannels(); }; @@ -121,8 +103,7 @@ export default function App() {
@@ -138,11 +119,7 @@ export default function App() {