diff --git a/backend/api/admin.go b/backend/api/admin.go index ab9c4e6..a846d92 100644 --- a/backend/api/admin.go +++ b/backend/api/admin.go @@ -88,6 +88,7 @@ func (a *App) HandleSetSource(w http.ResponseWriter, r *http.Request) { } a.mu.Lock() a.Status.ChannelName = "" + a.Status.Transitioning = false a.mu.Unlock() } @@ -139,6 +140,7 @@ func (a *App) HandleSetChannel(w http.ResponseWriter, r *http.Request) { a.Status.Source = "iptv" a.Status.ChannelName = found.Name a.Status.Live = true + a.Status.Transitioning = true a.mu.Unlock() log.Printf("[admin] channel set to %s (%s)", found.Name, found.ID) @@ -187,3 +189,4 @@ func (a *App) HandleProcessStatus(w http.ResponseWriter, r *http.Request) { 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 768e9dc..f69fbcf 100644 --- a/backend/api/status.go +++ b/backend/api/status.go @@ -17,6 +17,17 @@ 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 status.Transitioning && status.Health != nil && status.Health.Speed != "" { + a.mu.Lock() + a.Status.Transitioning = false + a.mu.Unlock() + status.Transitioning = false + } + w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(status) } diff --git a/backend/m3u/parser.go b/backend/m3u/parser.go index 0dbf320..76a725d 100644 --- a/backend/m3u/parser.go +++ b/backend/m3u/parser.go @@ -105,13 +105,17 @@ func generateID(name, url string) string { } // FilterChannels returns channels matching the given search term and/or group. +// Always returns a non-nil slice so it encodes as [] rather than null in JSON. func FilterChannels(channels []models.Channel, search string, group string) []models.Channel { if search == "" && group == "" { + if channels == nil { + return []models.Channel{} + } return channels } searchLower := strings.ToLower(search) - var result []models.Channel + result := []models.Channel{} for _, ch := range channels { if group != "" && !strings.EqualFold(ch.Group, group) { @@ -127,9 +131,10 @@ func FilterChannels(channels []models.Channel, search string, group string) []mo } // GetGroups returns a deduplicated, sorted list of group names from the channel list. +// Always returns a non-nil slice so it encodes as [] rather than null in JSON. func GetGroups(channels []models.Channel) []string { seen := make(map[string]bool) - var groups []string + groups := []string{} for _, ch := range channels { if ch.Group != "" && !seen[ch.Group] { diff --git a/backend/models/types.go b/backend/models/types.go index cfce374..3b437e5 100644 --- a/backend/models/types.go +++ b/backend/models/types.go @@ -9,11 +9,24 @@ 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"` + 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"` } // ProcessInfo represents the status of a managed child process. diff --git a/backend/process/ffmpeg.go b/backend/process/ffmpeg.go index d560ea0..e8429f9 100644 --- a/backend/process/ffmpeg.go +++ b/backend/process/ffmpeg.go @@ -1,9 +1,12 @@ package process import ( + "bufio" "fmt" + "io" "log" "os/exec" + "strings" "sync" "time" @@ -17,6 +20,7 @@ type FFmpegManager struct { startTime time.Time lastError string streamURL string + health models.StreamHealth } // NewFFmpegManager creates a new FFmpeg process manager. @@ -24,7 +28,7 @@ func NewFFmpegManager() *FFmpegManager { return &FFmpegManager{} } -// Start spawns an FFmpeg process to pull from streamURL and push RTMP to MediaMTX. +// 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() @@ -34,7 +38,11 @@ func (f *FFmpegManager) Start(streamURL string) error { f.stopLocked() f.cmd = exec.Command("ffmpeg", - "-re", + "-loglevel", "warning", + "-progress", "pipe:1", + "-reconnect", "1", + "-reconnect_streamed", "1", + "-reconnect_delay_max", "5", "-i", streamURL, "-c", "copy", "-f", "flv", @@ -42,6 +50,17 @@ func (f *FFmpegManager) Start(streamURL string) error { ) 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() @@ -51,18 +70,59 @@ 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) - // Monitor in background - go f.monitor() + // 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 } -func (f *FFmpegManager) monitor() { - err := f.cmd.Wait() +// 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) @@ -96,6 +156,19 @@ func (f *FFmpegManager) stopLocked() error { 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() diff --git a/backend/process/mediamtx.go b/backend/process/mediamtx.go index ec1e982..4cf2fb2 100644 --- a/backend/process/mediamtx.go +++ b/backend/process/mediamtx.go @@ -1,7 +1,9 @@ package process import ( + "bufio" "fmt" + "io" "log" "os/exec" "sync" @@ -54,6 +56,10 @@ func (m *MediaMTXManager) startLocked() error { 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) @@ -62,12 +68,22 @@ func (m *MediaMTXManager) startLocked() error { 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() diff --git a/docker-compose.yml b/docker-compose.yml index 45936cd..a94cac3 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -6,7 +6,7 @@ services: - "8080:8080" - "8888:8888" volumes: - - ./data/playlist.m3u:/data/playlist.m3u:ro + - ./real1.m3u:/data/playlist.m3u:ro environment: RESTREAMER_PORT: "8080" PLAYLIST_PATH: "/data/playlist.m3u" diff --git a/frontend/src/App.css b/frontend/src/App.css index 1404422..524279a 100644 --- a/frontend/src/App.css +++ b/frontend/src/App.css @@ -95,6 +95,76 @@ body { .status-channel { color: #67e8f9; + display: flex; + align-items: center; + gap: 8px; +} + +.status-channel.channel-transitioning { + animation: channel-pulse 1.2s ease-in-out infinite; +} + +.channel-switching { + font-size: 0.75rem; + font-weight: 600; + color: #fbbf24; + background: #713f12; + padding: 1px 6px; + border-radius: 3px; + animation: switching-blink 1s step-end infinite; +} + +@keyframes channel-pulse { + 0%, 100% { opacity: 1; } + 50% { opacity: 0.5; } +} + +@keyframes switching-blink { + 0%, 100% { opacity: 1; } + 50% { opacity: 0.4; } +} + +.status-health { + display: flex; + align-items: center; + gap: 8px; + font-size: 0.8rem; + font-family: "SF Mono", "Fira Code", monospace; +} + +.health-indicator { + padding: 2px 8px; + border-radius: 3px; + font-size: 0.75rem; + font-weight: 600; +} + +.health-good { + background: #14532d; + color: #4ade80; +} + +.health-warn { + background: #713f12; + color: #fbbf24; +} + +.health-bad { + background: #7f1d1d; + color: #f87171; +} + +.health-unknown { + background: #333; + color: #888; +} + +.health-stats { + color: #888; +} + +.health-drops { + color: #f87171; } /* Controls layout */ diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 2130c98..f341bab 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -32,9 +32,11 @@ export default function App() { useEffect(() => { fetchStatus(); fetchChannels(); - const interval = setInterval(fetchStatus, 5000); + // Poll faster while transitioning so the UI updates promptly + const rate = status?.transitioning ? 1000 : 5000; + const interval = setInterval(fetchStatus, rate); return () => clearInterval(interval); - }, [fetchStatus, fetchChannels]); + }, [fetchStatus, fetchChannels, status?.transitioning]); const handleSourceChanged = () => { fetchStatus(); @@ -50,7 +52,7 @@ export default function App() { return (
- +
diff --git a/frontend/src/components/ChannelList.tsx b/frontend/src/components/ChannelList.tsx index 39fdc85..c13d99d 100644 --- a/frontend/src/components/ChannelList.tsx +++ b/frontend/src/components/ChannelList.tsx @@ -15,14 +15,14 @@ export default function ChannelList({ channels, activeChannel, onRefresh }: Prop const [filtered, setFiltered] = useState(channels); useEffect(() => { - api.getGroups().then(setGroups).catch(() => {}); + api.getGroups().then((g) => setGroups(g ?? [])).catch(() => {}); }, []); useEffect(() => { const fetchFiltered = async () => { try { const result = await api.getChannels(search || undefined, group || undefined); - setFiltered(result); + setFiltered(result ?? []); } catch { setFiltered(channels); } diff --git a/frontend/src/components/Player.tsx b/frontend/src/components/Player.tsx index 018bb2c..a7c2887 100644 --- a/frontend/src/components/Player.tsx +++ b/frontend/src/components/Player.tsx @@ -1,7 +1,11 @@ import { useEffect, useRef } from "react"; import Hls from "hls.js"; -export default function Player() { +interface Props { + live: boolean; +} + +export default function Player({ live }: Props) { const videoRef = useRef(null); const hlsRef = useRef(null); @@ -9,12 +13,28 @@ export default function Player() { const video = videoRef.current; if (!video) return; - const src = `http://${window.location.hostname}:8888/live/stream/`; + // Tear down any existing instance when stream goes offline + if (!live) { + if (hlsRef.current) { + hlsRef.current.destroy(); + hlsRef.current = null; + } + video.removeAttribute("src"); + video.load(); + return; + } + + const src = `http://${window.location.hostname}:8888/live/stream/index.m3u8`; if (Hls.isSupported()) { const hls = new Hls({ enableWorker: true, lowLatencyMode: true, + liveSyncDurationCount: 3, + liveMaxLatencyDurationCount: 6, + liveBackBufferLength: 0, + maxBufferLength: 10, + maxMaxBufferLength: 30, }); hlsRef.current = hls; hls.loadSource(src); @@ -44,7 +64,7 @@ export default function Player() { hlsRef.current = null; } }; - }, []); + }, [live]); return (
diff --git a/frontend/src/components/StatusBar.tsx b/frontend/src/components/StatusBar.tsx index a427f70..bebfd14 100644 --- a/frontend/src/components/StatusBar.tsx +++ b/frontend/src/components/StatusBar.tsx @@ -4,6 +4,22 @@ interface Props { status: StreamStatus | null; } +function healthColor(speed: string): string { + const n = parseFloat(speed); + if (isNaN(n)) return "health-unknown"; + if (n >= 0.95) return "health-good"; + if (n >= 0.8) return "health-warn"; + return "health-bad"; +} + +function healthLabel(speed: string): string { + const n = parseFloat(speed); + if (isNaN(n)) return "?"; + if (n >= 0.95) return "Good"; + if (n >= 0.8) return "Slow upstream"; + return "Upstream too slow"; +} + export default function StatusBar({ status }: Props) { if (!status) { return ( @@ -14,6 +30,8 @@ export default function StatusBar({ status }: Props) { ); } + const h = status.health; + return (
@@ -24,12 +42,32 @@ export default function StatusBar({ status }: Props) { Source: {status.source === "obs" ? "OBS" : "IPTV"} - {status.source === "iptv" && status.channelName && ( + {status.source === "iptv" && status.channel_name && ( <> | - {status.channelName} + + {status.channel_name} + {status.transitioning && ( + Switching... + )} + )} + {h && h.speed && ( + + | + + {healthLabel(h.speed)} + + + {h.speed} / {h.bitrate} + {h.fps && h.fps !== "0.00" && <> / {h.fps}fps} + {h.drop_frames !== "0" && ( + / {h.drop_frames} dropped + )} + + + )}
); } diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 30a76f3..d273709 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -6,10 +6,20 @@ export interface Channel { streamUrl: string; } +export interface StreamHealth { + speed: string; + fps: string; + bitrate: string; + drop_frames: string; + dup_frames: string; +} + export interface StreamStatus { source: "obs" | "iptv"; - channelName: string; + channel_name: string; live: boolean; + transitioning: boolean; + health?: StreamHealth; } export interface ProcessInfo { diff --git a/mediamtx.yml b/mediamtx.yml index 354f87d..33fc8e2 100644 --- a/mediamtx.yml +++ b/mediamtx.yml @@ -6,6 +6,10 @@ logLevel: info logDestinations: [stdout] +# REST API (for diagnostics) +api: yes +apiAddress: :9997 + # RTMP server (ingest from OBS / FFmpeg) rtmp: yes rtmpAddress: :1935 @@ -14,12 +18,15 @@ rtmpAddress: :1935 hls: yes hlsAddress: :8888 hlsAlwaysRemux: yes -hlsSegmentCount: 3 +hlsSegmentCount: 7 hlsSegmentDuration: 1s hlsAllowOrigin: '*' +# RTSP server (used internally for FFmpeg → MediaMTX push, supports HEVC) +rtsp: yes +rtspAddress: :8554 + # Disable unused protocols -rtsp: no webrtc: no srt: no record: no