diff --git a/internal/server/hub.go b/internal/server/hub.go index 288a54e..5e63ba7 100644 --- a/internal/server/hub.go +++ b/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: diff --git a/internal/server/sse.go b/internal/server/sse.go index 09d5dd5..703eb7e 100644 --- a/internal/server/sse.go +++ b/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 }