DurableObjectSocketPubSub

En Workers, las conexiones Socket no viven en el aislado que atiende la petición, sino en el Durable Object de su sala. DurableObjectSocketPubSub es el único broker que llega a ellas: publish entrega el mensaje a la sala por RPC (o directamente, si ya se ejecuta dentro de esa sala) y se resuelve cuando la sala lo acepta; la sala lo reparte a sus peers en orden de publicación.

Es el broker por defecto en Workers: con el binding MEEKO_SOCKET y sin socket.pubsub, el Worker generado lo usa. Declárelo sólo para dejarlo explícito o con otro binding; cualquier otro broker se rechaza.

Uso#

// server/runtime.ts
import { DurableObjectSocketPubSub } from "@meekojs/meeko/cloudflare";
import type { DurableObjectNamespaceLike } from "@meekojs/meeko/cloudflare";
import { defineServerRuntime } from "@meekojs/meeko/server";

interface Env { readonly MEEKO_SOCKET?: DurableObjectNamespaceLike }

export default defineServerRuntime<Env>(({ env }) => ({
  socket: env.MEEKO_SOCKET ? { pubsub: new DurableObjectSocketPubSub(env.MEEKO_SOCKET) } : {},
}));

env.MEEKO_SOCKET no existe durante el prerender, que se ejecuta en Node.

Tipo#

class DurableObjectSocketPubSub implements SocketPubSub {
  constructor(namespace: DurableObjectNamespaceLike);
  publish(channel: string, room: string, message: SocketMessage): Promise<void>;
  subscribe(channel: string, room: string, listener: RoomListener): Unsubscribe;
}

interface DurableObjectNamespaceLike {
	idFromName(name: string): DurableObjectIdLike
	get(id: DurableObjectIdLike): DurableObjectStubLike
}

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>;
}

Parámetros#

new DurableObjectSocketPubSub(namespace):

Valor devuelto#

Un SocketPubSub:

Ejemplos#

Broadcast desde una API#

// server/api/documents/[id]/publish.post.ts
export default defineApiRoute({
  params: documentParams,
  response: documentSchema,
  auth: "required",
  handler: async ({ params }) => {
    const document = await publishDocument(params.id);
    await socket.documents.broadcast(document.id, "published", { id: document.id, at: document.publishedAt });
    return document;
  },
});

El mismo código funciona en Node con LocalSocketPubSub o RedisSocketPubSub: broadcast usa el broker del runtime.

Una sala por documento#

/_meeko/socket/documents?room=doc_42  → Durable Object "documents\u0000doc_42"
/_meeko/socket/documents?room=doc_43  → Durable Object "documents\u0000doc_43"

Cada sala procesa sus mensajes en un solo hilo: repartir por documento escala mejor que una sala global.

Errores comunes#

Un binding que no es de Durable Objects#

new DurableObjectSocketPubSub(env.DB as never);

TypeError: DurableObjectSocketPubSub requires a Durable Object namespace binding.

new DurableObjectSocketPubSub(env.MEEKO_SOCKET);

Suscribirse fuera de la sala#

new DurableObjectSocketPubSub(env.MEEKO_SOCKET).subscribe("chat", "general", (message) => console.log(message));

Error: DurableObjectSocketPubSub subscribes only inside the Socket room Durable Object; forward upgrades with createWorker. Las suscripciones sólo existen dentro de la sala; todo upgrade debe pasar por createWorker.

export default createWorker(createCloudflareApp); // reenvía cada upgrade a su sala

Una clase de salas que no es la de Meeko#

MEEKO_SOCKET apunta a una clase Durable Object propia

El primer publish rechaza con TypeError: The Socket Durable Object must be created with createSocketRoomClass: la sala debe ofrecer el RPC meekoPublish.

export const MeekoSocketRoom = createSocketRoomClass(createCloudflareApp, DurableObject);

Otro broker en Workers#

socket: { pubsub: new LocalSocketPubSub() }

El Worker y la sala rechazan la aplicación al crearla: 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: {} // el Worker usa DurableObjectSocketPubSub(env.MEEKO_SOCKET)

Notas#

• Entrega. A lo sumo una vez y en orden de publicación por sala. Una publicación hacia la propia sala no hace RPC.
• Validación. El hub valida evento y payload antes de publicar; la sala los vuelve a validar y aplica authorize y authorizeSend a cada peer.
• Hibernación. Una publicación despierta la sala si estaba hibernada; espera a que las conexiones se reconstruyan como mucho rebuildTimeoutMs.
• Sin TCP ni Redis. No hace falta un broker externo: el Durable Object es el punto único de cada sala.

Relacionado#

• createSocketRoomClassCrea la clase Durable Object que aloja los WebSockets de una sala de Socket en Cloudflare Workers, con la API de hibernación, reconstrucción al despertar y límites por sala.
• createWorkerSirve una aplicación de Meeko desde un Worker de módulo de Cloudflare, creada una vez por aislado con los bindings, con los upgrades de Socket reenviados al Durable Object de su sala.
• socketRoomNameEl nombre del Durable Object de una sala de Socket en Cloudflare Workers, un objeto por canal y sala, para idFromName.
• LocalSocketPubSubBroker de broadcasts de Socket dentro de un solo proceso, el que usa el hub por defecto en Node, Bun y Deno; no alcanza a otras instancias.
• RedisSocketPubSubReparte los broadcasts de Socket entre varias instancias o procesos con el pub/sub de Redis, también fragmentado en Redis Cluster.
• Socket y Durable ObjectsMeeko como framework full-stack
• Broadcast desde el servidorMeeko como framework full-stack