66 lines
1.5 KiB
Go
66 lines
1.5 KiB
Go
package httpapi
|
|
|
|
import (
|
|
"encoding/json"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
type queueEvent struct {
|
|
ProjectID string `json:"project_id"`
|
|
Revision int64 `json:"revision"`
|
|
Type string `json:"type"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
|
|
// eventPublisher is the boundary for replacing the per-process fallback with
|
|
// a shared Redis-backed bus when Kubernetes runs multiple API pods. The
|
|
// current MVP intentionally keeps the in-memory implementation local.
|
|
type eventPublisher interface {
|
|
subscribe(projectID string) (<-chan []byte, func())
|
|
publish(event queueEvent)
|
|
}
|
|
|
|
type eventHub struct {
|
|
mu sync.RWMutex
|
|
subscribers map[string]map[chan []byte]struct{}
|
|
}
|
|
|
|
func newEventHub() *eventHub {
|
|
return &eventHub{subscribers: make(map[string]map[chan []byte]struct{})}
|
|
}
|
|
|
|
func (h *eventHub) subscribe(projectID string) (<-chan []byte, func()) {
|
|
ch := make(chan []byte, 8)
|
|
h.mu.Lock()
|
|
if h.subscribers[projectID] == nil {
|
|
h.subscribers[projectID] = make(map[chan []byte]struct{})
|
|
}
|
|
h.subscribers[projectID][ch] = struct{}{}
|
|
h.mu.Unlock()
|
|
return ch, func() {
|
|
h.mu.Lock()
|
|
delete(h.subscribers[projectID], ch)
|
|
if len(h.subscribers[projectID]) == 0 {
|
|
delete(h.subscribers, projectID)
|
|
}
|
|
h.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (h *eventHub) publish(event queueEvent) {
|
|
body, err := json.Marshal(event)
|
|
if err != nil {
|
|
return
|
|
}
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
for ch := range h.subscribers[event.ProjectID] {
|
|
select {
|
|
case ch <- body:
|
|
default:
|
|
// A slow client can recover from the authoritative snapshot/revision.
|
|
}
|
|
}
|
|
}
|