Implementing Pub/Sub Pattern
1. Understanding Publish-Subscribe Pattern
Publishers emit messages to topics; subscribers receive matching messages without direct coupling. WebSocket commonly carries the client side over a broker (Redis, NATS, Kafka).
| Role | Action |
|---|---|
| Publisher | Send to topic |
| Broker | Fan-out |
| Subscriber | Receive matching |
| Topic | Routing key (e.g. orders.*) |
2. Implementing Topic Subscription
Example: Subscribe / unsubscribe
ws.send(JSON.stringify({ "type": "sub", "topic": "prices.BTC" }));
ws.send(JSON.stringify({ "type": "unsub", "topic": "prices.BTC" }));| Message | Server Action |
|---|---|
sub | Add to topic subscriber list |
unsub | Remove from list |
list | Return current subs |
3. Managing Subscriber Lists
| Structure | Use |
|---|---|
Map<topic, Set<WebSocket>> | Direct fan-out |
Map<WebSocket, Set<topic>> | Cleanup on disconnect |
| Trie | Wildcard matching |
| Redis SET / Pub-Sub | Multi-node |
4. Publishing Messages to Topics
Example: Publish helper
function publish(topic, payload) {
const subs = topics.get(topic);
if (!subs) return;
const msg = JSON.stringify({ "type":"msg", "topic":topic, "payload":payload });
for (const c of subs) if (c.readyState === 1) c.send(msg);
}5. Implementing Topic Wildcards
| Convention | Match |
|---|---|
* (single segment) | orders.* → orders.new |
# (multi segment) | orders.# → orders.eu.new |
:param | Named capture (custom) |
6. Handling Private Channels
| Type | Access |
|---|---|
| Public | Anyone may subscribe |
Private (private- prefix) | Auth required |
| Presence | Auth + member list |
| Encrypted | End-to-end key per channel |
7. Implementing Message Filtering
Example: Predicate filter
// per-subscriber filter
function publishFiltered(topic, payload) {
for (const c of topics.get(topic) ?? []) {
if (c.filter?.(payload) ?? true) c.send(JSON.stringify(payload));
}
}| Filter | Use |
|---|---|
| Field predicate | { "sym":"BTC", "px>":50000 } |
| Geo filter | Bounding box / radius |
| Permission filter | Hide fields user can't read |
8. Managing Subscription State
| State | Detail |
|---|---|
| Persist on reconnect | Resubscribe from client |
| Server snapshot | Send initial state on subscribe |
| Last seq | Resume from gap |
9. Implementing Presence
Example: Presence events
{ "type": "presence.join", "topic": "room-42", "user": { "id": "u1", "name": "Ada" } }
{ "type": "presence.leave", "topic": "room-42", "user": { "id": "u1" } }
{ "type": "presence.list", "topic": "room-42", "users": [ /* ... */ ] }| Event | Trigger |
|---|---|
| join | Subscriber added |
| leave | Subscriber removed/disconnected |
| update | Metadata change |
| sync | Periodic full state |
10. Handling Subscription Limits
| Limit | Detail |
|---|---|
| Max topics / connection | e.g. 100 |
| Max subscribers / topic | e.g. 10k (fan-out cost) |
| Rate of sub/unsub | Token bucket |
| Wildcard depth | Cap to prevent explosions |