Topics

A topic is a stream a client subscribes to by name, such as ticker:ACME or room:lobby. One socket can follow any number of them. Groups are decided by the server when a client connects; topics are chosen by the client, while connected.

Declaring a topic

A topic is a struct embedding chanx.BaseTopic, with a pattern. It has its own handlers and hooks, like a small consumer:

// TickerTopic is one stream per symbol: clients subscribe to "ticker:ACME".
type TickerTopic struct {
	chanx.BaseTopic `pattern:"ticker:{symbol}" description:"Live prices for one symbol"`
}

// Authorize decides who may subscribe; false answers an "unauthorized" error.
func (TickerTopic) Authorize(ctx *chanx.Context, p chanx.Params) (bool, error) {
	return p["symbol"] != "SECRET", nil
}

// OnSubscribe runs once per subscription: push the current state first.
func (TickerTopic) OnSubscribe(ctx *chanx.Context) error {
	return ctx.Push(PriceUpdate{Payload: PricePayload{Symbol: ctx.Params["symbol"], Price: 100}})
}

// HandleQuote handles a frame sent to the topic, and replies to the sender.
func (TickerTopic) HandleQuote(ctx *chanx.Context, m QuoteMessage) (PriceUpdate, error) {
	return PriceUpdate{Payload: PricePayload{Symbol: ctx.Params["symbol"], Price: m.Payload.Price}}, nil
}

// Passthrough forwards PriceUpdate events published to the topic to its
// subscribers unchanged. Without it, or an event handler, they are dropped.
func (TickerTopic) Passthrough() []chanx.Message { return []chanx.Message{PriceUpdate{}} }

Tag on BaseTopic	Meaning
pattern	the topic names it serves; {symbol} is a parameter, stopping at :
name	the group namespace; the type name by default
description	its description in AsyncAPI
layer	the layer alias it publishes on

Hook	Runs
Authorize(*Context, Params) (bool, error)	before a subscription: false refuses it
OnSubscribe(*Context) error	after it: push the current state
OnUnsubscribe(*Context) error	on unsubscribe, and on disconnect
Passthrough() []Message	events published to the topic, forwarded to subscribers

Hosting topics

A consumer lists the topics it hosts, and a topic can also have a route of its own:

// MarketConsumer hosts the ticker topic: one socket, any number of symbols.
type MarketConsumer struct{}

func (MarketConsumer) Topics() []chanx.TopicHandler { return []chanx.TopicHandler{TickerTopic{}} }

func Mount(engine *chanx.Engine) {
	engine.Handle("/ws/market", MarketConsumer{})
	engine.Handle("/ws/ticker/{symbol}", chanx.TopicConsumer(TickerTopic{})) // one symbol per socket
}

The conversation

Subscribe, with a ref; the confirmation carries it, then OnSubscribe pushes the state:

> {"action": "subscribe", "topic": "ticker:ACME", "ref": "s1"}
< {"action": "subscribed", "payload": null, "version": 1, "topic": "ticker:ACME", "ref": "s1"}
< {"action": "price_update", "payload": {"symbol": "ACME", "price": 100.0}, "version": 1, "topic": "ticker:ACME"}

A frame for a topic carries its topic, and so does every reply:

> {"action": "quote", "topic": "ticker:ACME", "ref": "q1", "payload": {"price": 101.5}}
< {"action": "price_update", "payload": {"symbol": "ACME", "price": 101.5}, "version": 1, "topic": "ticker:ACME", "ref": "q1"}

Refusals are error frames with a reason:

< {"action": "error", "payload": {"reason": "unauthorized", "detail": "Not allowed to subscribe to 'ticker:SECRET'"}, "version": 1, "topic": "ticker:SECRET"}
< {"action": "error", "payload": {"reason": "not_subscribed", "detail": "Not subscribed to 'ticker:OTHER'"}, "version": 1, "topic": "ticker:OTHER"}
< {"action": "error", "payload": {"reason": "unknown_topic", "detail": "Unknown topic 'weather:paris'"}, "version": 1, "topic": "weather:paris"}

A second subscribe to the same topic confirms again without rerunning OnSubscribe. Unsubscribe the same way:

> {"action": "unsubscribe", "topic": "ticker:ACME", "ref": "u1"}
< {"action": "unsubscribed", "payload": null, "version": 1, "topic": "ticker:ACME", "ref": "u1"}

On a dedicated route (/ws/ticker/ACME), the topic comes from the URL: the client is subscribed on connect, and its frames need no topic field.

Publishing to a topic

Inside a topic handler, ctx.BroadcastTopic(m) reaches every subscriber. From anywhere else:

// PublishPrice pushes a price to every subscriber of the symbol, from a worker or
// any process sharing the layer. seq lets clients detect a gap.
func PublishPrice(ctx context.Context, publisher *chanx.Publisher, symbol string, price float64, seq int64) error {
	topic, err := chanx.FormatTopic[TickerTopic](chanx.Params{"symbol": symbol})
	if err != nil {
		return err
	}
	update := PriceUpdate{Payload: PricePayload{Symbol: symbol, Price: price}}
	return chanx.BroadcastTopic[TickerTopic](ctx, publisher, topic, update, chanx.Seq(seq))
}

< {"action": "price_update", "payload": {"symbol": "ACME", "price": 102.25}, "version": 1, "topic": "ticker:ACME", "seq": 7}

seq is yours to assign: a client that sees 7 after 5 knows it missed one. Leave it unset on a topic with several writers.

Declare what a topic forwards

An event published to a topic reaches subscribers only through an event handler on the topic or its Passthrough. Without either, it is dropped and logged as "No event handlers registered".

Group names

A topic's subscribers are a layer group: TickerTopic.ticker-ACME, the type name and the topic. Python chanx names a topic class's group the same way, so a Go server and a Python server hosting a topic of the same name meet on a shared Redis without configuration. chanx.TopicGroup[T](topic) returns it, and chanx.FormatTopic[T](params) fills the pattern.

Next: Authentication.