watch

README

watch

Keyed state bus: one receiver, one key (or all of them), one slot — always the current value.

Part of gobus.

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

Contents

• Overview
• Quick start
• Registration is the snapshot
• Accept decides which value wins
• One key per receiver, or all of them
• Inspecting what is unread
• Publishing with no receivers
• Contexts and cancellation
• Close semantics
• Chan support
• Thread safety
• API reference
• Errors
• Examples

Overview

Watch is a single-producer, multi-consumer keyed state bus. Where conflate streams events, watch distributes the current value of a key: each Receiver holds one slot, so a slow consumer skips to the current value rather than replaying what it missed.

A receiver is created by Hub.Watch, which is also where its baseline comes from — Hub.WithBaseline passes the value the caller has just read — and Receiver.Close() is the matching unwatch. Hub.WatchAcross makes one that watches every key instead of one, still with a single slot.

Pick watch over conflate when a consumer follows a single object and only its current state matters, or when it reacts the same way to a change anywhere and wants one wake-up rather than one per key. Pick conflate when a consumer needs each key's own latest value, and for change streams where a create-then-delete pair must leave no residue.

Coming from gochan/watch? Two rules differ. Registration here does snapshot (gochan's deliberately does not), and the hub holds no seed of its own — the baseline is per receiver, supplied by the caller. Don't carry the sister package's rule across.

Quick start

hub := watch.New[ObjectID](watch.WithAccept(func(prev, next Stamped) bool {
    return next.Seq > prev.Seq
}))
defer hub.Close()

// Read your state and register in ONE critical section: Watch calls no
// caller code, so it is safe under your own lock.
q.mu.Lock()
cur := q.current(id)
rx := hub.Watch(id, hub.WithBaseline(cur))
q.mu.Unlock()
defer rx.Close()

for ev := range rx.Chan() {
    use(ev.Value)
}

Note the type arguments: watch.New[ObjectID] spells only K, because Option[V] carries V alone and WithAccept infers it from its argument.

Registration is the snapshot

WithBaseline takes the value you have just read and never hands it back — it is the baseline, not a delivery. It is the prev of the first Accept call, and a receiver reads a value only once a Send supersedes it.

Omit it and the receiver has no baseline: it has read nothing and holds nothing, so its first value is taken whatever it is, without consulting Accept. There is no prev to pass, and the zero V would be a value you never held — seeding it silently changes which first value wins. Accept governs every value after that.

rx := hub.Watch(id)                          // any current value will do
rx := hub.Watch(id, hub.WithBaseline(cur))   // measure against what I just read

The baseline is per receiver, not per hub, because each consumer reads at its own instant. WithAccept — the rule those values are judged by — is hub-wide for the opposite reason.

Because Watch calls no caller code, you can read your state and register in one critical section, which removes the register-before-read ordering rule conflate imposes. Nothing published in between can be lost, because there is no "in between".

One consequence to expect: two subscribers registering at different moments disagree about whether the same publish is news. A value can be new for a receiver that subscribed early and stale for one that subscribed late — which is exactly right, since each already holds what it read for itself.

Accept decides which value wins

Instead of a Merge, watch takes an optional Accept func(prev, next V) bool at hub construction:

hub := watch.New[ObjectID](watch.WithAccept(func(prev, next Stamped) bool {
    return next.Seq > prev.Seq
}))

It runs under the bus lock, once for each receiver watching the key, against that receiver's own current value. This is what makes the settled value independent of which of two concurrent Send calls takes the lock first, provided your rule is a strict order over V — a producer that computes a change under its own lock and publishes after releasing it can reach Send in the reverse of the order in which the changes became true, and Accept is what resolves that.

Without the option every value is accepted, which is last-writer-wins. The option may be omitted — a state bus has a meaningful identity rule, so omitting it is a statement rather than an oversight — but passing a nil Accept, or a nil Option to New, panics.

Accept is caller code running under the bus lock. It must not call back into the hub, and it must not take any lock a caller may hold while calling Watch, Send or Close — Watch is expressly safe under a producer's lock, so an Accept that takes that same lock inverts the two orders and deadlocks. Reading its two arguments and nothing else is always safe.

A rejected value changes nothing and the receiver is not told. A panic out of Accept leaves a partial fan-out: the receivers already reached keep the value, the rest are untouched, the send is not retried, and the hub stays usable.

One key per receiver, or all of them

There is no Unwatch and no key set: the constraint is structural, which is what removes the questions a mutable key set raises. A consumer watching N particular keys therefore holds N receivers and, if it uses Chan(), N goroutines.

Send for a key nobody watches is dropped, and nothing is retained: there is no receiver and therefore no buffer, so a later Watch never sees it. Once the last receiver for a key goes — by Close or by reaching a terminal ErrClosed — the hub releases the key entirely, so a key costs nothing after its last watcher.

WatchAcross

Hub.WatchAcross() is the one alternative: a receiver that watches every key, including keys nobody has published under yet and keys the consumer cannot name.

rx := hub.WatchAcross()             // no key argument: there is no key to name
defer rx.Close()

for {
    ev, err := rx.Recv()            // ev.Key names the key ev.Value came from
    if err != nil {
        return                    // ErrClosed
    }
    resync(ev.Key)
}

A wildcard subscription in the MQTT/NATS sense is the closest model most callers arrive with, and it differs in the two ways you are most likely to assume rather than check:

• It does not deliver every matching value. One slot means a value landing while an earlier one is unread replaces it. You read the current value, never the sequence.
• There is no pattern language. Every key or nothing — no prefix, glob or hierarchy, and nowhere to pass one. K is only comparable, so it carries no structure to match against and no later release can add one.

It is not a cheap way to subscribe to many keys. It holds one slot like every other receiver, so a burst across many keys collapses to a single pending value naming the last key to land — a hundred sends across fifty keys is one wake-up, not fifty, and the ninety-nine earlier values are never handed back. That collapse is the point: it serves a consumer whose reaction to any change is the same, typically "something moved, go re-read the store", and a wake per key would be pure waste to it. A consumer that needs each key's own latest value wants one Watch per key, or conflate, which keeps a slot per key and has the annihilation a create-then-delete pair needs.

Everything else matches Watch. It takes the same options: without one the first value is taken unjudged, and a WithBaseline value is the caller's own baseline and is never delivered back — a wildcard baseline being a prior value with no key attached, since you read it before knowing which key would move next; registration calls no caller code, so it is safe under your own lock; Accept is evaluated against this receiver's own slot; and all three Close methods behave identically. Event.Key names the key the slot's value was published under, assigned when the value lands — a value your Accept rejects moves neither the value nor the key.

Two structural differences worth knowing. A wildcard receiver holds no key against the hub, so it does not keep per-key state alive: a key still costs nothing once its last Watch receiver has gone. And with one registered the hub has no unwatched key, so no Send is dropped for want of a receiver.

Inspecting what is unread

Peek() returns the value a receive would hand back and leaves it unread — TryRecv minus the take, sharing its precedence exactly: ErrEmpty when nothing has superseded what this receiver has already seen, ErrClosed when the receiver or hub is closed or the sender has closed and the final value has been taken.

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

It reports what is unread, not the key's current state. A caught-up receiver gets ErrEmpty even though its slot holds a perfectly good value, and a closed handle gets ErrClosed even with a value waiting. If you want the current state on demand, keep your own copy of the last value read — the reading goroutine already has it, and it costs no lock.

Between two Peeks the value is not fixed: a Send your Accept takes replaces the slot, so the second Peek reports the newer value and the older one is never handed back by either path. That is the same skip-ahead every read on this bus is subject to, only visible without consuming. For a Watch receiver the key is fixed, since it watches one key for life; for a WatchAcross receiver the key travels with the value, so a replacing value can change it too.

Two cautions, as on conflate. 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. And while it is safe to call from any goroutine, it is only meaningful on the receiver's single consuming goroutine. Unlike conflate's Peek, a value already handed to the Chan feeder is still visible here: the feeder marks it read only once the consumer has taken it.

Publishing with no receivers

Send on a hub with no receiver at all returns nil without taking the bus lock, exactly as conflate does — the hub-wide lock is pure cost when there is nobody to fan out to. The result is unchanged, only the cost. TrySend and SendContext take the same path, a cancelled ctx is still reported rather than swallowed, and a closed sender still returns ErrClosed.

The subscriber-side ordering rule this forces on conflate does not apply here: Watch takes your snapshot as its argument, so there is no gap between reading state and registering for it to fall into. That is the point of registration being the snapshot.

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 watch-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 Accept runs under it.
• RecvContext returns ctx.Err() even when an unread value is waiting, and leaves that value unread. 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 holds its key against the hub for the hub's lifetime. defer rx.Close() covers it.
• To consume what is left first, loop on TryRecv until it returns any error. That 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. A receiver holding an unread value reads it once more, then sees ErrClosed; a caught-up receiver sees ErrClosed at once.
Receiver.Close()	The unwatch. This handle only: any unread value is discarded and the key is dropped from the hub once no other receiver watches it.
Hub.Close()	Hard tear-down: sender plus every live receiver, with no drain. Future Hub.Watch() 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 watch's Send never parks; it is a promise about this package, not a module-wide rule.

A receiver's slot holds one value, so Sender.Close() drains at most one value per receiver. Hub.Watch after a Sender.Close returns a live handle that holds nothing unread, so its first read is terminal — only Hub.Close returns pre-closed handles.

A receiver that reaches a terminal ErrClosed deregisters itself, releasing its key, so a long-lived hub pins neither abandoned receivers nor their keys by either exit path.

"No drain" on Hub.Close is a statement about the reading methods, which report ErrClosed at once. A Chan consumer can still receive one value after Close returns; see below.

Chan support

Chan() returns a per-receiver private channel fed by a per-receiver goroutine, carrying the same gobus.Event values the Recv methods return. Repeated calls return the same channel.

for ev := range rx.Chan() {
    use(ev.Value)
}

It is unbuffered, so a fast publisher builds no backlog: while the consumer is not reading, further sends only update the slot. The feeder marks a value read only once the consumer has taken it, so a newer value arriving mid-delivery makes the feeder re-snapshot rather than hand over the superseded one — which is what keeps a Chan consumer on the same latest-value footing as a Recv caller. (It is also why Peek still sees a value in flight, unlike on conflate.)

That is a latency property, not a guarantee. Once the feeder has committed to a delivery, anything making its select's other arms ready races that delivery, and Go chooses between ready arms at random. Two consequences:

• A superseded value is sometimes delivered, with the newer one immediately behind it.
• A Receiver.Close or Hub.Close can lose the race too, so one value can still be received after either returns, even though both abandon what is unread. The channel closes immediately after.

What holds is that values arrive in order and that a consumer which keeps reading converges on the current value.

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. 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. Send then touches only the receivers watching its key.

A Receiver is intended for a single consumer goroutine, but watch treats that as intent rather than invariant: a receiver using Chan() genuinely has two readers (the feeder and any direct TryRecv), so its read position lives under the hub lock rather than in the reading goroutine. Peek and TryRecv are safe from any goroutine; they are only meaningful on the consuming one, since a concurrent reader can take the value between your two calls.

Watch calls no caller code and is safe to call while holding your own lock. Accept is the constraint that keeps that true — see above.

API reference

Hub

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

New panics if any option is nil; WithAccept panics if fn is nil. The option is package-level rather than a method on the hub because it has to be built before the hub exists, and it carries V alone so a call site spells only K.

func (h *Hub[K, V]) Sender() *Sender[K, V]
func (h *Hub[K, V]) Watch(k K, opts ...WatchOption[K, V]) *Receiver[K, V]
func (h *Hub[K, V]) WatchAcross(opts ...WatchOption[K, V]) *Receiver[K, V]
func (h *Hub[K, V]) WithBaseline(cur V) WatchOption[K, V]
func (h *Hub[K, V]) Close()

Sender returns the singleton send-side handle; repeated calls return the same one. Watch makes a receiver for k; after Hub.Close the returned handle is pre-closed, and after Sender.Close it is live but holds nothing unread.

WithBaseline seeds the receiver's slot with the value the caller has just read, making it the prev of the first Accept. Without it the receiver holds nothing and takes its first value unjudged. The zero V is a usable baseline. Both constructors panic on a nil option; WithBaseline takes a value, so it has nothing to reject.

WatchAcross makes a receiver for every key, taking the same options. It holds one slot like any other receiver — the latest value published under any key, and the key it came from — so a burst across many keys leaves one pending value. See WatchAcross.

Accept

type Accept[V any] func(prev, next V) bool

Reports whether next replaces prev in a receiver's slot. Evaluated per receiver under the bus lock, against that receiver's own current value.

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. watch 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]) Peek() (gobus.Event[K, V], error)
func (rx *Receiver[K, V]) Chan() <-chan gobus.Event[K, V]
func (rx *Receiver[K, V]) Close()

Peek is a concrete-*Receiver accessor, not part of the gobus.Receiver interface; everything else on this list implements it. There is no TryRecvAll — a receiver holds one slot, so there is no backlog to take.

Errors

Error	Returned by	Means
gobus.ErrClosed	every send and receive path	The receiver or hub is closed, or the sender is closed and the final value has been taken.
gobus.ErrEmpty	TryRecv, Peek	Nothing has superseded what this receiver has already seen. Not terminal.
ctx.Err()	SendContext, RecvContext	The context was cancelled. Not terminal, and consumes nothing.

ErrFull is never returned: a receiver holds one slot, which the next accepted value overwrites, so there is no capacity to exhaust. There is no ErrLagged equivalent either — skipping to the current value is the contract rather than an error condition.

Examples

• examples/recv — a scheduler publishing job state outside its own lock, with Accept resolving the resulting reordering by sequence number, and a graceful Sender.Close. go run ./watch/examples/recv
• examples/chan — the same bus consumed through Chan() and select, showing values skipping forward under a fast producer and two different shutdown paths. go run ./watch/examples/chan

Package docs on pkg.go.dev

Documentation

Overview

Package watch provides a keyed latest-value state bus.

A Hub hands out a singleton Sender and any number of [Receiver]s. A receiver watches exactly one key, made by Hub.Watch, or every key, made by Hub.WatchAcross; Receiver.Close is the matching unwatch for either. A Sender.Send for a key reaches every receiver watching it, and a receiver that falls behind skips to the current value rather than replaying what it missed.

Registration is the snapshot

Hub.WithBaseline takes the value the caller has just read, and that value is the baseline every later value is measured against. It is per receiver, since each consumer reads at its own instant. This is the opposite of github.com/amorey/gochan/watch, whose hub holds one seed and whose registration deliberately does *not* snapshot. A reader arriving from the sister package must not carry that rule across.

The bus does not deliver the baseline back: it is the caller's own argument, and a receiver reads a value only once a Sender.Send supersedes it. Receiver.Peek shows that unread value without taking it, under the same closed > value precedence the taking paths use — so it too reports nothing for a receiver still on its baseline.

A receiver registered without a baseline has read nothing and holds nothing, so its first value is taken whatever it is: there is no prev to give Accept, and the zero V would be a value the caller never held. Accept runs on every value after that. Omit the baseline when any current value will do, supply one when the consumer already knows the state it is improving on.

One key for each receiver, or all of them

Hub.Watch binds a receiver to its key for life. There is no Unwatch and no mutable key set — the constraint is structural, so a consumer watching N particular keys holds N receivers and, if it uses Receiver.Chan, N goroutines.

Hub.WatchAcross is the one alternative: a receiver that watches every key, including keys nobody has published under yet and keys the consumer cannot name. It is not a way to subscribe to many keys cheaply — it still holds one slot, so a burst across many keys collapses to a single pending value naming the last key to land. That collapse is the point. It serves a consumer whose reaction to any change is the same, typically "go re-read the store", and it serves it in one wake-up rather than one per key.

A consumer that needs each key's own latest value, or the annihilation a create-then-delete pair needs, wants github.com/amorey/gobus/conflate, which keeps a slot per key and filters at enqueue.

Index

• type Accept
• 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]) Sender() *Sender[K, V]
  • func (h *Hub[K, V]) Watch(k K, opts ...WatchOption[K, V]) *Receiver[K, V]
  • func (h *Hub[K, V]) WatchAcross(opts ...WatchOption[K, V]) *Receiver[K, V]
  • func (h *Hub[K, V]) WithBaseline(cur V) WatchOption[K, V]
• type Option
  • func WithAccept[V any](fn Accept[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)
• 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
• type WatchOption

Types

type Accept

type Accept[V any] func(prev, next V) bool

Accept reports whether next replaces prev in a receiver's slot. It is the caller's rule for which of two values wins.

Accept runs under the bus lock, once for each receiver watching the key, with that receiver's own current value as prev. A receiver holding nothing — no Hub.WithBaseline and no value yet — takes its first value without consulting Accept, since there is no prev to pass. It must not call back into the hub, and it must not take any lock a caller may hold while calling Hub.Watch, Sender.Send or any Close — Watch is expressly safe to call under a producer's lock, so an Accept that takes that same lock inverts the two orders and deadlocks. Reading its two arguments and nothing else is always safe.

A panic out of Accept leaves a partial fan-out: the receivers already reached keep the value, the rest are untouched, the send is not retried and the hub stays usable.

type Hub

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

Hub is the construction handle for a watch pipeline.

func New

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

New creates a hub. 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. Use Sender.Close for the soft path. Future Hub.Watch calls return pre-closed handles. Idempotent.

"No drain" is a statement about the reading methods, which report gobus.ErrClosed at once. A Receiver.Chan consumer can still receive one value after Close returns: if the feeder had already committed to a delivery the close makes both arms of its select ready, and Go picks between ready arms at random. The channel closes immediately after. See Receiver.Chan.

Unlike Sender.Close, this one keeps the close-versus-send discipline: do not call it concurrently with an active Send from another goroutine. It tears down the receivers a send fans out to, so a racing send can deliver a value into a receiver that is being closed — the value is not lost racily so much as delivered to a handle that will never be read again, which is a harder thing for a caller to reason about than the sender-close case.

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 is closed it reports gobus.ErrClosed on use.

func (*Hub[K, V]) Watch

func (h *Hub[K, V]) Watch(k K, opts ...WatchOption[K, V]) *Receiver[K, V]

Watch makes a receiver for k. The receiver watches k for its whole life; Receiver.Close is the unwatch.

Without options the receiver has no baseline and its first value is taken whatever it is — there is nothing yet for Accept to judge it against, and nothing the consumer has read that it could fail to improve on. Accept runs on every offer after that.

Hub.WithBaseline supplies the value the caller has just read, making it the prev of the first Accept. It is a baseline, not a delivery: it is never handed back through a receive, so a receiver given one reads a value only once a Sender.Send supersedes it.

Watch calls no caller code, so it is safe to call while holding the producer's own lock — which is how a subscriber reads its state and registers in one critical section, with no value lost in between. See Accept for the rule an Accept must obey to keep that safe.

Panics if any option is nil. After Hub.Close the returned handle is pre-closed. After Sender.Close it is live but holds nothing unread, so its first read is terminal.

func (*Hub[K, V]) WatchAcross

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

WatchAcross makes a receiver watching every key — every key the hub carries now and every key it ever will — holding one slot, like every other receiver on this bus. That slot is the latest value published under *any* key, plus the key it came from. It takes the same options as Hub.Watch.

A wildcard subscription in the MQTT or NATS sense is the closest model most callers arrive with, and it differs in the two ways most likely to be assumed rather than checked:

• It does not deliver every matching value. One slot means a value that lands while an earlier one is still unread replaces it, so what a consumer reads is the current value, never the sequence. A burst across fifty keys is one wake-up, not fifty. If that loses information the consumer needs, this is the wrong method — see below.
• There is no pattern language. It matches every key or nothing; there is no prefix, glob or hierarchy, and there is nowhere to pass one. K is merely comparable, so it carries no structure to match against and no later release can add one.

One slot is the contract, not an artifact of how the slot is written. A burst across many keys leaves exactly one value pending, so a consumer whose whole reaction is "something changed, go re-read the store" wakes once rather than once per key. A consumer that needs each key's own latest value wants one Hub.Watch per key, or github.com/amorey/gobus/conflate — which keeps a slot per key and has the annihilation a create-then-delete pair needs.

gobus.Event.Key names the key the slot's value was published under, assigned when a value lands. A value the hub's Accept rejects changes neither the value nor the key, since the slot still holds what it held before. A receiver still on its baseline has no key at all — it has read nothing — which is unobservable, because every read reports gobus.ErrEmpty until a value lands.

Everything else matches Hub.Watch: without options the first value is taken unjudged, a Hub.WithBaseline value is never delivered back, no caller code runs during registration, and the close behavior of all three Close methods is the same. A wildcard baseline is a prior value with no key attached, since the caller read it before knowing which key would move next. It differs in holding no key against the hub — a wildcard receiver keeps no per-key state alive, so a key still costs nothing once its last Hub.Watch receiver has gone.

Panics if any option is nil.

func (*Hub[K, V]) WithBaseline

func (h *Hub[K, V]) WithBaseline(cur V) WatchOption[K, V]

WithBaseline makes cur the receiver's starting value: the prev of its first Accept, and never a delivery. Use it when the caller has just read the current state and wants the bus to measure against that read rather than take the next value on trust. The zero V is a usable baseline.

It is per receiver, not per hub, because each consumer's baseline is the value it read at its own instant. WithAccept, the rule those values are judged by, is hub-wide for the opposite reason. See docs/adr/2026-08-22-watch-optional-baseline.md.

type Option

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

Option configures a hub built by New.

It carries V alone, not K, so WithAccept infers V from its argument and a call site passing one spells only K. Adding a K-dependent option would force both type arguments at every such call site; do not add one without meaning to. This is also why the option is a package-level function rather than a method on the hub, as conflate's per-receiver options must be: those configure a handle whose hub has already fixed both types, while this one has to be built before the hub exists.

func WithAccept

func WithAccept[V any](fn Accept[V]) Option[V]

WithAccept sets the rule deciding whether a value replaces the one in a receiver's slot. Without it every value is accepted, which is last-writer-wins. Panics if fn is nil.

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 watches exactly one key, or every key when minted by Hub.WatchAcross. Either way it holds a single slot rather than a queue: a value that Accept takes overwrites the one before it, so a slow reader skips to the current value instead of building a backlog.

func (*Receiver[K, V]) Chan

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

Chan returns a per-receiver channel yielding values as they become current. It is unbuffered, so a fast publisher builds no backlog: while the consumer is not reading, further sends only update the slot, and the value waiting to be delivered is replaced by the current one. Repeated calls return the same channel.

The channel closes when the feeder observes receiver-close, or sender/hub-close with nothing left to drain.

Reading it is not a guarantee that every value read is current at the moment it is read. Once the feeder has committed to a delivery, anything that makes its select's other arms ready is racing that delivery, and Go chooses between ready arms at random. Two consequences:

• A newer value arriving can lose the race, so a superseded value is sometimes delivered, with the newer one immediately behind it. Values still arrive in order, and a consumer that keeps reading converges on the current value.
• A Receiver.Close or Hub.Close can lose it too, so one value can still be received after either returns, even though both are documented as abandoning what is unread. The channel closes immediately after.

Abandoning the channel without calling Receiver.Close pins the feeder goroutine — it parks forever waiting for the next value. Always Close the receiver when you stop reading.

func (*Receiver[K, V]) Close

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

Close is the unwatch: it closes this handle, discards any unread value and drops the key from the hub once no other receiver watches it. Other receivers and the sender are unaffected. Idempotent.

A Receiver.Chan consumer can still receive one value after Close returns; see Chan for why.

func (*Receiver[K, V]) Peek

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

Peek returns the value a receive would hand back, without taking it: a subsequent Receiver.Recv or Receiver.TryRecv still returns it. It is TryRecv minus the take, and shares its precedence exactly — gobus.ErrClosed if the receiver or hub is closed, or the sender is closed and the final value has been taken; gobus.ErrEmpty if nothing has superseded what this receiver has already seen.

It is therefore *not* a read of the key's current state: a receiver that has caught up reports ErrEmpty even though its slot holds a perfectly good value, and a closed handle reports ErrClosed with one waiting. Keep your own copy of the last value read if you need the current state on demand — that is what the reading goroutine already has, and it costs no lock.

Between two Peeks the value is not fixed: a Sender.Send this receiver's Accept takes replaces the slot, so the second Peek reports the newer value and the older one is never handed back by either path. That is the same skip-ahead every read on this bus is subject to, only visible without consuming. For a receiver from Hub.Watch the key *is* fixed, since it watches one key for its whole life; for one from Hub.WatchAcross the key travels with the value, so a replacing value can change it too.

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.

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 value before the caller can act on it. Unlike conflate's Peek, a value already handed to the feeder is still visible here — the feeder marks it read only once the consumer has taken it, so Peek reports the value in flight rather than ErrEmpty.

func (*Receiver[K, V]) Recv

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

Recv blocks until a value this receiver has not taken is available, then returns it. It returns gobus.ErrClosed once the receiver or hub is closed, or once the sender is closed and the final value has been taken.

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 read never consumes the value it declined, and that reaching ctx.Err() neither closes nor deregisters the receiver. What is watch-specific is the cost of ignoring the latter: an abandoned handle holds its key against the hub for the hub's lifetime. `defer rx.Close()` covers it, as it does for any abandoned receiver.

func (*Receiver[K, V]) TryRecv

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

TryRecv returns the current value if this receiver has not taken it, gobus.ErrEmpty if nothing has changed since it subscribed, or gobus.ErrClosed if the receiver or hub is closed, or the sender is closed and the final value has been taken.

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. A receiver holding an unread value reads it once more before subsequent reads return gobus.ErrClosed; a receiver already caught up sees ErrClosed at once. Further sends 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.

This is a promise about watch 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 as the value of k to every receiver watching k, and to every receiver from Hub.WatchAcross. Never blocks. A Send for a key nobody watches is discarded: there is no receiver and therefore no buffer, so a later Hub.Watch never sees it. A wildcard receiver is a watcher of every key, so a hub with one has no unwatched key.

For each watching receiver the hub's Accept decides whether v replaces that receiver's current value. A rejected value changes nothing, and the receiver is not told. Because Accept is evaluated per receiver against that receiver's own slot, one value can be new for a receiver that subscribed early and stale for one that subscribed late.

Returns gobus.ErrClosed if the sender or hub has been closed.

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 once — at the point the send is resolved, under the bus lock, rather than on entry. A cancellation landing while the call waits for that lock is therefore honoured, and nothing is published for a ctx that has since expired.

Precedence is closed > cancelled: a sender already closed 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.

Only ctx's Done channel is read 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: Send never blocks, so there is no separate non-blocking path. Provided to satisfy gobus.Sender.

type WatchOption

type WatchOption[K comparable, V any] func(*watchConfig[V])

WatchOption configures a receiver minted by Hub.Watch or Hub.WatchAcross. Options are built by the hub's own Hub.WithBaseline method, which fixes 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.