broker

Documentation

Overview

Package broker is a small in-process pub/sub for session events.

Subscribers receive a buffered channel; if the consumer is slow and the buffer fills, events are dropped silently rather than blocking the publisher. This keeps the sender's stream goroutine moving even when an SSE client stalls.

Index

• type Broker
  • func New() *Broker
  • func (b *Broker) HasViewers(sessionID string) bool
  • func (b *Broker) Publish(ev Event)
  • func (b *Broker) Subscribe(sessionID string) (<-chan Event, func())
  • func (b *Broker) SubscribeAll() (<-chan Event, func())
• type Event

Types

type Broker

type Broker struct {
	// contains filtered or unexported fields
}

func New

func New() *Broker

func (*Broker) HasViewers

func (b *Broker) HasViewers(sessionID string) bool

HasViewers reports whether any per-session subscriber is currently attached to sessionID — i.e. a browser has its /events stream open on that session. Global (SubscribeAll) subscribers don't count, so the push dispatcher asking this never sees itself. Used to suppress notifications for the session a user is actively looking at.

func (*Broker) Publish

func (b *Broker) Publish(ev Event)

Publish delivers ev to every current subscriber of ev.SessionID and to every global (SubscribeAll) subscriber. The non-blocking sends run while holding the read lock so a concurrent cancel — which closes a channel under the write lock — can't close a channel mid-send. Each send is non-blocking (drop-on-full), so holding the lock never stalls the publisher.

func (*Broker) Subscribe

func (b *Broker) Subscribe(sessionID string) (<-chan Event, func())

Subscribe returns a buffered channel of events for sessionID and a cancel function to stop receiving and free the slot.

func (*Broker) SubscribeAll

func (b *Broker) SubscribeAll() (<-chan Event, func())

SubscribeAll returns a buffered channel receiving events for EVERY session, plus a cancel function. Used by consumers that watch all sessions at once (the web-push dispatcher) rather than one open session (an SSE client). Same drop-on-slow-consumer semantics as Subscribe.

type Event

type Event struct {
	SessionID string          `json:"session_id"`
	Type      string          `json:"type"`
	Raw       json.RawMessage `json:"raw,omitempty"`
}