A stream unfetched and untouched for two minutes is stopped, freeing its tuner when an app quits or a TV is switched off without saying so. A player holding one paused calls POST /api/heartbeat (Heartbeat in the desktop app and web.js), which keeps it and says whether it's still there.
- commit
- 05a14fcd3a6fe9ee28890acb5ad176b1d7a14a34
- parent
- 83eee8152f368132ee417a80bcbf5c20cd1c7513
- author
- FelineStateMachine <226808055+FelineStateMachine@users.noreply.github.com>
- kind
- feat risk score 1.9 of 3, Jev's estimate
| M | app.go | 10 | ++++ |
| M | frontend/ | 2 | + |
| M | frontend/ | 2 | + |
| M | frontend/ | 4 | ++ |
| M | internal/ | 18 | ++++++++ |
| M | internal/ | 9 | ++++ |
| M | internal/ | 7 | +++ |
| M | internal/ | 10 | ++++ |
| M | internal/ | 68 | ++++++++++++++++++++++++++++- |
| A | internal/ | 71 | ++++++++++++++++++++++++++++++ |
@@ -256,6 +256,16 @@ func (a *App) StopTV() {
256256 }257257 }258258 259+// Heartbeat keeps this app's stream from being stopped as idle while it260+// holds it (paused), and reports whether the server still has it.261+func (a *App) Heartbeat() (bool, error) {262+ c, s, err := a.server()263+ if err != nil {264+ return false, err265+ }266+ return c.Heartbeat(a.ctx, s.ClientID)267+}268+259269 // DVR returns recording rules, schedule and library.260270 func (a *App) DVR() (*dvr.State, error) {261271 c, _, err := a.server()@@ -247,6 +247,8 @@
247247 async StopTV() {248248 try { await call('POST', '/api/stop', { client: load().clientId }); } catch { /* as the desktop app: ignored */ }249249 },250+ // Keeps a paused stream from being stopped as idle; false once it's gone.251+ Heartbeat: () => call('POST', '/api/heartbeat', { client: load().clientId }).then((r) => !!(r && r.stream)),250252 // How long a channel change took, for the server's admin page.251253 ReportTune: (m) => call('POST', '/api/metrics', m).then(() => undefined),252254 DVR: () => call('GET', '/api/dvr'),@@ -20,6 +20,8 @@ export function DeleteRule(arg1:string):Promise<void>;
2020 2121 export function ExtendRecording(arg1:string,arg2:number):Promise<time.Time>;2222 23+export function Heartbeat():Promise<boolean>;24+2325 export function Log(arg1:string,arg2:string):Promise<void>;2426 2527 export function MarkWatched(arg1:string,arg2:boolean):Promise<void>;@@ -26,6 +26,10 @@ export function ExtendRecording(arg1, arg2) {
2626 return window['go']['main']['App']['ExtendRecording'](arg1, arg2);2727 }2828 29+export function Heartbeat() {30+ return window['go']['main']['App']['Heartbeat']();31+}32+2933 export function Log(arg1, arg2) {3034 return window['go']['main']['App']['Log'](arg1, arg2);3135 }@@ -40,6 +40,9 @@ type Backend interface {
4040 Weather(ctx context.Context) (*weather.Report, error)4141 Tune(ctx context.Context, client, number string) (*stream.Playback, error)4242 Stop(ctx context.Context, client string) error43+ // Heartbeat keeps client's stream from being stopped as idle, and44+ // reports whether it still has one.45+ Heartbeat(ctx context.Context, client string) (bool, error)4346 DVR(ctx context.Context) (*dvr.State, error)4447 Record(ctx context.Context, req dvr.Request) (*dvr.Rule, error)4548 DeleteRule(ctx context.Context, id string) error@@ -112,6 +115,11 @@ type progressReq struct {
112115 Watched *bool `json:"watched,omitempty"`113116 }114117 118+// heartbeatReply says whether the client still has a stream.119+type heartbeatReply struct {120+ Stream bool `json:"stream"`121+}122+115123 // extendReq extends a recording by Minutes; the reply says Until when.116124 type extendReq struct {117125 Minutes int `json:"minutes,omitempty"`@@ -164,6 +172,16 @@ func (s *Server) Handler() http.Handler {
164172 }165173 reply(w)(struct{}{}, b.Stop(r.Context(), req.Client))166174 })175+ // A player holding a stream without fetching from it (paused) says176+ // so every half minute or so; a stream idle two minutes is stopped.177+ mux.HandleFunc("POST /api/heartbeat", func(w http.ResponseWriter, r *http.Request) {178+ var req clientReq179+ if !decode(w, r, &req) {180+ return181+ }182+ on, err := b.Heartbeat(r.Context(), req.Client)183+ reply(w)(heartbeatReply{Stream: on}, err)184+ })167185 // The app reports how long its channel changes took, from the key168186 // press to the first frame playing, for the admin page.169187 mux.HandleFunc("POST /api/metrics", func(w http.ResponseWriter, r *http.Request) {@@ -78,6 +78,9 @@ func (f *fake) Tune(_ context.Context, client, number string) (*stream.Playback,
7878 return &stream.Playback{ID: "abc", Path: "/live/abc/index.m3u8"}, nil7979 }8080 func (f *fake) Stop(_ context.Context, client string) error { f.stopped = client; return nil }81+func (f *fake) Heartbeat(_ context.Context, client string) (bool, error) {82+ return client == "mac" && f.stopped == "", nil83+}8184 func (f *fake) DVR(context.Context) (*dvr.State, error) { return &dvr.State{Available: true}, nil }8285 func (f *fake) Record(_ context.Context, r dvr.Request) (*dvr.Rule, error) {8386 return &dvr.Rule{ID: "r1", Kind: r.Kind, Channel: r.Channel}, nil@@ -140,9 +143,15 @@ func TestClientServerRoundTrip(t *testing.T) {
140143 if err := c.DeleteRecording(ctx, "e/1"); err != nil || be.deleted != "e/1" {141144 t.Errorf("delete: %v, backend saw %q", err, be.deleted)142145 }146+ if on, err := c.Heartbeat(ctx, "mac"); err != nil || !on {147+ t.Errorf("heartbeat: %v, %v", on, err)148+ }143149 if err := c.Stop(ctx, "mac"); err != nil || be.stopped != "mac" {144150 t.Errorf("stop: %v, backend saw %q", err, be.stopped)145151 }152+ if on, err := c.Heartbeat(ctx, "mac"); err != nil || on {153+ t.Errorf("heartbeat after stop: %v, %v", on, err)154+ }146155 if pb, err := c.PlayRecording(ctx, "mac", "rec1", 754); err != nil || pb.Offset != 754 {147156 t.Errorf("play recording: %+v, %v", pb, err)148157 }@@ -105,6 +105,13 @@ func (c *Client) Stop(ctx context.Context, client string) error {
105105 return c.call(ctx, http.MethodPost, "/api/stop", clientReq{Client: client}, nil)106106 }107107 108+// Heartbeat implements Backend.109+func (c *Client) Heartbeat(ctx context.Context, client string) (bool, error) {110+ var out heartbeatReply111+ err := c.call(ctx, http.MethodPost, "/api/heartbeat", clientReq{Client: client}, &out)112+ return out.Stream, err113+}114+108115 // DVR implements Backend.109116 func (c *Client) DVR(ctx context.Context) (*dvr.State, error) {110117 var s dvr.State@@ -832,6 +832,16 @@ func (s *Service) Stop(_ context.Context, client string) error {
832832 return nil833833 }834834 835+// Heartbeat keeps client's stream (live TV or a recording) from being836+// stopped as idle while its player holds it without fetching (paused),837+// and reports whether client still has one.838+func (s *Service) Heartbeat(_ context.Context, client string) (bool, error) {839+ if s.streams == nil {840+ return false, nil841+ }842+ return s.streams.Touch(client), nil843+}844+835845 // DVR returns rules, schedule and recordings.836846 func (s *Service) DVR(ctx context.Context) (*dvr.State, error) {837847 if s.dvr == nil {@@ -22,6 +22,7 @@ import (
2222 "strconv"2323 "strings"2424 "sync"25+ "sync/atomic"2526 "time"2627 2728 "airwaves/internal/phase"@@ -31,6 +32,10 @@ import (
3132 // maxSessions bounds concurrent transcodes; the oldest is evicted.3233 const maxSessions = 43334 35+// IdleAfter is how long a stream lasts without a request for its files or36+// a Touch: a player gone (an app quit, a TV switched off) frees its tuner.37+const IdleAfter = 2 * time.Minute38+3439 // Playback locates a running stream.3540 type Playback struct {3641 ID string `json:"id"`@@ -57,6 +62,8 @@ type Server struct {
5762 5863 mu sync.Mutex5964 sessions map[string]*session // by client65+ idle time.Duration // IdleAfter, but in tests66+ quit chan struct{} // closed by Close, ending the reaper6067 }6168 6269 type session struct {@@ -64,11 +71,18 @@ type session struct {
6471 client string6572 dir string6673 started time.Time74+ used atomic.Int64 // when last used, in Unix nanoseconds6775 cancel context.CancelFunc6876 done chan struct{}6977 stderr *tail7078 }7179 80+func (sess *session) touch(t time.Time) { sess.used.Store(t.UnixNano()) }81+82+func (sess *session) idleFor(now time.Time) time.Duration {83+ return now.Sub(time.Unix(0, sess.used.Load()))84+}85+7286 // FindFFmpeg locates the ffmpeg binary.7387 func FindFFmpeg() (string, error) {7488 if bin, err := exec.LookPath("ffmpeg"); err == nil {@@ -94,7 +108,8 @@ func NewServer(loopback bool) (*Server, error) {
94108 if err != nil {95109 return nil, err96110 }97- s := &Server{ffmpeg: bin, root: root, sessions: make(map[string]*session)}111+ s := &Server{ffmpeg: bin, root: root, sessions: make(map[string]*session), idle: IdleAfter, quit: make(chan struct{})}112+ go s.reap()98113 if loopback {99114 ln, err := net.Listen("tcp", "127.0.0.1:0")100115 if err != nil {@@ -148,6 +163,7 @@ func (s *Server) Start(ctx context.Context, client string, in tuner.Input) (*Pla
148163 id: id, client: client, dir: dir, started: time.Now(),149164 cancel: cancel, done: make(chan struct{}), stderr: &tail{max: 4096},150165 }166+ sess.touch(sess.started)151167 cmd.Stderr = sess.stderr152168 var source io.WriteCloser153169 if in.Source != nil {@@ -461,6 +477,50 @@ func TrackName(a tuner.AudioTrack, i int) (name, lang string) {
461477 // Stop ends client's stream and removes its files.462478 func (s *Server) Stop(client string) { s.stop(client) }463479 480+// Touch marks client's stream as in use, for a player holding one without481+// fetching from it (paused, say), and reports whether client has one.482+func (s *Server) Touch(client string) bool {483+ s.mu.Lock()484+ defer s.mu.Unlock()485+ sess := s.sessions[client]486+ if sess != nil {487+ sess.touch(time.Now())488+ }489+ return sess != nil490+}491+492+// reap stops streams left idle, until Close.493+func (s *Server) reap() {494+ for {495+ s.mu.Lock()496+ every := s.idle / 4497+ s.mu.Unlock()498+ select {499+ case <-s.quit:500+ return501+ case now := <-time.After(every):502+ s.reapIdle(now)503+ }504+ }505+}506+507+// reapIdle stops the streams idle longer than the server allows at now.508+func (s *Server) reapIdle(now time.Time) {509+ var idle []*session510+ s.mu.Lock()511+ for client, sess := range s.sessions {512+ if sess.idleFor(now) > s.idle {513+ delete(s.sessions, client)514+ idle = append(idle, sess)515+ }516+ }517+ s.mu.Unlock()518+ for _, sess := range idle {519+ log.Printf("stream: %s idle, stopped", sess.client)520+ sess.stop()521+ }522+}523+464524 // stop ends client's stream, reporting whether it had one.465525 func (s *Server) stop(client string) bool {466526 s.mu.Lock()@@ -490,6 +550,11 @@ func (s *Server) Close() {
490550 all = append(all, x)491551 }492552 s.sessions = make(map[string]*session)553+ select {554+ case <-s.quit:555+ default:556+ close(s.quit)557+ }493558 s.mu.Unlock()494559 for _, x := range all {495560 x.stop()@@ -521,6 +586,7 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
521586 http.NotFound(w, r)522587 return523588 }589+ sess.touch(time.Now())524590 switch filepath.Ext(file) {525591 case ".m3u8":526592 w.Header().Set("Content-Type", "application/vnd.apple.mpegurl")@@ -0,0 +1,71 @@
1+package stream2+3+import (4+ "context"5+ "net/http"6+ "net/http/httptest"7+ "os"8+ "path/filepath"9+ "testing"10+ "time"11+)12+13+// fakeSession adds a session without ffmpeg, last used at used.14+func fakeSession(t *testing.T, s *Server, client string, used time.Time) *session {15+ t.Helper()16+ dir := filepath.Join(t.TempDir(), client)17+ if err := os.MkdirAll(dir, 0o755); err != nil {18+ t.Fatal(err)19+ }20+ if err := os.WriteFile(filepath.Join(dir, "stream_0.m3u8"), []byte("#EXTM3U\n"), 0o644); err != nil {21+ t.Fatal(err)22+ }23+ _, cancel := context.WithCancel(context.Background())24+ sess := &session{id: client + "-id", client: client, dir: dir, started: used, cancel: cancel, done: make(chan struct{}), stderr: &tail{max: 64}}25+ close(sess.done)26+ sess.touch(used)27+ s.mu.Lock()28+ s.sessions[client] = sess29+ s.mu.Unlock()30+ return sess31+}32+33+// TestIdleStreams: a stream nobody fetches from or touches is stopped once34+// idle too long; one fetched from or touched is kept.35+func TestIdleStreams(t *testing.T) {36+ s := &Server{sessions: map[string]*session{}, idle: time.Minute, quit: make(chan struct{})}37+ start := time.Now().Add(-2 * time.Minute)38+ gone := fakeSession(t, s, "gone", start)39+ fakeSession(t, s, "paused", start)40+ fetched := fakeSession(t, s, "watching", start)41+42+ if !s.Touch("paused") {43+ t.Fatal("Touch: no stream for paused")44+ }45+ rec := httptest.NewRecorder()46+ s.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/live/"+fetched.id+"/stream_0.m3u8", nil))47+ if rec.Code != http.StatusOK {48+ t.Fatalf("playlist: HTTP %d", rec.Code)49+ }50+51+ s.reapIdle(time.Now())52+ if s.Touch("gone") {53+ t.Error("an idle stream wasn't stopped")54+ }55+ if _, err := os.Stat(gone.dir); err == nil {56+ t.Error("an idle stream's files are left")57+ }58+ for _, client := range []string{"paused", "watching"} {59+ if !s.Touch(client) {60+ t.Errorf("%s: stopped though in use", client)61+ }62+ }63+64+ // Later, with nothing since, they go too.65+ s.reapIdle(time.Now().Add(2 * time.Minute))66+ for _, client := range []string{"paused", "watching"} {67+ if s.Touch(client) {68+ t.Errorf("%s: still there after idling", client)69+ }70+ }71+}