conflate

README

conflate

Keyed latest-value fan-out: one slot per key, coalesced latest-wins or by a caller-supplied merge.

Part of gobus.

import "github.com/amorey/gobus/conflate"

Contents

• Overview
• Quick start
• Semantics
• Receiver options
• Inspecting the backlog head
• Taking the whole backlog at once
• Publishing with no receivers
• Contexts and cancellation
• Close semantics
• Chan support
• Thread safety
• API reference
• Errors
• Examples

Overview

Conflate is a single-producer, multi-consumer keyed latest-value bus. Every value published through the singleton Sender is fanned out to every live Receiver, but each receiver holds one slot per key plus an insertion-ordered key queue. A Send for a key that already has an undelivered value coalesces into that slot via the hub's Merge instead of appending, and the key keeps its original queue position. That merge is latest-wins unless WithDefaultMerge set another.

Because conflate keeps the latest value per key, a slow receiver never lags-as-loss: it catches up to the current state of every key in first-touch order, and its memory stays bounded by the live key set rather than by write volume. There is no capacity argument and no lag error, because there is no unbounded backlog to grow.

Typical uses: streaming resource state to watchers (Kubernetes-style informers), UI or dashboard update feeds where only the current state of each entity matters, cache invalidation fan-out, incremental index maintenance, and any "coalesce writes per entity, deliver at consumer pace" pipeline.

Pick conflate over watch when a consumer follows many keys at once, or when it must observe a key's history collapsed rather than just its latest state — in particular when a create-then-delete pair must leave no residue, which is what Merge's annihilation gives you.

Quick start

// No options: the newer value supersedes the undelivered one.
hub := conflate.New[string, Update]()
defer hub.Close()

tx := hub.Sender()
defer tx.Close()

rx := hub.Receiver()
defer rx.Close()

go func() {
    for {
        ev, err := rx.Recv()
        if err != nil {
            return // ErrClosed
        }
        // ev.Value is the *latest* Update for ev.Key, not every intermediate one.
        apply(ev.Key, ev.Value)
    }
}()

for _, u := range updates {
    tx.Send(u.Key, u)
}

Semantics

Latest-value-per-key delivery. Each receiver owns an insertion-ordered queue of keys plus one value slot per key. A Send for a key with no pending slot appends the key at the back of the queue; a Send for a key that is already pending coalesces into the existing slot via Merge and leaves the key's queue position unchanged. Delivery order is therefore first-touch order, and a hot key does not starve a cold one by repeatedly jumping the queue.

Coalescing happens at Send, not at Recv. A read is a plain pop. What the consumer gets is whatever the slot had accumulated by the moment it read.

Bounded by the key set, not the write volume. A thousand sends across four keys leave at most four pending entries.

Merge may annihilate. Returning keep == false drops the key entirely — queue entry and value slot both — so a create/delete pair the consumer never observed leaves no residue at all.

Send never blocks. Slow receivers cannot apply backpressure to the publisher; they simply coalesce more aggressively.

A receiver starts empty. It observes values sent after it was created, not the producer's history. See Publishing with no receivers for the ordering rule that follows.

Merge and any key filter are caller code running under the bus lock. They must not call back into the hub, and they must not take a lock another goroutine may hold while calling into the bus. Reading their arguments and nothing else is always safe.

Receiver options

Receivers take composable options, minted by the hub itself — that fixes K and V from the hub, so call sites need no type arguments and an option built from a differently-typed hub is a compile error. (WithDefaultMerge, which configures the hub, is package-level for the reason given below.)

rx := hub.Receiver()                                  // every key, hub's merge
rx := hub.Receiver(hub.WithKey(k))                    // one key, no predicate
rx := hub.Receiver(hub.WithKeyFilter(wanted))         // one key subset
rx := hub.Receiver(hub.WithMerge(stricter))           // own coalescing policy
rx := hub.Receiver(hub.WithKeyFilter(wanted), hub.WithMerge(stricter))

WithKey and WithKeyFilter both filter at enqueue, so an unwanted key never occupies a slot — that is how a consumer watching part of a high-cardinality producer stays bounded by the keys it wants.

WithKey names one key and the send path compares it, running no caller code on that receiver's behalf. WithKeyFilter takes a predicate, which runs under the bus lock for every send. Reach for WithKey when one key is all you need; the zero K is a usable key.

The two are the same setting, so the later of the two wins rather than composing — a receiver is filtered by a key or by a predicate, never by both. ANDing them would have no use anyway: a predicate over a single known key is a constant.

WithMerge gives a single consumer its own coalescing policy, which matters when consumers of the same producer disagree about what may be dropped; one hub-wide Merge cannot express that.

Later options win over earlier ones for the same setting. A nil option, or a nil function passed to any option constructor, panics: omitting an option states its default, supplying a nil one is a mistake.

Inspecting the backlog head

Peek() returns the oldest pending event without removing it — TryRecv minus the pop, sharing its precedence exactly: ErrEmpty when nothing is pending, ErrClosed when the receiver or hub is closed or the sender has closed and the queue has drained.

ev, err := rx.Peek()   // what would Recv hand me next, without taking it?

It is not a raw read of the queue, so a closed handle reports ErrClosed even with a value at the head. The corollary matters if you track a cursor: ErrClosed is not a statement that the backlog was empty, because Receiver.Close() and Hub.Close() abandon whatever is still queued. Only Sender.Close() drains first.

The value you get back is the current merged contents of the head key's slot, so it can change between two Peeks while the head key does not — coalescing leaves queue position alone. Annihilation is the exception: a Merge returning keep == false for the head key removes it, and the next Peek reports a different key.

Two cautions. Peek takes the same hub lock that serializes the entire Send fan-out, so polling it in a loop degrades every publisher and every other receiver on the bus — call it once per unit of work. When that unit of work is a whole burst, TryRecvAll is how you say so in one call. And while it is safe to call from any goroutine, it is only meaningful on the receiver's single consuming goroutine: anything else consuming concurrently can take the event you just looked at. An event already handed to the Chan feeder has left the queue, so Peek reports ErrEmpty while that one event is in flight.

The intended use is a consumer that needs to know how far its backlog reaches — a resumable cursor or watermark. Fold the ordering quantity into V and let the bus's own Merge carry it: stamp each value at publication, keep the earliest stamp when two values for a key coalesce, and read it off the head. The bus makes no ordering claim of its own here, so the premise that publication order matches that quantity's order is yours to hold; if it doesn't, the head key's stamp is not the lowest one pending.

Taking the whole backlog at once

TryRecvAll() pops everything pending, in queue order, and empties the queue — TryRecv applied to the whole queue instead of the head, under one acquisition of the hub lock. It shares that precedence too: ErrEmpty when nothing is pending, ErrClosed on the same terminal conditions.

evs, err := rx.TryRecvAll()   // everything pending, as of one instant

The atomicity is the contract, not an optimization. A loop of TryRecv is a sequence of instants, so the batch it assembles has no defined membership — a Send landing between two iterations joins it, one landing just after the loop observes empty does not — and no caller can close that gap from outside the lock. This matters most for the cursor case above. Conflate delivers in first-touch order, which carries no relation to any ordering quantity inside V, so a consumer that sorts its batch by that quantity needs the batch to be a complete set: take a proper subset and the next batch interleaves with the one you already committed. TryRecvAll supplies the set; you still sort it.

Two properties of the returned slice are worth building on. It is in queue order, which is first-touch order — sort if you need value order. And it holds one entry per key, since a receiver's queue holds each key once, so it folds into a map without deduping. That second one belongs to a single call: an event that has already left the receiver's slots cannot be coalesced into, so a batch you assemble from a Recv plus a TryRecvAll can contain the same key twice.

Partial results are not a case — either every pending event with a nil error, or no events and an error — so you can test the error and ignore the slice. Against Sender.Close() the whole queue still comes back with a nil error and only the next call reports ErrClosed.

There is no max parameter. A cap would hand back exactly the split the method exists to prevent, and it is unnecessary: a receiver's memory is already bounded by the live key set rather than by write volume, so "everything pending" is bounded by construction. Note the trade against a TryRecv loop, though: total lock acquisitions go down, but this one is held for the length of the queue, so worst-case publisher latency goes up. No caller code runs inside it — no Merge, no key filter — which is what keeps that hold bounded.

Publishing with no receivers

Send on a hub with no live receiver returns nil without taking the bus lock. That lock is hub-wide — every Send fan-out, every pop, Recv, Peek, TryRecv, TryRecvAll and Close serializes through it — so the cost of an unwatched hub otherwise lands on the producer's hot path, per publish, whether or not anything reads the result. This matters when values are published from inside a producer's own critical sections: without it, every write in a subsystem pays a mutex for a bus nobody is subscribed to.

The result is unchanged, only the cost: a send with no receivers already fanned out to nobody. TrySend and SendContext take the same path, and a cancelled ctx is still reported rather than swallowed. A closed sender still returns ErrClosed on an empty receiver set.

The corollary is an ordering requirement on your side, and it is the one thing to get right:

rx := hub.Receiver()   // register FIRST
state := snapshot()    // then read your snapshot

Take the snapshot before registering and any value published in the gap reaches no receiver — conflate has no replay, and the next Send for that key coalesces into a slot your consumer was never told had been skipped. This has always been conflate's delivery model; a publisher that skips the lock simply makes it unforgiving.

(watch removes this rule by taking your snapshot as its registration argument. See its README.)

Contexts and cancellation

Precedence is closed > cancelled > value on the receive side and closed > cancelled on the send side; both are pinned by the module's conformance suite. The full rationale lives in the root README. What is conflate-specific:

• SendContext never blocks, so ctx is consulted exactly once, where the send is resolved — under the bus lock, not on entry. A ctx cancelled while the call waited for the lock publishes nothing and returns ctx.Err(). Waiting for that lock is real work: your Merge and key filters run under it.
• RecvContext returns ctx.Err() even when an event is pending, and leaves that event queued. Without it, a consumer looping against a fast publisher would take a value every iteration and never notice its own shutdown.
• ctx.Err() is not an end-of-stream and does not deregister the receiver. A caller that stops on it must Close the handle, or it keeps coalescing — one slot per live key — for the hub's lifetime. defer rx.Close() covers it.
• To consume what is left first, call TryRecvAll once: the error tells you which state you stopped in — ErrEmpty while the sender is open, ErrClosed once it has closed and the queue is drained. The flush is not a substitute for the Close: against a still-open sender it ends on ErrEmpty, which is not terminal and does not deregister.

Close semantics

Call	Effect
Sender.Close()	Graceful end-of-stream. Each receiver drains its pending per-key values once, then sees ErrClosed / a closed Chan.
Receiver.Close()	This handle only. Other receivers and the sender keep running; this handle's pending values are abandoned and its Chan feeder shuts down.
Hub.Close()	Hard tear-down: sender plus every live receiver, with no drain. Future Hub.Receiver() calls return pre-closed handles.

All idempotent. Don't call Hub.Close concurrently with an active Send from another goroutine — it tears down the receivers that send is fanning out to.

Sender.Close is safe to call concurrently with a Send or SendContext from another goroutine. The two serialize, so a racing send resolves to exactly one of two outcomes — it publishes and returns nil, or it returns ErrClosed and publishes nothing. There is no third outcome and no partial one. Which ordering wins is unspecified: a caller that needs a value visible before shutdown must order the two itself, and a caller shutting down that doesn't care whether the last value lands needs no fence on its write path. This holds because conflate's Send never parks; it is a promise about this package, not a module-wide rule.

A receiver that reaches the terminal ErrClosed after a Sender.Close drain deregisters itself from the hub, so a long-lived hub doesn't pin abandoned receivers.

Chan support

Chan() returns a per-receiver private channel fed by a per-receiver goroutine. It yields pending events in first-touch key order, carrying the same gobus.Event values the Recv methods return, and repeated calls return the same channel.

for ev := range rx.Chan() {
    apply(ev.Key, ev.Value)
}

The channel is unbuffered on purpose: coalescing continues in the receiver's per-key slots while the consumer is busy, so a fast publisher produces no backlog beyond the live key set. One caveat — an event already handed to the feeder has left the receiver's slots, so a Send for that key while the feeder is parked on delivery enqueues the key afresh rather than coalescing into the in-flight event. (This is also why Peek reports ErrEmpty for an event in flight.)

The channel closes when the feeder observes receiver-close, or sender/hub-close with nothing left to drain. Abandoning the channel without calling Receiver.Close() pins the feeder goroutine — it parks forever waiting for the next event. Always Close the receiver when you stop reading.

Thread safety

Sender is safe to share across goroutines: Send and Close both serialize through the hub lock, and Send first reads a lock-free receiver count so it takes that lock only when a receiver is registered.

A Receiver is intended for a single consumer goroutine, and conflate relies on it: the receiver owns an insertion-ordered queue meant to be popped by one reader. Peek, TryRecv and TryRecvAll are safe to call from any goroutine but are only meaningful on that one, since a concurrent consumer can take the event between your two calls.

One mutex guards every receiver on the hub. Send fans a write across all of them under it, and each receiver pops from its own queue under the same lock.

API reference

Hub

func New[K comparable, V any](opts ...Option[V]) *Hub[K, V]
func WithDefaultMerge[V any](merge Merge[V]) Option[V]

Creates a hub whose receivers coalesce per key. Without options the newer value supersedes the undelivered one, which is latest-value-per-key. WithDefaultMerge sets a different hub-wide policy: the merge every receiver uses unless Hub.WithMerge gave it one of its own. New panics on a nil option, WithDefaultMerge on a nil Merge.

WithDefaultMerge is a package-level function, unlike the receiver options below, because it is built before the hub exists. It carries V alone, so a call site spells only K:

hub := conflate.New[string, Update]()                              // latest wins
hub := conflate.New[string](conflate.WithDefaultMerge(annihilating))

func (h *Hub[K, V]) Sender() *Sender[K, V]
func (h *Hub[K, V]) Receiver(opts ...ReceiverOption[K, V]) *Receiver[K, V]
func (h *Hub[K, V]) WithKey(k K) ReceiverOption[K, V]
func (h *Hub[K, V]) WithKeyFilter(keep func(K) bool) ReceiverOption[K, V]
func (h *Hub[K, V]) WithMerge(merge Merge[V]) ReceiverOption[K, V]
func (h *Hub[K, V]) Close()

Sender returns the singleton send-side handle; repeated calls return the same one. Receiver returns a fresh handle per subscriber. Both return pre-closed handles once the hub is closed. WithKeyFilter and WithMerge panic on a nil function; Receiver panics on a nil option. WithKey takes a value, so it has nothing to reject.

Merge

type Merge[V any] func(prev, next V) (merged V, keep bool)

Combines an undelivered pending value with a newly sent one for the same key. Invoked only when the key already has a pending slot. keep == false annihilates: the key is dropped from the queue and the slot alike. Called under the bus lock, so it must not call back into the hub.

Sender

func (tx *Sender[K, V]) Send(k K, v V) error
func (tx *Sender[K, V]) TrySend(k K, v V) error
func (tx *Sender[K, V]) SendContext(ctx context.Context, k K, v V) error
func (tx *Sender[K, V]) Close()

Send never blocks. TrySend is equivalent to it — there is no separate non-blocking path — and exists to satisfy gobus.Sender. All three return ErrClosed once the sender or hub is closed. conflate never returns ErrFull.

Receiver

func (rx *Receiver[K, V]) Recv() (gobus.Event[K, V], error)
func (rx *Receiver[K, V]) RecvContext(ctx context.Context) (gobus.Event[K, V], error)
func (rx *Receiver[K, V]) TryRecv() (gobus.Event[K, V], error)
func (rx *Receiver[K, V]) TryRecvAll() ([]gobus.Event[K, V], error)
func (rx *Receiver[K, V]) Peek() (gobus.Event[K, V], error)
func (rx *Receiver[K, V]) Chan() <-chan gobus.Event[K, V]
func (rx *Receiver[K, V]) Close()

Peek and TryRecvAll are concrete-*Receiver accessors, not part of the gobus.Receiver interface; everything else on this list implements it.

Errors

Error	Returned by	Means
gobus.ErrClosed	every send and receive path	The receiver or hub is closed, or the sender is closed and this receiver's queue has drained.
gobus.ErrEmpty	TryRecv, TryRecvAll, Peek	Nothing is pending. Not terminal.
ctx.Err()	SendContext, RecvContext	The context was cancelled. Not terminal, and consumes nothing.

ErrFull is never returned: there is no capacity argument, because coalescing bounds a receiver's buffer by the live key set rather than the write volume. There is no ErrLagged equivalent either — a receiver that falls behind does not lose values it could be told about, it collapses them, which is the contract rather than an error condition.

Examples

• examples/recv — the classic resource-watch pattern: a fast producer, slow subscribers, annihilation of a create/delete pair, and a graceful Sender.Close drain. go run ./conflate/examples/recv
• examples/chan — the same bus consumed through Chan() and select. go run ./conflate/examples/chan

Package docs on pkg.go.dev

Documentation

Overview

Package conflate provides a single-producer, multi-consumer keyed latest-value fan-out bus.

A Hub hands out a singleton Sender and any number of [Receiver]s. Where github.com/amorey/gochan/watch keeps one latest-value slot and github.com/amorey/gochan/broadcast keeps a fixed ring of every value (returning ErrLagged on overflow), conflate keeps the latest value *per key*: each receiver holds one slot per key plus an insertion-ordered queue, and a Sender.Send for a key that is already pending coalesces into that slot rather than appending. A slow receiver therefore never lags-as-loss — it catches up to the latest value of every key, in first-touch order, and its memory stays bounded by the live key set rather than by write volume.

Coalescing policy is a Merge function, deciding how an undelivered pending value combines with a newly sent one. By default the newer value wins, which is latest-value-per-key. WithDefaultMerge supplies another for a bus that must combine values rather than replace them, or annihilate the slot entirely (e.g. a create followed by a delete the consumer never observed).

Typical uses

Streaming resource state to watchers (Kubernetes-style informers), UI or dashboard update feeds where only the current state of each entity matters, cache invalidation fan-out, incremental index maintenance, and any "coalesce writes per entity, deliver at consumer pace" pipeline.

Semantics

Latest-value-per-key delivery. Each receiver owns an insertion-ordered queue of keys plus one value slot per key. A Send for a key with no pending slot appends the key at the back of the queue; a Send for a key that is already pending coalesces into the existing slot via Merge and leaves the key's queue position unchanged. Delivery order is therefore first-touch order, and a hot key does not starve a cold one by repeatedly jumping the queue.

Bounded by the key set, not the write volume. A thousand sends across four keys leave at most four pending entries. This is what makes conflate safe for an unbounded producer feeding a slow consumer: there is no capacity argument and no lag error because there is no unbounded backlog to grow.

Merge may annihilate. Returning keep == false drops the key entirely — queue entry and value slot both — so a create/delete pair the consumer never observed leaves no residue at all.

Send never blocks. Slow receivers cannot apply backpressure to the publisher; they simply coalesce more aggressively.

A send with no receivers is cheap. Sender.Send returns without taking the bus lock when no receiver is registered, so a hot producer does not pay for a hub nobody is watching. The result is the same one it always gave: nil, with nothing published. The corollary is a subscriber-side ordering requirement — register with Hub.Receiver before taking a snapshot of the producer's state, never after — because a value published in that gap reaches no receiver and conflate has no replay.

Per-receiver policy overrides, passed as options to Hub.Receiver. Hub.WithKey and Hub.WithKeyFilter filter keys at enqueue time, so a receiver interested in part of a high-cardinality producer stays bounded by the keys it actually wants; WithKey names one key and runs no caller code, WithKeyFilter takes a predicate. Hub.WithMerge lets a single consumer coalesce by its own policy without affecting the rest of the bus — necessary when consumers of the same producer disagree about what may be dropped, since one hub-wide merge cannot express that. The options compose. WithDefaultMerge sets the hub's fallback at construction; Hub.WithMerge overrides it for one receiver.

The backlog head is observable without consuming it. Receiver.Peek returns the oldest pending event and leaves it in place, under the same closed > value precedence the popping paths use.

The whole backlog is takeable as one cut. Receiver.TryRecvAll pops everything pending under a single acquisition of the bus lock, so a consumer that reads a burst as one unit gets a batch with defined membership — everything pending as of one instant — rather than whatever a loop of Receiver.TryRecv happened to observe across several.

Sender close drains. Sender.Close lets each receiver drain its pending values once before subsequent reads report gobus.ErrClosed. Hub.Close is hard tear-down with no drain.

A single Receiver is intended for one consumer goroutine.

Index

• type Hub
  • func New[K comparable, V any](opts ...Option[V]) *Hub[K, V]
  • func (h *Hub[K, V]) Close()
  • func (h *Hub[K, V]) Receiver(opts ...ReceiverOption[K, V]) *Receiver[K, V]
  • func (h *Hub[K, V]) Sender() *Sender[K, V]
  • func (h *Hub[K, V]) WithKey(k K) ReceiverOption[K, V]
  • func (h *Hub[K, V]) WithKeyFilter(keep func(K) bool) ReceiverOption[K, V]
  • func (h *Hub[K, V]) WithMerge(merge Merge[V]) ReceiverOption[K, V]
• type Merge
• type Option
  • func WithDefaultMerge[V any](merge Merge[V]) Option[V]
• type Receiver
  • func (rx *Receiver[K, V]) Chan() <-chan gobus.Event[K, V]
  • func (rx *Receiver[K, V]) Close()
  • func (rx *Receiver[K, V]) Peek() (gobus.Event[K, V], error)
  • func (rx *Receiver[K, V]) Recv() (gobus.Event[K, V], error)
  • func (rx *Receiver[K, V]) RecvContext(ctx context.Context) (gobus.Event[K, V], error)
  • func (rx *Receiver[K, V]) TryRecv() (gobus.Event[K, V], error)
  • func (rx *Receiver[K, V]) TryRecvAll() ([]gobus.Event[K, V], error)
• type ReceiverOption
• type Sender
  • func (tx *Sender[K, V]) Close()
  • func (tx *Sender[K, V]) Send(k K, v V) error
  • func (tx *Sender[K, V]) SendContext(ctx context.Context, k K, v V) error
  • func (tx *Sender[K, V]) TrySend(k K, v V) error

Types

type Hub

type Hub[K comparable, V any] struct {
	// contains filtered or unexported fields
}

Hub is the construction handle for a conflate pipeline.

func New

func New[K comparable, V any](opts ...Option[V]) *Hub[K, V]

New creates a hub whose receivers coalesce per key. Without options a Send for a key that is already pending replaces the undelivered value, which is latest-value-per-key. WithDefaultMerge sets a different hub-wide policy — one that combines values (summing deltas, unioning sets) or annihilates them.

Panics if any option is nil.

func (*Hub[K, V]) Close

func (h *Hub[K, V]) Close()

Close is hard tear-down: the sender and every live receiver are closed immediately, with no final drain of pending values. Use Sender.Close for the soft path. Future Hub.Receiver calls return pre-closed handles. Idempotent.

func (*Hub[K, V]) Receiver

func (h *Hub[K, V]) Receiver(opts ...ReceiverOption[K, V]) *Receiver[K, V]

Receiver returns a new receiver bound to the hub, configured by opts. A receiver starts empty: it observes values sent after it was created, not the producer's history. If the hub is already closed the receiver is pre-closed and reports gobus.ErrClosed on use.

Options are minted by the hub itself — Hub.WithKeyFilter and Hub.WithMerge — and compose freely:

rx := hub.Receiver()                           // every key, hub's merge
rx := hub.Receiver(hub.WithKey(k))             // one key, no predicate
rx := hub.Receiver(hub.WithKeyFilter(wanted))  // one key subset
rx := hub.Receiver(hub.WithMerge(stricter))    // own coalescing policy
rx := hub.Receiver(hub.WithKeyFilter(wanted), hub.WithMerge(stricter))

Later options win over earlier ones for the same setting.

func (*Hub[K, V]) Sender

func (h *Hub[K, V]) Sender() *Sender[K, V]

Sender returns the singleton send-side handle. Repeated calls return the same handle. After the hub has been closed the returned handle reports gobus.ErrClosed on use.

func (*Hub[K, V]) WithKey

func (h *Hub[K, V]) WithKey(k K) ReceiverOption[K, V]

WithKey returns an option restricting a receiver to the single key k. Other keys are dropped at Send time and never buffered, so the receiver's memory is bounded by that one key.

It is Hub.WithKeyFilter for the case that needs no predicate: the send path compares k rather than calling into caller code, so the fan-out runs no arbitrary function under the bus lock on this receiver's behalf. The zero K is a usable key.

WithKey and WithKeyFilter are the same setting, so the later of the two decides which keys the receiver takes.

func (*Hub[K, V]) WithKeyFilter

func (h *Hub[K, V]) WithKeyFilter(keep func(K) bool) ReceiverOption[K, V]

WithKeyFilter returns an option restricting a receiver to keys for which keep returns true; all other keys are dropped at Send time and never buffered. Filtering at enqueue keeps a selective receiver's memory bounded by the keys it actually wants rather than the producer's whole key space. keep is called under the bus lock, so it must not call back into the hub. Panics if keep is nil. For a single key, Hub.WithKey says the same thing without the call.

It is a method on Hub rather than a package-level function so that K and V are fixed by the hub: callers write no type arguments, and an option built for the wrong key type is a compile error.

func (*Hub[K, V]) WithMerge

func (h *Hub[K, V]) WithMerge(merge Merge[V]) ReceiverOption[K, V]

WithMerge returns an option making a receiver coalesce with its own merge instead of the hub's. This lets a single consumer apply a different policy — e.g. annihilating pending values others must retain — without affecting the rest of the bus. Use it when consumers of the same producer have genuinely different requirements about what may be dropped, which a single hub-wide merge cannot express. merge is called under the bus lock, like the hub's. Panics if merge is nil.

type Merge

type Merge[V any] func(prev, next V) (merged V, keep bool)

Merge combines an undelivered pending value with a newly sent value for the same key. It is invoked only when the key already has a pending (not yet delivered) slot. It returns the surviving value and whether to keep the slot; keep == false drops the key entirely (annihilation).

Merge is called under the bus lock, so it must not call back into the hub.

type Option

type Option[V any] func(*config[V])

Option configures a hub built by New. It is package-level, unlike the per-receiver options, because it is built before the hub exists.

It carries V alone, so a call site spells only K. A K-dependent option would force both type arguments at every such call site; don't add one without meaning to.

func WithDefaultMerge

func WithDefaultMerge[V any](merge Merge[V]) Option[V]

WithDefaultMerge sets the hub-wide coalescing policy: the merge every receiver uses unless Hub.WithMerge gave it one of its own. Without it the hub keeps the newer value, which is latest-value-per-key. Panics if merge is nil. See docs/adr/2026-08-22-conflate-default-merge.md.

type Receiver

type Receiver[K comparable, V any] struct {
	// contains filtered or unexported fields
}

Receiver is a receive-side handle, intended for one consumer goroutine. It holds an insertion-ordered key queue plus a per-key value slot; coalescing happens on Send into these structures, so a read is a plain pop under the lock.

func (*Receiver[K, V]) Chan

func (rx *Receiver[K, V]) Chan() <-chan gobus.Event[K, V]

Chan returns a per-receiver native channel that yields pending events in first-touch key order, carrying the same gobus.Event values the Recv methods return. The channel is unbuffered: coalescing continues to happen in the receiver's slots while the consumer is busy, so a fast publisher produces no backlog beyond the live key set. Repeated calls on the same receiver return the same channel.

Note that an event already handed to the feeder has left the receiver's slots, so a Send for that key while the feeder is parked on delivery enqueues the key afresh rather than coalescing into the in-flight event.

The channel is closed when the feeder observes receiver-close, or sender/hub-close with nothing left to drain. Abandoning the channel without calling Receiver.Close pins the feeder goroutine — it will park forever waiting for the next event. Always Close the receiver when you stop reading.

func (*Receiver[K, V]) Close

func (rx *Receiver[K, V]) Close()

Close closes this receiver only; other receivers and the sender are unaffected. Any pending values are abandoned and the Chan feeder, if started, shuts down and closes the channel. Idempotent.

func (*Receiver[K, V]) Peek

func (rx *Receiver[K, V]) Peek() (gobus.Event[K, V], error)

Peek returns the oldest pending event without removing it, so a subsequent Recv or TryRecv still returns it. It returns gobus.ErrEmpty if nothing is pending, or gobus.ErrClosed if the receiver or hub is closed (or the sender is closed and the pending values have drained) — the same precedence Receiver.TryRecv applies. Peek is TryRecv without the pop, not a raw read of the queue: a closed handle reports ErrClosed even with a value at the head. Note that ErrClosed is therefore not a statement that the backlog was empty: Hub.Close and Receiver.Close abandon whatever is still queued.

The returned value is the current merged contents of the head key's slot. A Send that coalesces into that slot between two Peeks changes what the second Peek reports but leaves the key's queue position — and so its identity as the head — unchanged. A Send whose Merge annihilates the head key removes it, so the next Peek reports a different key: the head key is stable under coalescing, not under annihilation.

Peek takes the hub lock, the same one that serializes the whole Send fan-out, so polling it in a loop slows every publisher and every other receiver on the hub. Call it once per unit of work, not as a spin. When the unit of work is a whole burst, Receiver.TryRecvAll is how to take it in one call rather than a loop.

Peek is safe to call from any goroutine, but like the rest of the receive side it is only meaningful on the receiver's single consuming goroutine: a concurrent Recv, TryRecv or Receiver.Chan feeder may take the peeked event before the caller can act on it. An event already handed to the Chan feeder has left the queue, so Peek reports ErrEmpty while it is in flight.

func (*Receiver[K, V]) Recv

func (rx *Receiver[K, V]) Recv() (gobus.Event[K, V], error)

Recv blocks until an event is pending, then pops and returns the oldest one: the key that was touched least recently, carrying its latest merged value. It returns gobus.ErrClosed once the receiver or hub is closed (with the sender's soft close, after the pending values have drained).

func (*Receiver[K, V]) RecvContext

func (rx *Receiver[K, V]) RecvContext(ctx context.Context) (gobus.Event[K, V], error)

RecvContext blocks like Recv but returns ctx.Err() if ctx is cancelled first. Cancellation does not close this receiver.

It implements the closed > cancelled > value precedence documented on gobus.Receiver — including that a cancelled ctx never consumes the event it declined, and that reaching ctx.Err() neither closes nor deregisters the receiver. What is conflate-specific is the cost of ignoring the latter: an abandoned handle keeps coalescing, so it holds one slot per live key for the hub's lifetime. `defer rx.Close()` covers it, as it does for any abandoned receiver.

To consume what is left before closing, call Receiver.TryRecvAll once — it takes the whole remaining queue, and its error reports which state the flush stopped in — or loop on Receiver.TryRecv until it reports any error. Then Close. The flush alone is not a substitute for the Close: against a still-open sender either form ends on ErrEmpty, which is not terminal and does not deregister — only a drain that reaches ErrClosed does that on its own.

func (*Receiver[K, V]) TryRecv

func (rx *Receiver[K, V]) TryRecv() (gobus.Event[K, V], error)

TryRecv pops the oldest pending event without blocking. It returns gobus.ErrEmpty if nothing is pending, or gobus.ErrClosed if the receiver or hub is closed (or the sender is closed and the pending values have drained).

func (*Receiver[K, V]) TryRecvAll

func (rx *Receiver[K, V]) TryRecvAll() ([]gobus.Event[K, V], error)

TryRecvAll pops every pending event without blocking and returns them in queue order, leaving the receiver empty. It returns gobus.ErrEmpty if nothing is pending, or gobus.ErrClosed if the receiver or hub is closed (or the sender is closed and the pending values have drained) — the same precedence Receiver.TryRecv applies, closed beating empty beating value.

Partial results are not a case: either it returns every pending event with a nil error, or no events and an error. There is never "some values and ErrClosed", so a caller may test the error and ignore the slice. Against Sender.Close, the soft path, the whole queue comes back with a nil error and only the next call is terminal. Note the corollary that ErrClosed is therefore not a statement that the backlog was empty: Hub.Close and Receiver.Close abandon whatever is still queued.

The whole queue is taken under one acquisition of the hub lock, and that atomicity is the contract rather than an optimization: the result is everything pending as of one instant. A loop of Receiver.TryRecv is a sequence of instants, so the batch it assembles has no defined membership — a Send landing between two iterations joins it and one landing just after the loop observes empty does not, and no caller can close that gap from outside the lock.

Two properties of the returned slice are worth stating, because a batch consumer will build on both. It is in **queue order**, which is first-touch order and carries no relation to any ordering quantity inside V — a caller that needs value order sorts what it gets. And it holds **one entry per key**, since a receiver's queue holds each key once, so it can be folded into a map or dispatched over per key without deduping. That second property belongs to a single call: a batch assembled from more than one receive can repeat a key, because an event that has already left the receiver's slots cannot be coalesced into and a later Send for it enqueues afresh.

The critical section is O(live keys) and runs no caller code — no Merge, no key filter — which is what makes holding the hub lock across the whole queue acceptable. Note the trade against a TryRecv loop: total acquisitions go down, but this one is held longer, so worst-case publisher latency goes up.

TryRecvAll is safe to call from any goroutine but, like the rest of the receive side, is only meaningful on the receiver's single consuming goroutine. An event already handed to the Receiver.Chan feeder has left the queue, so a cut does not include it.

type ReceiverOption

type ReceiverOption[K comparable, V any] func(*receiverConfig[K, V])

ReceiverOption configures a receiver minted by Hub.Receiver. Options are built by the hub's own Hub.WithKey, Hub.WithKeyFilter and Hub.WithMerge methods, which fix K and V from the hub — so option call sites need no type arguments and a mismatched option fails to compile rather than at run time.

The set of options is closed: the parameter type is unexported, so code outside this package cannot name it to write one of its own.

type Sender

type Sender[K comparable, V any] struct {
	// contains filtered or unexported fields
}

Sender is the singleton send-side handle. Safe to share across goroutines.

func (*Sender[K, V]) Close

func (tx *Sender[K, V]) Close()

Close closes the sender. Receivers drain their pending values once before subsequent reads return gobus.ErrClosed, and their Chan feeders close the channel after the same drain. Further Send calls return ErrClosed. Idempotent.

Safe to call concurrently with a Sender.Send or Sender.SendContext from another goroutine. The two serialize on the one state every send path reads without the bus lock — the poisoned live count — so a racing send resolves to exactly one of the two orderings: it publishes and returns nil, or it returns ErrClosed and publishes nothing. There is no third outcome and no partial one. *Which* ordering wins is unspecified, so a caller that needs a send to be visible before shutdown must order the two itself; a caller shutting down and not caring whether the last value lands need not fence anything. A send that wins the ordering is still drained by this soft close, so "publishes" means the value reaches its receivers, not merely that it was enqueued before a tear-down discarded it.

This is a promise about conflate specifically, and it holds because Send never parks: the whole of Close runs under s.mu, and the only step of a send that runs outside it is the atomic load Close poisons. Do not read it as a module-wide rule — Hub.Close keeps the close-versus-send discipline stated in its own doc, since it tears down receivers a send is fanning out to.

func (*Sender[K, V]) Send

func (tx *Sender[K, V]) Send(k K, v V) error

Send publishes v under key k to every receiver. Never blocks. For a receiver that already has k pending, the caller's Merge coalesces into that slot; otherwise k is appended at the back of the receiver's queue.

A Send to a hub with no live receiver does nothing and returns nil. That has always been the answer — there is nobody to fan out to — but it is now reached without taking the bus lock at all, so a hot producer pays nothing for a bus nobody is reading. Only the cost changes, never the result.

This makes one existing requirement load-bearing in a place a reader would not think to look. A subscriber must call Hub.Receiver *before* it takes its own snapshot of the producer's state, never after: a value published in the gap reaches no receiver, and conflate has no replay to recover it. The requirement is not new — a receiver has always observed only what was sent after it was created — but a publisher that skips the lock cannot notice a subscriber that arrives late, so getting the order wrong loses the value permanently rather than merely racily.

func (*Sender[K, V]) SendContext

func (tx *Sender[K, V]) SendContext(ctx context.Context, k K, v V) error

SendContext behaves like Send, but reports a cancelled ctx instead of publishing. Send never blocks, so ctx is consulted exactly once — there is no parked state in which a cancellation could arrive — and gates the call on nothing beyond reaching the point where the send is resolved.

That single check is made where the send is *resolved*, not on entry. A live ctx on a hub with no live receiver resolves at the lock-free receiver-count read, with nothing to publish and nothing to report. Every other send — including every cancelled one — resolves under s.mu, so that txClosed and ctxDone are read from one consistent view and closed > cancelled is derived rather than assembled from two reads taken at different moments. The rest of this comment is about that path.

So a ctx that was live when SendContext was called but is cancelled by the time this send reaches the front of the lock reports ctx.Err() and publishes nothing. This is deliberate: a caller passing a context is asking for the publish to be bounded by it, and the wait for s.mu is real work — the caller's own Merge and key filters run under that lock. Enqueueing a value on behalf of a context that has since expired, on the grounds that it was live a moment earlier, is the weaker answer. It also keeps every verdict in this package derived from state read at the decision point: txClosed and rx.done are re-read under the lock for the same reason, and recvLoop places its own cancellation check under s.mu rather than trusting an entry-time snapshot.

A cancellation racing this call may therefore land either side of the lock, and the two outcomes — published, or ctx.Err() — are both correct resolutions of that race; the caller cannot have been relying on which.

Precedence is closed > cancelled: a closed sender reports gobus.ErrClosed even for an already-cancelled ctx, since that is the durable answer and a retry with a fresh context would only return it again. A cancelled ctx on a live sender still reports ctx.Err().

Only ctx's Done channel is consulted under the bus lock; ctx.Err() is called after it is released, so a context implementation that locks cannot deadlock against another goroutine's Send or Close. See sendLocked.

func (*Sender[K, V]) TrySend

func (tx *Sender[K, V]) TrySend(k K, v V) error

TrySend is equivalent to Send for conflate: Send never blocks, so there is no separate non-blocking path. Provided to satisfy the common gobus.Sender interface.