2fa8f2435f
Capture, decrypt, and restream toolkit with compiled-in app modules (RTE, TG4, BBC), on-device MITM proxy, streamd control plane, and www. BBC module.yaml is published (clear streams); other module values stay local.
71 lines
2.1 KiB
Go
71 lines
2.1 KiB
Go
package supervisor
|
|
|
|
import (
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"drmdecryption/apps/streamd/internal/db"
|
|
)
|
|
|
|
// Stub marks streams as starting/stopped without spawning media workers.
|
|
// Replaced by a real NRE/ffmpeg supervisor in Phase 2.
|
|
type Stub struct {
|
|
Store *db.Store
|
|
mu sync.Mutex
|
|
pids map[int64]int
|
|
}
|
|
|
|
func NewStub(store *db.Store) *Stub {
|
|
return &Stub{Store: store, pids: map[int64]int{}}
|
|
}
|
|
|
|
func (s *Stub) Start(id int64) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
st, err := s.Store.GetStream(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if st.MPD == "" || st.Key == "" {
|
|
_, err = s.Store.DB.Exec(`
|
|
INSERT INTO stream_runtime(stream_id,health,play_path,last_error,updated_at)
|
|
VALUES(?,?,?,?,?)
|
|
ON CONFLICT(stream_id) DO UPDATE SET health=excluded.health, last_error=excluded.last_error, updated_at=excluded.updated_at`,
|
|
id, "down", "/hls/"+st.Name+"/index.m3u8", "missing mpd/key - capture required", time.Now().UTC().Format(time.RFC3339))
|
|
return err
|
|
}
|
|
pid := 10000 + int(id)
|
|
s.pids[id] = pid
|
|
_, err = s.Store.DB.Exec(`
|
|
INSERT INTO stream_runtime(stream_id,health,pid,play_path,uptime_s,playlist_age_s,last_error,updated_at)
|
|
VALUES(?,?,?,?,0,1,'',?)
|
|
ON CONFLICT(stream_id) DO UPDATE SET
|
|
health=excluded.health, pid=excluded.pid, play_path=excluded.play_path,
|
|
playlist_age_s=excluded.playlist_age_s, last_error='', updated_at=excluded.updated_at`,
|
|
id, "ok", pid, "/hls/"+st.Name+"/index.m3u8", time.Now().UTC().Format(time.RFC3339))
|
|
return err
|
|
}
|
|
|
|
func (s *Stub) Stop(id int64) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
delete(s.pids, id)
|
|
st, err := s.Store.GetStream(id)
|
|
name := fmt.Sprintf("%d", id)
|
|
if err == nil {
|
|
name = st.Name
|
|
}
|
|
_, err = s.Store.DB.Exec(`
|
|
INSERT INTO stream_runtime(stream_id,health,pid,play_path,uptime_s,bitrate_mbps,playlist_age_s,last_error,updated_at)
|
|
VALUES(?,?,0,?,0,0,0,'',?)
|
|
ON CONFLICT(stream_id) DO UPDATE SET
|
|
health='stopped', pid=0, uptime_s=0, bitrate_mbps=0, playlist_age_s=0, last_error='', updated_at=excluded.updated_at`,
|
|
id, "stopped", "/hls/"+name+"/index.m3u8", time.Now().UTC().Format(time.RFC3339))
|
|
return err
|
|
}
|
|
|
|
func (s *Stub) Restart(id int64) error {
|
|
_ = s.Stop(id)
|
|
return s.Start(id)
|
|
}
|