Documentation
¶
Overview ¶
Package pubsub provides a lightweight in-process broker for fan-out event delivery between services and the UI.
Delivery semantics:
Broker.Publish is best-effort and lossy under contention. If a subscriber's channel is full, the event is dropped for that subscriber, a warning is logged, and a counter is incremented. This is the right choice for high-frequency intermediate updates (e.g. streaming token deltas) where only the latest state matters.
Broker.PublishMustDeliver is bounded-blocking. For each subscriber it first tries a non-blocking send, then falls back to a per-subscriber blocking send with a hard timeout. On timeout the event is dropped for that subscriber, an error is logged, and the must-deliver drop counter is incremented. The publisher never blocks indefinitely. This is the right choice for terminal events (finish, tool result, error, cancel) that must not be silently coalesced away.
Drop counters (Broker.DropCount, Broker.MustDeliverDropCount) are exposed so callers can surface saturation in telemetry.
Index ¶
- type Broker
- func (b *Broker[T]) DropCount() uint64
- func (b *Broker[T]) GetSubscriberCount() int
- func (b *Broker[T]) MustDeliverDropCount() uint64
- func (b *Broker[T]) Publish(t EventType, payload T)
- func (b *Broker[T]) PublishMustDeliver(ctx context.Context, t EventType, payload T)
- func (b *Broker[T]) SetMustDeliverTimeout(d time.Duration)
- func (b *Broker[T]) Shutdown()
- func (b *Broker[T]) Subscribe(ctx context.Context) <-chan Event[T]
- type Event
- type EventType
- type Payload
- type PayloadType
- type Publisher
- type Subscriber
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Broker ¶
type Broker[T any] struct { // contains filtered or unexported fields }
func NewBrokerWithOptions ¶
func (*Broker[T]) DropCount ¶
DropCount returns the cumulative number of events dropped by Broker.Publish because a subscriber's channel was full.
func (*Broker[T]) GetSubscriberCount ¶
func (*Broker[T]) MustDeliverDropCount ¶
MustDeliverDropCount returns the cumulative number of events dropped by Broker.PublishMustDeliver after the per-subscriber timeout expired.
func (*Broker[T]) Publish ¶
Publish delivers an event to every active subscriber.
Delivery is non-blocking and lossy: if a subscriber's channel is full the event is dropped for that subscriber, a warning is logged, and Broker.DropCount is incremented. Use Broker.PublishMustDeliver for events that must not be silently dropped.
func (*Broker[T]) PublishMustDeliver ¶
PublishMustDeliver delivers an event with bounded-blocking semantics. For each subscriber it first attempts a non-blocking send, then falls back to a blocking send bounded by a per-subscriber timeout (default [defaultMustDeliverTimeout]). On timeout the event is dropped for that subscriber, Broker.MustDeliverDropCount is incremented, and an error is logged. The publisher never blocks indefinitely.
Use this for terminal events that must reach subscribers (finish, tool result, error, cancel). Callers must still tolerate rare drops after timeout — recovery is the subscriber's responsibility (e.g. a re-fetch on the next session-visible event).
func (*Broker[T]) SetMustDeliverTimeout ¶
SetMustDeliverTimeout overrides the per-subscriber timeout used by Broker.PublishMustDeliver. A zero or negative value resets to the default. Intended primarily for tests.
type Payload ¶
type Payload struct {
Type PayloadType `json:"type"`
Payload json.RawMessage `json:"payload"`
}
Payload wraps a discriminated JSON payload with a type tag.
type PayloadType ¶
type PayloadType = string
PayloadType identifies the type of event payload for discriminated deserialization over JSON.
const ( PayloadTypeLSPEvent PayloadType = "lsp_event" PayloadTypeMCPEvent PayloadType = "mcp_event" PayloadTypePermissionRequest PayloadType = "permission_request" PayloadTypePermissionNotification PayloadType = "permission_notification" PayloadTypeMessage PayloadType = "message" PayloadTypeSession PayloadType = "session" PayloadTypeFile PayloadType = "file" PayloadTypeAgentEvent PayloadType = "agent_event" PayloadTypeConfigChanged PayloadType = "config_changed" PayloadTypeSkillsEvent PayloadType = "skills_event" PayloadTypeRunComplete PayloadType = "run_complete" )
type Publisher ¶
type Publisher[T any] interface { Publish(EventType, T) PublishMustDeliver(context.Context, EventType, T) }
Publisher can publish events of type T.
Publish is best-effort and lossy under back-pressure; PublishMustDeliver applies the bounded-blocking semantics used for terminal events that must reach subscribers (finish, tool result, error, cancel, RunComplete). See Broker.Publish and Broker.PublishMustDeliver.
Source Files