refactor(eventbus): unsubscribe with context.AfterFunc and key subscribers by channel

Drops the per-subscriber goroutine and the subscriber struct. Subscribers()
moves to export_test.go.

commit: 58ac19e70fed8c963b404bd373f18b0f335836c1
author: brice · 12d ago · 2026-10-10 08:11 UTC
parent: a771dd2
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +24 -26

M internal/adapters/eventbus/bus.go +14 -26


@@ -22,11 +22,7 @@ type Bus struct {
22	22		buffer int
23	23
24	24		mu sync.Mutex
25		-	subs map[*subscriber]struct{}
26		-	}
27		-
28		-	type subscriber struct {
29		-	ch chan domain.Event
	25	+	subs map[chan domain.Event]struct{}
30	26		}
31	27
32	28		// New builds a bus whose subscribers buffer DefaultBuffer events.

@@ -35,7 +31,7 @@ func New() *Bus { return NewWithBuffer(DefaultBuffer) }
35	31		// NewWithBuffer builds a bus whose subscribers buffer n events (at least 2,
36	32		// so that a gap marker and the event that follows it fit).
37	33		func NewWithBuffer(n int) *Bus {
38		-	return &Bus{buffer: max(n, 2), subs: make(map[*subscriber]struct{})}
	34	+	return &Bus{buffer: max(n, 2), subs: make(map[chan domain.Event]struct{})}
39	35		}
40	36
41	37		// Publish delivers ev to every subscriber. A subscriber whose buffer is full

@@ -44,9 +40,9 @@ func NewWithBuffer(n int) *Bus {
44	40		func (b *Bus) Publish(_ context.Context, ev domain.Event) {
45	41		b.mu.Lock()
46	42		defer b.mu.Unlock()
47		-	for s := range b.subs {
	43	+	for ch := range b.subs {
48	44		select {
49		-	case s.ch <- ev:
	45	+	case ch <- ev:
50	46		continue
51	47		default:
52	48		}

@@ -55,36 +51,28 @@ func (b *Bus) Publish(_ context.Context, ev domain.Event) {
55	51		drain:
56	52		for {
57	53		select {
58		-	case <-s.ch:
	54	+	case <-ch:
59	55		default:
60	56		break drain
61	57		}
62	58		}
63		-	s.ch <- domain.EventGap{}
64		-	s.ch <- ev
	59	+	ch <- domain.EventGap{}
	60	+	ch <- ev
65	61		}
66	62		}
67	63
68	64		// Subscribe registers a subscriber until ctx ends, then closes its channel.
69	65		func (b *Bus) Subscribe(ctx context.Context) <-chan domain.Event {
70		-	s := &subscriber{ch: make(chan domain.Event, b.buffer)}
	66	+	ch := make(chan domain.Event, b.buffer)
71	67		b.mu.Lock()
72		-	b.subs[s] = struct{}{}
	68	+	b.subs[ch] = struct{}{}
73	69		b.mu.Unlock()
74	70
75		-	go func() {
76		-	<-ctx.Done()
	71	+	context.AfterFunc(ctx, func() {
77	72		b.mu.Lock()
78		-	delete(b.subs, s)
79		-	close(s.ch)
	73	+	delete(b.subs, ch)
	74	+	close(ch)
80	75		b.mu.Unlock()
81		-	}()
82		-	return s.ch
83		-	}
84		-
85		-	// Subscribers returns the number of live subscribers, for tests.
86		-	func (b *Bus) Subscribers() int {
87		-	b.mu.Lock()
88		-	defer b.mu.Unlock()
89		-	return len(b.subs)
	76	+	})
	77	+	return ch
90	78		}

 	buffer int
 	mu   sync.Mutex
-	subs map[*subscriber]struct{}
+	subs map[chan domain.Event]struct{}
-}
-type subscriber struct {
-	ch chan domain.Event
 }
 // New builds a bus whose subscribers buffer DefaultBuffer events.
 // NewWithBuffer builds a bus whose subscribers buffer n events (at least 2,
 // so that a gap marker and the event that follows it fit).
 func NewWithBuffer(n int) *Bus {
-	return &Bus{buffer: max(n, 2), subs: make(map[*subscriber]struct{})}
+	return &Bus{buffer: max(n, 2), subs: make(map[chan domain.Event]struct{})}
 }
 // Publish delivers ev to every subscriber. A subscriber whose buffer is full
 func (b *Bus) Publish(_ context.Context, ev domain.Event) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	for s := range b.subs {
+	for ch := range b.subs {
 		select {
-		case s.ch <- ev:
+		case ch <- ev:
 			continue
 		default:
 		}
 	drain:
 		for {
 			select {
-			case <-s.ch:
+			case <-ch:
 			default:
 				break drain
 			}
 		}
-		s.ch <- domain.EventGap{}
+		ch <- domain.EventGap{}
-		s.ch <- ev
+		ch <- ev
 	}
 }
 // Subscribe registers a subscriber until ctx ends, then closes its channel.
 func (b *Bus) Subscribe(ctx context.Context) <-chan domain.Event {
-	s := &subscriber{ch: make(chan domain.Event, b.buffer)}
+	ch := make(chan domain.Event, b.buffer)
 	b.mu.Lock()
-	b.subs[s] = struct{}{}
+	b.subs[ch] = struct{}{}
 	b.mu.Unlock()
-	go func() {
+	context.AfterFunc(ctx, func() {
-		<-ctx.Done()
 		b.mu.Lock()
-		delete(b.subs, s)
+		delete(b.subs, ch)
-		close(s.ch)
+		close(ch)
 		b.mu.Unlock()
-	}()
+	})
-	return s.ch
+	return ch
-}
-// Subscribers returns the number of live subscribers, for tests.
-func (b *Bus) Subscribers() int {
-	b.mu.Lock()
-	defer b.mu.Unlock()
-	return len(b.subs)
 }

A internal/adapters/eventbus/export_test.go +10 -0

@@ -0,0 +1,10 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package eventbus
	4	+
	5	+	// Subscribers returns the number of live subscribers.
	6	+	func (b *Bus) Subscribers() int {
	7	+	b.mu.Lock()
	8	+	defer b.mu.Unlock()
	9	+	return len(b.subs)
	10	+	}

@@ -0,0 +1,10 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package eventbus
	4	+
	5	+	// Subscribers returns the number of live subscribers.
	6	+	func (b *Bus) Subscribers() int {
	7	+	b.mu.Lock()
	8	+	defer b.mu.Unlock()
	9	+	return len(b.subs)
	10	+	}