LocalSocketPubSub

LocalSocketPubSub distributes Socket broadcasts among peers connected to this process: each channel/room pair is a set of in-memory listeners, and publish calls them and waits for their local delivery attempts. It is the hub's default broker in Node, Bun and Deno when server/runtime.ts does not declare socket.pubsub.

With a single instance, nothing else is needed. With multiple replicas or processes, a broadcast in one does not reach peers in the others: use RedisSocketPubSub. It is not supported in Cloudflare Workers. Since 1.7, if a production application has channels and does not declare socket.pubsub, Meeko warns once that it is using this broker; explicitly declaring it (new LocalSocketPubSub()) indicates a single-instance deployment and removes the warning.

Usage#

// server/runtime.ts
import { defineServerRuntime, LocalSocketPubSub } from "@meekojs/meeko/server";

export default defineServerRuntime(() => ({
  socket: {
    pubsub: new LocalSocketPubSub({ onListenerError: (error) => console.warn("[socket] listener", error) }),
  },
}));

Without socket.pubsub, the hub creates an identical one with console.error as observer, and warns once in production (since 1.7).

Type#

class LocalSocketPubSub implements SocketPubSub {
  constructor(options?: LocalSocketPubSubOptions);
  publish(channel: string, room: string, message: SocketMessage): Promise<void>;
  subscribe(
    channel: string,
    room: string,
    listener: (message: SocketMessage) => void | Promise<void>,
  ): Unsubscribe;
}

interface LocalSocketPubSubOptions {
	onListenerError?(error: unknown): void
}

interface SocketPubSub {
  publish(channel: string, room: string, message: SocketMessage): Promise<void>;
  subscribe(
    channel: string,
    room: string,
    listener: (message: SocketMessage) => void | Promise<void>,
  ): Unsubscribe | Promise<Unsubscribe>;
}

interface SocketMessage {
	readonly event: string
	readonly payload: unknown
}

Parameters#

new LocalSocketPubSub(options?):

Return value#

A SocketPubSub:

Examples#

Broadcast from an API#

// server/api/orders/[id]/ready.post.ts
export default defineApiRoute({
  params: orderParams,
  response: orderSchema,
  handler: async ({ params }) => {
    const order = await markReady(params.id);
    // Validates the event and payload against the channel contract and publishes to the hub's broker.
    await socket.operations.broadcast(order.storeId, "kitchenChanged", { orderId: order.id, status: "ready" });
    return order;
  },
});

broadcast uses the runtime's broker: this same code works with LocalSocketPubSub, RedisSocketPubSub or the Workers Durable Object.

Listen to a channel/room pair in a test#

import { LocalSocketPubSub } from "@meekojs/meeko/server";

const pubsub = new LocalSocketPubSub();
const received: unknown[] = [];
const stop = pubsub.subscribe("chat", "general", (message) => { received.push(message.payload); });
await pubsub.publish("chat", "general", { event: "message", payload: { text: "Hello" } });
await pubsub.publish("chat", "otra", { event: "message", payload: { text: "No llega" } });
stop();
console.log(received); // [{ text: "Hello" }]

Choose the broker according to the deployment#

// server/runtime.ts
export default defineServerRuntime(async ({ env }) => {
  if (!env.MEEKO_REDIS_URL) return { socket: { pubsub: new LocalSocketPubSub() } }; // a single instance
  const { publisher, subscriber } = await connectRedis(env.MEEKO_REDIS_URL);
  return {
    socket: { pubsub: new RedisSocketPubSub({ publisher, subscriber, prefix: "app:prod:" }) },
    close: async () => { await subscriber.quit(); await publisher.quit(); },
  };
});

Common mistakes#

Multiple instances with the local broker#

2 replicas behind a load balancer, without socket.pubsub (LocalSocketPubSub by default)

broadcast resolves, but only peers connected to the replica that published it receive the message: users on the other replica do not see the change. Since 1.7, a production application with channels and without socket.pubsub warns once per process at startup, like the in-memory rateLimit counter: [meeko] Los canales de Socket usan LocalSocketPubSub, el broker en memoria de cada instancia: …. Meeko cannot count replicas, so the warning also appears with a single instance; declare new LocalSocketPubSub() to remove it.

socket: { pubsub: new RedisSocketPubSub({ publisher: redis, subscriber, prefix: "app:prod:" }) }

Use it in Cloudflare Workers#

socket: { pubsub: new LocalSocketPubSub() } // in a Worker

Connections live in each room's Durable Object, not in the isolate. The Worker rejects the application on creation: Socket channels on Cloudflare Workers require DurableObjectSocketPubSub: leave runtime.socket.pubsub unset and bind MEEKO_SOCKET, or set it to new DurableObjectSocketPubSub(env.MEEKO_SOCKET)….

socket: env.MEEKO_SOCKET ? { pubsub: new DurableObjectSocketPubSub(env.MEEKO_SOCKET) } : {}

Notes#

• Delivery. At most once per peer, in publication order; a peer whose queue exceeds maxPendingDeliveries closes with 1013.
• Validation. The hub validates the event and payload before publishing and again on delivery; the broker only transports them.
• Memory. A room's listeners are released when its last peer leaves.
• Development. In meeko dev, recreating the application (creating or deleting a route) closes open connections and clients reconnect.

Related#

• RedisSocketPubSubDistributes Socket broadcasts across multiple instances or processes with Redis pub/sub, including sharded pub/sub in Redis Cluster.
• DurableObjectSocketPubSubThe Socket broker for Cloudflare Workers, where each channel and room is a Durable Object that receives publications over RPC and distributes them to its peers.
• defineSocketChannelDeclares a WebSocket channel in server/socket with schemas for each direction, per-room and per-event guards, and handlers for client messages.
• createSocketHubCreates the Socket hub for a custom host, authorizing handshakes, accepting connections, validating events and distributing broadcasts across rooms and instances.
• Broadcast and multiple instancesMeeko as a full-stack framework
• Broadcast from the serverMeeko as a full-stack framework