WebSockets en Bun

Bun.serve() soporta WebSockets del lado del servidor, con compresión sobre la marcha, soporte TLS y una API de publicación-suscripción nativa de Bun.

Nota

⚡️ 7 veces más rendimiento

Los WebSockets de Bun son rápidos. Para un chatroom simple en Linux x64, Bun puede manejar 7 veces más solicitudes por segundo que Node.js + "ws".

Mensajes enviados por segundo	Runtime	Clientes
~700,000	(`Bun.serve`) Bun v0.2.1 (x64)	16
~100,000	(`ws`) Node v18.10.0 (x64)	16

Internamente, la implementación WebSocket de Bun está construida sobre uWebSockets.

Iniciar un servidor WebSocket

A continuación hay un servidor WebSocket simple construido con Bun.serve, en el que todas las solicitudes entrantes se actualizan a conexiones WebSocket en el manejador fetch. Los manejadores de socket se declaran en el parámetro websocket.

Bun.serve({
  fetch(req, server) {
    // actualizar la solicitud a un WebSocket
    if (server.upgrade(req)) {
      return; // no devolver una Response
    }
    return new Response("Actualización fallida", { status: 500 });
  },
  websocket: {}, // manejadores
});

Los siguientes manejadores de eventos WebSocket son soportados:

Bun.serve({
  fetch(req, server) {}, // lógica de actualización
  websocket: {
    message(ws, message) {}, // se recibe un mensaje
    open(ws) {}, // se abre un socket
    close(ws, code, message) {}, // se cierra un socket
    drain(ws) {}, // el socket está listo para recibir más datos
  },
});

Una API diseñada para velocidad

En Bun, los manejadores se declaran una vez por servidor, en lugar de por socket.

ServerWebSocket espera que pases un objeto WebSocketHandler al método Bun.serve() que tiene métodos para open, message, close, drain y error. Esto es diferente de la clase WebSocket del lado del cliente que extiende EventTarget (onmessage, onopen, onclose).

Los clientes tienden a no tener muchas conexiones de socket abiertas, así que una API basada en eventos tiene sentido.

Pero los servidores tienden a tener muchas conexiones de socket abiertas, lo que significa:

El tiempo dedicado a agregar/eliminar listeners de eventos para cada conexión se acumula
Memoria extra dedicada a almacenar referencias a funciones de callback para cada conexión
Usualmente, las personas crean nuevas funciones para cada conexión, lo que también significa más memoria

Entonces, en lugar de usar una API basada en eventos, ServerWebSocket espera que pases un solo objeto con métodos para cada evento en Bun.serve() y se reutiliza para cada conexión.

Esto lleva a menos uso de memoria y menos tiempo dedicado a agregar/eliminar listeners de eventos.

El primer argumento de cada manejador es la instancia de ServerWebSocket manejando el evento. La clase ServerWebSocket es una implementación rápida y nativa de Bun de WebSocket con algunas características adicionales.

Bun.serve({
  fetch(req, server) {}, // lógica de actualización
  websocket: {
    message(ws, message) {
      ws.send(message); // eco del mensaje
    },
  },
});

Enviar mensajes

Cada instancia de ServerWebSocket tiene un método .send() para enviar mensajes al cliente. Soporta varios tipos de entrada.

Bun.serve({
  fetch(req, server) {}, // lógica de actualización
  websocket: {
    message(ws, message) {
      ws.send("Hola mundo"); // cadena
      ws.send(response.arrayBuffer()); // ArrayBuffer
      ws.send(new Uint8Array([1, 2, 3])); // TypedArray | DataView
    },
  },
});

Encabezados

Una vez que la actualización tiene éxito, Bun enviará una respuesta 101 Cambiando Protocolos según la especificación. Se pueden adjuntar headers adicionales a esta Response en la llamada a server.upgrade().

Bun.serve({
  fetch(req, server) {
    const sessionId = await generateSessionId();
    server.upgrade(req, {
      headers: { 
        "Set-Cookie": `SessionId=${sessionId}`, 
      }, 
    });
  },
  websocket: {}, // manejadores
});

Datos contextuales

Se pueden adjuntar data contextuales a un nuevo WebSocket en la llamada .upgrade(). Estos datos están disponibles en la propiedad ws.data dentro de los manejadores WebSocket.

Para tipar fuertemente ws.data, agrega una propiedad data al objeto del manejador websocket. Esto tipa ws.data en todos los hooks del ciclo de vida.

type WebSocketData = {
  createdAt: number;
  channelId: string;
  authToken: string;
};

Bun.serve({
  fetch(req, server) {
    const cookies = new Bun.CookieMap(req.headers.get("cookie")!);

    server.upgrade(req, {
      // este objeto debe conformar a WebSocketData
      data: {
        createdAt: Date.now(),
        channelId: new URL(req.url).searchParams.get("channelId"),
        authToken: cookies.get("X-Token"),
      },
    });

    return undefined;
  },
  websocket: {
    // TypeScript: especificar el tipo de ws.data así
    data: {} as WebSocketData,
    // manejador llamado cuando se recibe un mensaje
    async message(ws, message) {
      // ws.data ahora está correctamente tipado como WebSocketData
      const user = getUserFromToken(ws.data.authToken);

      await saveMessageToDatabase({
        channel: ws.data.channelId,
        message: String(message),
        userId: user.id,
      });
    },
  },
});

Nota

Nota: Anteriormente, podías especificar el tipo de ws.data usando un parámetro de tipo en Bun.serve, como Bun.serve<MyData>({...}). Este patrón se eliminó debido a una limitación en TypeScript en favor de la propiedad data mostrada arriba.

Para conectarse a este servidor desde el navegador, crea un nuevo WebSocket.

const socket = new WebSocket("ws://localhost:3000/chat");

socket.addEventListener("message", event => {
  console.log(event.data);
});

Nota

Identificar usuarios

Las cookies que están actualmente establecidas en la página se enviarán con la solicitud de actualización de WebSocket y estarán disponibles en req.headers en el manejador fetch. Analiza estas cookies para determinar la identidad del usuario que se conecta y establece el valor de data en consecuencia.

Pub/Sub

La implementación ServerWebSocket de Bun implementa una API nativa de publicación-suscripción para difusión basada en temas. Los sockets individuales pueden .subscribe() a un tema (especificado con un identificador de cadena) y .publish() mensajes a todos los demás suscriptores a ese tema (excluyéndose a sí mismo). Esta API de difusión basada en temas es similar a MQTT y Redis Pub/Sub.

const server = Bun.serve({
  fetch(req, server) {
    const url = new URL(req.url);
    if (url.pathname === "/chat") {
      console.log(`¡actualización!`);
      const username = getUsernameFromReq(req);
      const success = server.upgrade(req, { data: { username } });
      return success ? undefined : new Response("Error de actualización de WebSocket", { status: 400 });
    }

    return new Response("Hola mundo");
  },
  websocket: {
    // TypeScript: especificar el tipo de ws.data así
    data: {} as { username: string },
    open(ws) {
      const msg = `${ws.data.username} ha entrado al chat`;
      ws.subscribe("el-chat-grupal");
      server.publish("el-chat-grupal", msg);
    },
    message(ws, message) {
      // este es un chat grupal
      // así que el servidor retransmite los mensajes entrantes a todos
      server.publish("el-chat-grupal", `${ws.data.username}: ${message}`);

      // inscribir suscripciones actuales
      console.log(ws.subscriptions); // ["el-chat-grupal"]
    },
    close(ws) {
      const msg = `${ws.data.username} ha dejado el chat`;
      ws.unsubscribe("el-chat-grupal");
      server.publish("el-chat-grupal", msg);
    },
  },
});

console.log(`Escuchando en ${server.hostname}:${server.port}`);

Llamar a .publish(data) enviará el mensaje a todos los suscriptores de un tema excepto el socket que llamó a .publish(). Para enviar un mensaje a todos los suscriptores de un tema, usa el método .publish() en la instancia Server.

const server = Bun.serve({
  websocket: {
    // ...
  },
});

// escuchar algún evento externo
server.publish("el-chat-grupal", "Hola mundo");

Compresión

La compresión por mensaje se puede habilitar con el parámetro perMessageDeflate.

Bun.serve({
  websocket: {
    perMessageDeflate: true, 
  },
});

La compresión se puede habilitar para mensajes individuales pasando un boolean como segundo argumento a .send().

ws.send("Hola mundo", true);

Para un control detallado sobre las características de compresión, consulta la Referencia.

Contrapresión

El método .send(message) de ServerWebSocket devuelve un number que indica el resultado de la operación.

-1 — El mensaje se encoló pero hay contrapresión
0 — El mensaje se descartó debido a un problema de conexión
1+ — El número de bytes enviados

Esto te da mejor control sobre la contrapresión en tu servidor.

Tiempos de espera y límites

Por defecto, Bun cerrará una conexión WebSocket si está inactiva durante 120 segundos. Esto se puede configurar con el parámetro idleTimeout.

Bun.serve({
  fetch(req, server) {}, // lógica de actualización
  websocket: {
    idleTimeout: 60, // 60 segundos
  },
});

Bun también cerrará una conexión WebSocket si recibe un mensaje más grande que 16 MB. Esto se puede configurar con el parámetro maxPayloadLength.

Bun.serve({
  fetch(req, server) {}, // lógica de actualización
  websocket: {
    maxPayloadLength: 1024 * 1024, // 1 MB
  },
});

Conectarse a un servidor `Websocket`

Bun implementa la clase WebSocket. Para crear un cliente WebSocket que se conecte a un servidor ws:// o wss://, crea una instancia de WebSocket, como lo harías en el navegador.

const socket = new WebSocket("ws://localhost:3000");

// Con negociación de subprotocolo
const socket2 = new WebSocket("ws://localhost:3000", ["soap", "wamp"]);

En los navegadores, las cookies que están actualmente establecidas en la página se enviarán con la solicitud de actualización de WebSocket. Esta es una característica estándar de la API WebSocket.

Para conveniencia, Bun te permite establecer encabezados personalizados directamente en el constructor. Esta es una extensión específica de Bun del estándar WebSocket. Esto no funcionará en navegadores.

const socket = new WebSocket("ws://localhost:3000", {
  headers: {
    /* encabezados personalizados */
  }, 
});

Para agregar listeners de eventos al socket:

// se recibe un mensaje
socket.addEventListener("message", event => {});

// socket abierto
socket.addEventListener("open", event => {});

// socket cerrado
socket.addEventListener("close", event => {});

// manejador de errores
socket.addEventListener("error", event => {});

Referencia

namespace Bun {
  export function serve(params: {
    fetch: (req: Request, server: Server) => Response | Promise<Response>;
    websocket?: {
      message: (ws: ServerWebSocket, message: string | ArrayBuffer | Uint8Array) => void;
      open?: (ws: ServerWebSocket) => void;
      close?: (ws: ServerWebSocket, code: number, reason: string) => void;
      error?: (ws: ServerWebSocket, error: Error) => void;
      drain?: (ws: ServerWebSocket) => void;

      maxPayloadLength?: number; // predeterminado: 16 * 1024 * 1024 = 16 MB
      idleTimeout?: number; // predeterminado: 120 (segundos)
      backpressureLimit?: number; // predeterminado: 1024 * 1024 = 1 MB
      closeOnBackpressureLimit?: boolean; // predeterminado: false
      sendPings?: boolean; // predeterminado: true
      publishToSelf?: boolean; // predeterminado: false

      perMessageDeflate?:
        | boolean
        | {
            compress?: boolean | Compressor;
            decompress?: boolean | Compressor;
          };
    };
  }): Server;
}

type Compressor =
  | `"disable"`
  | `"shared"`
  | `"dedicated"`
  | `"3KB"`
  | `"4KB"`
  | `"8KB"`
  | `"16KB"`
  | `"32KB"`
  | `"64KB"`
  | `"128KB"`
  | `"256KB"`;

interface Server {
  pendingWebSockets: number;
  publish(topic: string, data: string | ArrayBufferView | ArrayBuffer, compress?: boolean): number;
  upgrade(
    req: Request,
    options?: {
      headers?: HeadersInit;
      data?: any;
    },
  ): boolean;
}

interface ServerWebSocket {
  readonly data: any;
  readonly readyState: number;
  readonly remoteAddress: string;
  readonly subscriptions: string[];
  send(message: string | ArrayBuffer | Uint8Array, compress?: boolean): number;
  close(code?: number, reason?: string): void;
  subscribe(topic: string): void;
  unsubscribe(topic: string): void;
  publish(topic: string, message: string | ArrayBuffer | Uint8Array): void;
  isSubscribed(topic: string): boolean;
  cork(cb: (ws: ServerWebSocket) => void): void;
}

Iniciar un servidor WebSocket ​

Una API diseñada para velocidad ​

Enviar mensajes ​

Encabezados ​

Datos contextuales ​

Pub/Sub ​

Compresión ​

Contrapresión ​

Tiempos de espera y límites ​

Conectarse a un servidor Websocket ​

Referencia ​

Iniciar un servidor WebSocket

Una API diseñada para velocidad

Enviar mensajes

Encabezados

Datos contextuales

Pub/Sub

Compresión

Contrapresión

Tiempos de espera y límites

Conectarse a un servidor `Websocket`

Referencia