Add SSE events on the http server and polochonfs

Author: gregdel

app/app/app.go

@@ -116,7 +116,9 @@ func (a *App) init() error {
 		}
 
 		// Add the http server
-		a.subApps = append(a.subApps, server.New(config, library, authManager))
+		srv := server.New(config, library, authManager)
+		config.Notifiers = append(config.Notifiers, srv.Hub())
+		a.subApps = append(a.subApps, srv)
 	}
 
 	log.Debug("app configuration loaded")

app/server/http.go

@@ -29,6 +29,7 @@ type Server struct {
 	library        *library.Library
 	authManager    *auth.Manager
 	gracefulServer *http.Server
+	hub            *sseHub
 	log            *logrus.Entry
 	render         *render.Render
 }
@@ -40,6 +41,7 @@ func New(config *configuration.Config, vs *library.Library, auth *auth.Manager)
 		config:      config,
 		library:     vs,
 		authManager: auth,
+		hub:         newSSEHub(),
 		render:      render.New(),
 	}
 }
@@ -66,6 +68,11 @@ func (s *Server) Stop(log *logrus.Entry) {
 	}
 }
 
+// Hub returns the SSE hub as a Notifier
+func (s *Server) Hub() polochon.Notifier {
+	return s.hub
+}
+
 func (s *Server) wishlist(w http.ResponseWriter, r *http.Request) {
 	log := s.logEntry(r)
 	log.Infof("getting wishlist")

app/server/library.go

@@ -11,7 +11,9 @@ func (s *Server) libraryRefresh(w http.ResponseWriter, req *http.Request) {
 	if err := s.library.RebuildIndex(log); err != nil {
 		log.WithField("function", "rebuild_index").Error(err)
 		s.renderError(w, req, err)
+		return
 	}
 
+	s.hub.broadcast()
 	s.renderOK(w, nil)
 }

app/server/movies.go

@@ -102,5 +102,6 @@ func (s *Server) deleteMovie(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	s.hub.broadcast()
 	s.renderOK(w, nil)
 }

app/server/routes.go

@@ -192,6 +192,12 @@ func (s *Server) httpServer(log *logrus.Entry) *http.Server {
 			handler:  s.serveEpisodeSubtitle,
 			excluded: !s.config.HTTPServer.ServeFiles,
 		},
+		{
+			name:    "Events",
+			path:    "/events",
+			methods: "GET",
+			handler: s.events,
+		},
 		{
 			name:    "Wishlist",
 			path:    "/wishlist",

app/server/seasons.go

@@ -98,5 +98,6 @@ func (s *Server) deleteSeason(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	s.hub.broadcast()
 	s.renderOK(w, nil)
 }

app/server/shows.go

@@ -105,6 +105,7 @@ func (s *Server) deleteShow(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	s.hub.broadcast()
 	s.renderOK(w, nil)
 }
 
@@ -152,6 +153,7 @@ func (s *Server) deleteEpisode(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	s.hub.broadcast()
 	s.renderOK(w, nil)
 }
 

app/server/sse.go

@@ -0,0 +1,90 @@
+package server
+
+import (
+	"io"
+	"net/http"
+	"sync"
+
+	polochon "github.com/odwrtw/polochon/lib"
+	"github.com/sirupsen/logrus"
+)
+
+const sseModuleName = "sse"
+
+// Compile-time assertion.
+var _ polochon.Notifier = (*sseHub)(nil)
+
+type sseHub struct {
+	mu      sync.Mutex
+	clients map[chan struct{}]struct{}
+}
+
+func newSSEHub() *sseHub {
+	return &sseHub{
+		clients: make(map[chan struct{}]struct{}),
+	}
+}
+
+// Module interface.
+func (h *sseHub) Init(_ []byte) error                    { return nil }
+func (h *sseHub) Name() string                           { return sseModuleName }
+func (h *sseHub) Status() (polochon.ModuleStatus, error) { return polochon.StatusOK, nil }
+
+// Notifier interface.
+func (h *sseHub) Notify(_ any, _ *logrus.Entry) error {
+	h.broadcast()
+	return nil
+}
+
+func (h *sseHub) subscribe() chan struct{} {
+	ch := make(chan struct{}, 1)
+	h.mu.Lock()
+	h.clients[ch] = struct{}{}
+	h.mu.Unlock()
+	return ch
+}
+
+func (h *sseHub) unsubscribe(ch chan struct{}) {
+	h.mu.Lock()
+	delete(h.clients, ch)
+	h.mu.Unlock()
+	close(ch)
+}
+
+func (h *sseHub) broadcast() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	for ch := range h.clients {
+		// Non-blocking send to coalesce duplicate events.
+		select {
+		case ch <- struct{}{}:
+		default:
+		}
+	}
+}
+
+func (s *Server) events(w http.ResponseWriter, r *http.Request) {
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		http.Error(w, "streaming not supported", http.StatusInternalServerError)
+		return
+	}
+
+	ch := s.hub.subscribe()
+	defer s.hub.unsubscribe(ch)
+
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	flusher.Flush()
+
+	for {
+		select {
+		case <-ch:
+			_, _ = io.WriteString(w, "data:\n\n")
+			flusher.Flush()
+		case <-r.Context().Done():
+			return
+		}
+	}
+}

app/server/sse_test.go

@@ -0,0 +1,103 @@
+package server
+
+import (
+	"bufio"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func TestSSEHubSubscribeBroadcast(t *testing.T) {
+	hub := newSSEHub()
+
+	ch1 := hub.subscribe()
+	ch2 := hub.subscribe()
+
+	hub.broadcast()
+
+	select {
+	case <-ch1:
+	case <-time.After(time.Second):
+		t.Fatal("ch1 did not receive broadcast")
+	}
+	select {
+	case <-ch2:
+	case <-time.After(time.Second):
+		t.Fatal("ch2 did not receive broadcast")
+	}
+
+	hub.unsubscribe(ch1)
+	hub.broadcast()
+
+	// ch1 should be closed
+	select {
+	case _, ok := <-ch1:
+		if ok {
+			t.Fatal("ch1 should be closed")
+		}
+	default:
+		t.Fatal("ch1 should be closed and readable")
+	}
+
+	// ch2 should still receive
+	select {
+	case <-ch2:
+	case <-time.After(time.Second):
+		t.Fatal("ch2 did not receive broadcast after ch1 unsubscribed")
+	}
+
+	hub.unsubscribe(ch2)
+}
+
+func TestSSEHubCoalescing(t *testing.T) {
+	hub := newSSEHub()
+	ch := hub.subscribe()
+	defer hub.unsubscribe(ch)
+
+	// Multiple broadcasts before consuming should coalesce into one.
+	hub.broadcast()
+	hub.broadcast()
+	hub.broadcast()
+
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		t.Fatal("did not receive coalesced broadcast")
+	}
+
+	// Channel should be empty now.
+	select {
+	case <-ch:
+		t.Fatal("should not receive a second event (coalescing)")
+	default:
+	}
+}
+
+func TestSSEHandler(t *testing.T) {
+	hub := newSSEHub()
+	s := &Server{hub: hub}
+
+	ts := httptest.NewServer(http.HandlerFunc(s.events))
+	defer ts.Close()
+
+	resp, err := http.Get(ts.URL)
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if ct := resp.Header.Get("Content-Type"); ct != "text/event-stream" {
+		t.Fatalf("expected Content-Type text/event-stream, got %q", ct)
+	}
+
+	hub.broadcast()
+
+	scanner := bufio.NewScanner(resp.Body)
+	for scanner.Scan() {
+		if scanner.Text() == "data:" {
+			return
+		}
+	}
+	t.Fatal("did not receive SSE data line")
+}

app/server/subtitles.go

@@ -98,6 +98,7 @@ func (s *Server) updateSubtitle(v polochon.Video, w http.ResponseWriter, r *http
 		return
 	}
 
+	s.hub.broadcast()
 	s.renderOK(w, sub)
 }
 
@@ -131,6 +132,7 @@ func (s *Server) uploadSubtitle(v polochon.Video, w http.ResponseWriter, r *http
 		return
 	}
 
+	s.hub.broadcast()
 	s.renderOK(w, sub)
 }