Filter stream events before queueing them

internal/server/hub.go

@@ -8,18 +8,18 @@ import (
 
 type hub struct {
 	mu   sync.Mutex
-	subs map[chan api.Event]bool
+	subs map[chan api.Event]func(api.Event) bool
 }
 
 func newHub() *hub {
-	return &hub{subs: map[chan api.Event]bool{}}
+	return &hub{subs: map[chan api.Event]func(api.Event) bool{}}
 }
 
-func (h *hub) subscribe() chan api.Event {
+func (h *hub) subscribe(wanted func(api.Event) bool) chan api.Event {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 	ch := make(chan api.Event, 256)
-	h.subs[ch] = true
+	h.subs[ch] = wanted
 	return ch
 }
 
@@ -37,7 +37,10 @@ func (h *hub) unsubscribe(ch chan api.Event) {
 func (h *hub) publish(ev api.Event) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
-	for ch := range h.subs {
+	for ch, wanted := range h.subs {
+		if !wanted(ev) {
+			continue
+		}
 		select {
 		case ch <- ev:
 		default:

internal/server/sse.go

@@ -55,7 +55,7 @@ func (s *Server) StreamEvents(w http.ResponseWriter, r *http.Request, params api
 	w.Header().Set("Cache-Control", "no-cache")
 	w.WriteHeader(http.StatusOK)
 
-	live := s.hub.subscribe()
+	live := s.hub.subscribe(wanted)
 	defer s.clientsChanged()
 	defer s.hub.unsubscribe(live)
 	s.clientsChanged()
@@ -96,9 +96,6 @@ func (s *Server) StreamEvents(w http.ResponseWriter, r *http.Request, params api
 			if ev.Seq != 0 && ev.Seq <= last {
 				continue
 			}
-			if !wanted(ev) {
-				continue
-			}
 			if !writeEvent(w, ev) {
 				return
 			}