feat(evt): add EmitAsync for non-blocking event dispatch

Emit blocks the caller until all listeners return (listeners run
concurrently with each other, but the caller waits via wg.Wait). EmitAsync
detaches dispatch into a background goroutine so producers that must not
wait on listener latency (e.g. a listener doing network I/O) return
immediately.
This commit is contained in:
Dimitri Herzog
2026-06-06 21:52:52 +02:00
parent 21de3077ec
commit 3da3b40366
2 changed files with 74 additions and 4 deletions
+29 -4
View File
@@ -4,9 +4,13 @@
// Producers and consumers use the package-level generic functions Emit, Subscribe,
// and Unsubscribe; the bus itself never grows public fields per event.
//
// All operations are nil-safe: passing a nil bus makes Emit / Subscribe /
// Unsubscribe no-ops. This is used by resolver.Bootstrap to silence its
// internal CachingResolver.
// All operations are nil-safe: passing a nil bus makes Emit / EmitAsync /
// Subscribe / Unsubscribe no-ops. This is used by resolver.Bootstrap to silence
// its internal CachingResolver.
//
// Emit blocks the caller until every listener returns (listeners run
// concurrently with each other, but not with the caller). Use EmitAsync when
// the producer must not wait on listener latency.
package evt
import (
@@ -28,7 +32,9 @@ func NewBus() *Bus {
return &Bus{signals: make(map[reflect.Type]any)}
}
// Emit publishes an event of type T to all subscribers. A nil bus is a no-op.
// Emit publishes an event of type T to all subscribers and blocks until every
// listener returns. Listeners run concurrently with each other but not with the
// caller. A nil bus is a no-op.
func Emit[T any](b *Bus, ctx context.Context, event T) {
if b == nil {
return
@@ -36,6 +42,25 @@ func Emit[T any](b *Bus, ctx context.Context, event T) {
signalFor[T](b).Emit(ctx, event)
}
// EmitAsync publishes an event of type T without blocking the caller: dispatch
// to listeners happens in a detached background goroutine and EmitAsync returns
// immediately. Use it when the producer must not wait on listener latency (e.g.
// an emit whose listener performs network I/O). A nil bus is a no-op.
//
// Because dispatch is detached, the caller cannot observe when listeners finish,
// and ordering relative to later Emit/EmitAsync calls is not guaranteed. The
// supplied ctx is captured by the goroutine; if the caller cancels it right
// after the call, listeners observe a cancelled context.
func EmitAsync[T any](b *Bus, ctx context.Context, event T) {
if b == nil {
return
}
signal := signalFor[T](b)
go signal.Emit(ctx, event)
}
// Subscribe registers a listener for events of type T under the given key. A
// nil bus is a no-op. The key may later be passed to Unsubscribe to remove
// the listener.
+45
View File
@@ -97,6 +97,51 @@ var _ = Describe("Bus", func() {
})
})
Describe("EmitAsync", func() {
When("the bus is nil", func() {
It("is a no-op", func() {
Expect(func() {
evt.EmitAsync[fooEvent](nil, context.Background(), fooEvent{N: 1})
}).ShouldNot(Panic())
})
})
When("a listener is registered", func() {
It("delivers the event eventually", func(specCtx context.Context) {
received := make(chan fooEvent, 1)
evt.Subscribe(bus, "k", func(_ context.Context, e fooEvent) {
received <- e
})
evt.EmitAsync(bus, specCtx, fooEvent{N: 99})
Eventually(received).WithTimeout(time.Second).Should(Receive(Equal(fooEvent{N: 99})))
})
})
When("a listener blocks", func() {
It("returns to the caller without waiting for the listener", func(specCtx context.Context) {
release := make(chan struct{})
started := make(chan struct{}, 1)
evt.Subscribe(bus, "slow", func(_ context.Context, _ fooEvent) {
started <- struct{}{}
<-release // block until the test releases us
})
DeferCleanup(func() { close(release) })
returned := make(chan struct{})
go func() {
evt.EmitAsync(bus, specCtx, fooEvent{})
close(returned)
}()
// EmitAsync must return promptly even though the listener is still blocked.
Eventually(returned).WithTimeout(time.Second).Should(BeClosed())
Eventually(started).WithTimeout(time.Second).Should(Receive())
})
})
})
Describe("Subscribe with nil bus", func() {
It("is a no-op for Subscribe and Unsubscribe", func() {
Expect(func() {