Consumers

A consumer serves one route. It is a struct: its exported methods taking a *chanx.Context and a message are handlers, and methods with reserved names are hooks. You mount it with engine.Handle.

A complete consumer

type ChatConsumer struct {
	Moderation        // its HandleReport is promoted
	Rooms      *Rooms // dependencies are plain fields
}

// Groups is called once per connection: join the room named by the path.
func (c *ChatConsumer) Groups(conn *chanx.Conn) []string {
	return []string{"room." + conn.Param("room")}
}

// OnConnect runs once the connection is set up and its groups are joined.
func (c *ChatConsumer) OnConnect(ctx *chanx.Context) error {
	return ctx.Push(WelcomeMessage{Payload: c.Rooms.Banner})
}

// OnDisconnect runs while the groups are still joined, so it can still broadcast.
func (c *ChatConsumer) OnDisconnect(ctx *chanx.Context) {
	_ = ctx.Broadcast(ChatNotification{Payload: ChatPayload{Message: "someone left"}})
}

// HandleChat replies nothing directly: it broadcasts to the room.
func (c *ChatConsumer) HandleChat(ctx *chanx.Context, m ChatMessage) error {
	return ctx.Broadcast(ChatNotification{Payload: m.Payload}, chanx.ExcludeSelf())
}

// HandlePing replies: the return value goes back to the sender, echoing its ref.
func (c *ChatConsumer) HandlePing(ctx *chanx.Context, _ chanx.PingMessage) (chanx.PongMessage, error) {
	return chanx.PongMessage{}, nil
}

// Operations documents handlers in AsyncAPI, keyed by method name.
func (c *ChatConsumer) Operations() map[string]chanx.Operation {
	return map[string]chanx.Operation{
		"HandleChat": {
			Summary: "Send a chat message",
			Replies: []chanx.Message{ChatNotification{}}, // what HandleChat broadcasts
		},
	}
}

// Emits lists what the consumer sends outside any handler.
func (c *ChatConsumer) Emits() []chanx.Message { return []chanx.Message{WelcomeMessage{}} }

engine.Handle("/ws/chat/{room}", &ChatConsumer{Rooms: rooms})

Handlers

A handler is any exported method with one of these shapes:

func (c *ChatConsumer) HandleChat(ctx *chanx.Context, m ChatMessage) error                  // no reply
func (c *ChatConsumer) HandlePing(ctx *chanx.Context, m chanx.PingMessage) (chanx.PongMessage, error) // replies
func (c *ChatConsumer) HandleShipped(ctx *chanx.EventContext, e OrderShipped) (Notification, error) // an event

• The message type picks the action. Two handlers for one action panic at Handle.

• A returned message is the reply. It goes to the sender, with the frame's ref and "version": 1:

  > {"action": "ping", "ref": "42"}
  < {"action": "pong", "payload": null, "version": 1, "ref": "42"}

• A returned error is logged and the client gets a generic error frame, without the error's text:

  < {"action": "error", "payload": {"detail": "Failed to process message for report"}, "version": 1, "ref": "r1"}

  A panic is recovered and answered the same way.

• The method name is the AsyncAPI operation, in snake case: name it HandlePing to get Python's handle_ping.

• Each frame runs in its own goroutine, so a slow handler never blocks the socket. chanx.Sequential() on the route handles frames one at a time, in order, and chanx.MaxInFlight(n) caps how many run at once.

*chanx.Context is also a context.Context, canceled when the socket closes: pass it to your database calls.

Sending

Call	Goes to
return reply, nil	the sender, with its ref
ctx.Send(m)	the sender, with its ref: for several replies to one frame
ctx.Push(m)	the sender, without a ref: server-initiated
ctx.Broadcast(m)	the connection's groups, through the layer
ctx.Broadcast(m, chanx.ExcludeSelf())	the same, minus the sender
ctx.Broadcast(m, chanx.ToGroups("a", "b"))	other groups

Hooks

Matched by name and signature; a wrong signature panics at Handle.

Hook	Python chanx	Runs
Authenticate(*Context, *Request) (any, error)	authenticator_class	once, before anything; see Authentication
AuthenticationData(*Context, any) any	the authenticator's data	fills the authentication message
Credentials() []Credential	the authenticator's credentials	documents the auth in AsyncAPI
Groups(*Conn) []string	groups	on connect: the groups to join
OnConnect(*Context) error	post_authentication	once set up; an error closes with 1011
OnDisconnect(*Context)	websocket_disconnect	on close, groups still joined
Emits() []Message	outputs of handlers	declares what the consumer sends besides replies
Operations() map[string]Operation	@ws_handler(summary=..., output_type=...)	documents handlers, by method name
Passthrough() []Message	passthrough_events	events forwarded to the client unchanged
Topics() []TopicHandler	topics	the topics it hosts
OnValidationError(*Context, []Issue)	handle_validation_error	replaces the validation error frame
OnHandlerError(*Context, string, error)	handler error handling	replaces the generic error frame
OnEventError(*EventContext, string, error)	event error handling	an event handler failed

Operations sets Summary, Description, Tags, Name and Replies per handler. Replies matters for a handler that broadcasts instead of returning: without it, the AsyncAPI document and generated clients do not know what it sends.

One value, every connection

The consumer value you pass to Handle serves every connection, like an http.Handler. Keep dependencies on it (a database, a service), never per-connection state. That lives on the connection:

n, _ := chanx.ConnValue[int](ctx.Conn, visitsKey{})
ctx.Conn.SetValue(visitsKey{}, n+1)

ctx.Conn also holds the User, the path Params, the handshake Request and the layer ChannelName.

Mixins

Embed a struct, and its handlers are promoted:

// Moderation is a mixin: embed it, and its handlers come along.
type Moderation struct{}

func (Moderation) HandleReport(ctx *chanx.Context, m ReportMessage) (ReportedMessage, error) {
	if strings.TrimSpace(m.Payload.Reason) == "" {
		return ReportedMessage{}, errors.New("empty reason")
	}
	return ReportedMessage{}, nil
}

The embedding consumer's Groups and Passthrough are merged with the mixin's.

Without methods

chanx.NewConsumer builds the same from functions, for closures and dynamic setups. Hooks are fields; chanx.On registers a handler with a reply, chanx.Handle one without, chanx.OnEvent and chanx.HandleEvent the event versions:

// NewChatHub builds a consumer from functions, for closures and dynamic setups.
func NewChatHub(banner string) *chanx.Consumer {
	hub := chanx.NewConsumer("chat_hub")
	hub.Groups = func(conn *chanx.Conn) []string { return []string{"room." + conn.Param("room")} }
	hub.OnConnect = func(ctx *chanx.Context) error { return ctx.Push(WelcomeMessage{Payload: banner}) }
	hub.Emits = []chanx.Message{WelcomeMessage{}, ChatNotification{}}

	chanx.Handle(hub, func(ctx *chanx.Context, m ChatMessage) error {
		return ctx.Broadcast(ChatNotification{Payload: m.Payload}, chanx.ExcludeSelf())
	}, chanx.Summary("Send a chat message"))
	chanx.On(hub, func(ctx *chanx.Context, _ chanx.PingMessage) (chanx.PongMessage, error) {
		return chanx.PongMessage{}, nil
	})
	return hub
}

Catching mistakes early

Handle panics at startup on a definition mistake: a handler with a wrong signature, two handlers for one action, an unknown layer alias. chanx.Check reports the same as an error, for a one-line unit test:

func TestConsumer(t *testing.T) {
	if err := chanx.Check(&ChatConsumer{}); err != nil {
		t.Fatal(err)
	}
}

Next: Groups and events.