Deprecated: moved into github.com/kitsunium/sdk/internal (ADR 0171), which no consumer imports; update with: go get github.com/kitsunium/sdk/pkg/v1/<family>@latest for each family you import && go mod tidy
README
¶
topic
Typed in-process broadcast Topic[T] for the SDK kernel — stdlib-only, no
domain vocabulary. Every published value reaches every current subscriber.
var updates topic.Topic[Position]
sub := updates.Subscribe(topic.DropOldest(64)) // the policy is NOT optional
defer sub.Unsubscribe()
go func() {
for {
select {
case p := <-sub.Values():
render(p)
case <-sub.Done(): // the value channel is never closed; this is the end signal
return
}
}
}()
updates.Publish(ctx, p) // returns how many subscribers took it
The zero Topic is ready to use, as sync.Mutex is.
The delivery policy has no default, on purpose. Block(n) couples the
producer to the subscriber (bounded by Publish's context); DropOldest(n)
keeps the freshest window; DropNewest(n) keeps the earliest one. Both drop
policies count what they discard, readable through sub.Dropped() — so a
slow subscriber never stalls the producer, and never loses data, in silence. A
zero DeliveryConfig is refused at the call that omitted it (ADR 0031).
Unsubscribing during a live fan-out is safe and costs no other subscriber
its copy: Publish delivers outside every lock, and the value channel is never
closed, so a "send on closed channel" panic in the publisher is unreachable
rather than unlikely. The price is that ranging over sub.Values() never
ends — select on sub.Done() alongside it.
Publish allocates nothing (see BENCH.md); the cost sits on membership churn
instead. See CLAUDE.md for the design.
Documentation
¶
Overview ¶
Package topic — the per-subscriber backpressure decision, made at the call site and impossible to leave unmade.
Package topic — one subscriber's handle, and the delivery paths behind it.
Package topic — one published version of a Topic's membership.
Package topic provides Topic[T], a typed in-process broadcast: every published value reaches every current subscriber. It is a kernel primitive — stdlib-only (plus the kernel's own copy-on-write snapshot), and domain-neutral: Topic, Publish, Subscribe, and a T, with no Event, no Message and no Handler in a signature.
A slow subscriber never stalls the producer in silence ¶
This is the property that makes or breaks a broadcast primitive, and it is why there is no default delivery mode. Every Topic.Subscribe names one, at the call site, by building a DeliveryConfig with Block, DropOldest or DropNewest:
- Block couples the producer to this subscriber on purpose. Publish waits, bounded by the caller's own context, and returns a delivered count below Topic.Subscribers when the value did not land.
- DropOldest keeps the freshest window: the stale value is discarded to make room, and counted in Listener.Dropped.
- DropNewest keeps the earliest window: the arriving value is discarded, and counted.
A DeliveryConfig has no usable zero value — ADR 0031, applied by making the bad state unspellable rather than merely checked. Whatever a zero meant it would be a backpressure decision the caller never made, and the two candidates are the exact failures this primitive exists to prevent: a silent stall or a silent loss.
Silence is closed off from both ends. With a drop policy the loss is COUNTED and readable; with Block the coupling is written at the call site and bounded by Publish's context.
Leaving during a fan-out ¶
Listener.Unsubscribe is safe at any moment, including while a Publish is delivering to that very subscriber, and it never costs another subscriber its copy. Two decisions buy that:
- Publish loads the subscriber list from a copy-on-write snapshot and delivers OUTSIDE any lock, so an unsubscribe never waits on a delivery — including the delivery that only the unsubscribe could have unblocked.
- The value channel is NEVER closed. Unsubscribe closes a separate done channel, which releases a publisher parked on a Block delivery. Closing the value channel instead would put "send on closed channel" one race away, and that panic lands in the PUBLISHER — code that did nothing wrong. Here it is unreachable rather than unlikely.
The cost of that choice is stated rather than hidden: ranging over Listener.Values never ends. Select on Listener.Done alongside it, exactly as with a context.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DeliveryConfig ¶
type DeliveryConfig struct {
// contains filtered or unexported fields
}
DeliveryConfig is the answer to one question, asked at every Topic.Subscribe: what happens when this subscriber is not keeping up?
There is no usable zero value, and that is the design. Its fields are unexported so a DeliveryConfig{} literal cannot be written, and a `var p DeliveryConfig` is refused by Subscribe rather than silently picking a backpressure policy on the caller's behalf — ADR 0031's "a zero value is a safe default or an explicit refusal, never an inert policy", applied where no safe default exists because either candidate is a documented failure mode.
Build one with Block, DropOldest or DropNewest.
func Block ¶
func Block(capacity int) DeliveryConfig
Block makes a full subscriber buffer stall the publisher until room appears.
It is the honest choice when every value matters more than the producer's latency — and it is a real coupling: one subscriber that stops reading holds up the fan-out to everyone visited after it. Two things keep that from being a hang: Topic.Publish takes a context that bounds the wait, and a subscriber that unsubscribes releases the publisher immediately.
capacity is the buffer depth; anything below 1 is clamped to 1.
func DropNewest ¶
func DropNewest(capacity int) DeliveryConfig
DropNewest keeps the EARLIEST window: when the buffer is full the arriving value is discarded.
It is the choice for a log or an audit trail, where the first N records describe how a burst began and the last N describe only that it continued. Every discard is counted in Listener.Dropped.
capacity is the buffer depth; anything below 1 is clamped to 1.
func DropOldest ¶
func DropOldest(capacity int) DeliveryConfig
DropOldest keeps the FRESHEST window: when the buffer is full the stalest value is discarded to make room for the arriving one.
It is the choice for state — a metric, a health status, a position — where an old value is worse than no value. Every discard is counted in Listener.Dropped, which is what makes the loss visible rather than silent.
capacity is the buffer depth; anything below 1 is clamped to 1.
type Listener ¶
type Listener[T any] struct { // contains filtered or unexported fields }
Listener is one subscriber's handle on a Topic.
Obtain one from Topic.Subscribe; the zero value is not usable and cannot be made so. Safe for concurrent use.
func (*Listener[T]) Done ¶
func (s *Listener[T]) Done() <-chan struct{}
Done closes when this subscriber leaves — by its own Unsubscribe, or because the Topic closed. It is the termination signal the value channel deliberately does not carry.
func (*Listener[T]) Dropped ¶
Dropped reports how many values never reached this subscriber: discarded by a drop policy, or abandoned because Publish's context expired while a Block delivery waited.
It does NOT count values a departed subscriber missed — leaving is not falling behind, and conflating the two would make a clean shutdown look like a capacity problem.
Monotonic, and a point-in-time reading.
func (*Listener[T]) Unsubscribe ¶
func (s *Listener[T]) Unsubscribe()
Unsubscribe removes this subscriber from its topic. It is idempotent, safe to defer, and safe to call from inside the goroutine reading Listener.Values — including while a publisher is mid-delivery to it.
Values already delivered remain readable afterwards; the channel is not closed and not drained.
func (*Listener[T]) Values ¶
func (s *Listener[T]) Values() <-chan T
Values is the channel this subscriber reads.
It is never closed, so ranging over it never ends. Select on it together with Listener.Done (or with a context), which is the shape this primitive is built for.
type Topic ¶
type Topic[T any] struct { // contains filtered or unexported fields }
Topic broadcasts every published value to every current subscriber.
The zero value is ready to use, as sync.Mutex and sync.Map are; a Topic must not be copied after first use. Safe for concurrent use by any number of publishers and subscribers.
func (*Topic[T]) Close ¶
func (t *Topic[T]) Close()
Close releases every current subscriber — their Listener.Done channel closes, and a publisher parked on a Block delivery to any of them returns — and makes every later Topic.Subscribe hand back an already-done handle.
It does NOT close the value channels: values already delivered stay readable, and a publisher racing this call cannot crash on a closed channel. Close is idempotent.
func (*Topic[T]) Publish ¶
Publish delivers v to every current subscriber and returns how many of them took it.
A delivered count below Topic.Subscribers is the primitive telling the caller something did not land: a drop policy discarded the value (counted in that subscriber's Listener.Dropped), a subscriber left mid-fan-out, or ctx expired while a Block delivery was waiting. It is never zero-information.
ctx bounds only the Block policy — the drop policies never wait. It is the producer's escape hatch from a subscriber that has stopped reading, and the reason Block is a coupling rather than a hang.
Values from ONE publisher reach each subscriber in publication order. Values from concurrent publishers interleave in an unspecified order, and so does the order in which subscribers are visited inside a single Publish.
func (*Topic[T]) Subscribe ¶
func (t *Topic[T]) Subscribe(delivery DeliveryConfig) *Listener[T]
Subscribe registers a new subscriber and returns its handle.
delivery says what happens when this subscriber falls behind and MUST be built with Block, DropOldest or DropNewest; a zero DeliveryConfig panics, at the call that made the omission. See the package documentation for why there is no default.
Subscribing to a closed Topic is neither an error nor a panic: the returned Listener is already done, so a shutdown race reads as a shutdown at the call site rather than as a crash.
func (*Topic[T]) Subscribers ¶
Subscribers reports how many subscribers are currently registered. It is a point-in-time reading — useful next to a Publish result or in a gauge, never for a decision, since it can change before the next statement.
Source Files