bus.go RawHistoryFolder

1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3// Package eventbus implements ports.EventBus: in-process fan-out to
4// subscribers, never blocking the publisher.
5package eventbus
6
7import (
8	"context"
9	"sync"
10
11	"bnema.dev/ermite/internal/domain"
12	"bnema.dev/ermite/internal/ports"
13)
14
15// DefaultBuffer is the number of events a subscriber may lag behind.
16const DefaultBuffer = 64
17
18var _ ports.EventBus = (*Bus)(nil)
19
20// Bus is the in-process event bus.
21type Bus struct {
22	buffer int
23
24	mu   sync.Mutex
25	subs map[chan domain.Event]struct{}
26}
27
28// New builds a bus whose subscribers buffer DefaultBuffer events.
29func New() *Bus { return NewWithBuffer(DefaultBuffer) }
30
31// NewWithBuffer builds a bus whose subscribers buffer n events (at least 2,
32// so that a gap marker and the event that follows it fit).
33func NewWithBuffer(n int) *Bus {
34	return &Bus{buffer: max(n, 2), subs: make(map[chan domain.Event]struct{})}
35}
36
37// Publish delivers ev to every subscriber. A subscriber whose buffer is full
38// is cleared and gets a domain.EventGap followed by ev: it lost events and
39// must refresh its view, but it is never blocked on and never blocks others.
40func (b *Bus) Publish(_ context.Context, ev domain.Event) {
41	b.mu.Lock()
42	defer b.mu.Unlock()
43	for ch := range b.subs {
44		select {
45		case ch <- ev:
46			continue
47		default:
48		}
49		// Full. Only Publish sends and it holds b.mu, so after draining
50		// there is room for the marker and the event.
51	drain:
52		for {
53			select {
54			case <-ch:
55			default:
56				break drain
57			}
58		}
59		ch <- domain.EventGap{}
60		ch <- ev
61	}
62}
63
64// Subscribe registers a subscriber until ctx ends, then closes its channel.
65func (b *Bus) Subscribe(ctx context.Context) <-chan domain.Event {
66	ch := make(chan domain.Event, b.buffer)
67	b.mu.Lock()
68	b.subs[ch] = struct{}{}
69	b.mu.Unlock()
70
71	context.AfterFunc(ctx, func() {
72		b.mu.Lock()
73		delete(b.subs, ch)
74		close(ch)
75		b.mu.Unlock()
76	})
77	return ch
78}