Groups and events

A group is a name connections join, such as room.lobby or user.ann. A broadcast to a group goes through the channel layer, so it reaches every member on every process that shares the layer. Events are the server's side of this: messages between servers, that each connection turns into what its client should see.

Groups

Groups returns the groups a connection joins on connect; it leaves them on disconnect:

func (ChatConsumer) Groups(c *chanx.Conn) []string {
	return []string{"room." + c.Param("room"), "user." + c.User.(User).ID}
}

From a handler, ctx.Broadcast(m) sends to the connection's own groups. Join and leave at run time with ctx.Conn.Join(ctx, group) and ctx.Conn.Leave(ctx, group).

A broadcast arrives exactly as a reply would, without a ref:

{"action": "chat_notification", "payload": {"message": "hi"}}

Group names

Layers accept letters, digits, ., - and _, under 100 characters. Build names from user input with protocol.SafeGroupName("room", name), which every chanx implementation computes the same way.

Events

A broadcast sends a message to clients as it is. An event is different: it is delivered to each connection's event handler, which decides what the client receives, if anything.

// OrderShipped is an event: a message between servers. A worker publishes it; each
// connection's event handler decides what, if anything, its client sees.
type OrderShipped struct {
	chanx.BaseMessage `action:"order_shipped"`
	Payload           struct {
		OrderID string `json:"order_id"`
	} `json:"payload"`
}

// Notification is what the client receives.
type Notification struct {
	chanx.BaseMessage `action:"notification"`
	Payload           NotificationPayload `json:"payload"`
}

type NotificationConsumer struct{}

// Groups puts each connection in its user's group.
func (NotificationConsumer) Groups(c *chanx.Conn) []string {
	return []string{"user." + c.Param("user")}
}

// HandleOrderShipped takes an *chanx.EventContext: it handles a layer event, not
// a client frame. Its return value is sent to the client.
func (NotificationConsumer) HandleOrderShipped(ctx *chanx.EventContext, e OrderShipped) (Notification, error) {
	return Notification{Payload: NotificationPayload{Text: "Order " + e.Payload.OrderID + " shipped"}}, nil
}

The handler takes a *chanx.EventContext; its return value goes to the client. A handler returning only an error sends nothing, for an event that only updates state.

Events forwarded unchanged need no handler, only a declaration:

// Announcement goes to clients as it is: no handler needed.
type Announcement struct {
	chanx.BaseMessage `action:"announcement"`
	Payload           string `json:"payload"`
}

// Passthrough lists events forwarded to the client unchanged.
func (NotificationConsumer) Passthrough() []chanx.Message { return []chanx.Message{Announcement{}} }

From outside a socket

A chanx.Publisher pushes to groups from anywhere: an HTTP handler, a worker, a cron job. It needs the layers, not the engine, so a worker process holding no sockets can use it too.

// ShippedHandler is a plain HTTP endpoint: POST /orders/{id}/shipped?user=ann
// pushes to every socket of that user, on any server sharing the layer.
func ShippedHandler(publisher *chanx.Publisher) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var event OrderShipped
		event.Payload.OrderID = r.PathValue("id")
		group := "user." + r.URL.Query().Get("user")
		if err := publisher.BroadcastEvent(r.Context(), event, group); err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		_ = json.NewEncoder(w).Encode(map[string]string{"status": "sent"})
	})
}

curl -X POST "localhost:8000/orders/A17/shipped?user=ann"

Every socket Ann has open receives:

{"action": "notification", "payload": {"text": "Order A17 shipped"}}

Publisher method	Does
BroadcastEvent(ctx, event, groups...)	an event to every connection in the groups
SendEvent(ctx, channel, event)	an event to one connection, by its ctx.Conn.ChannelName
BroadcastMessage(ctx, groups, m)	a message to clients unchanged, no event handler
chanx.BroadcastTopic[T](ctx, p, topic, event)	an event to a topic's subscribers

Get one with engine.Publisher(), ctx.Publisher() inside a handler, or chanx.NewPublisher(layers) in a process without an engine. .Layer("alias") picks another layer.

Python processes on the same Redis receive these events too, and Go connections receive theirs: the layer messages are the same. See Redis and Python.

Completion frames

With SendCompletion on, the server confirms each processed frame, event and group message, for clients that wait until a request is done:

{"action": "complete", "payload": null, "version": 1, "ref": "7"}
{"action": "event_complete", "payload": null}
{"action": "group_complete", "payload": null}

Turn it on for every route with chanx.Config{SendCompletion: true}, or one with the chanx.SendCompletion(true) route option.

Next: Topics.