bun-websocket-server

Bun WebSocket 服务端

Bun 内置了 WebSocket 支持，并与 Bun.serve() 集成。

快速开始

const server = Bun.serve({
  fetch(req, server) {
    // Upgrade to WebSocket
    if (server.upgrade(req)) {
      return; // Upgraded successfully
    }
    return new Response("Not a WebSocket request", { status: 400 });
  },
  websocket: {
    open(ws) {
      console.log("Client connected");
    },
    message(ws, message) {
      console.log("Received:", message);
      ws.send(`Echo: ${message}`);
    },
    close(ws) {
      console.log("Client disconnected");
    },
  },
});

console.log(`WebSocket server running on ws://localhost:${server.port}`);

WebSocket 处理程序

Bun.serve({
  fetch(req, server) {
    server.upgrade(req);
  },
  websocket: {
    // Client connected
    open(ws) {
      console.log("New connection");
    },

    // Message received
    message(ws, message) {
      // message is string | Buffer
      if (typeof message === "string") {
        console.log("Text:", message);
      } else {
        console.log("Binary:", message);
      }
    },

    // Connection closed
    close(ws, code, reason) {
      console.log(`Closed: ${code} - ${reason}`);
    },

    // Drain event (buffer flushed)
    drain(ws) {
      console.log("Buffer drained");
    },

    // Ping received
    ping(ws, data) {
      // Pong sent automatically
    },

    // Pong received
    pong(ws, data) {
      console.log("Pong received");
    },
  },
});

发送消息

websocket: {
  message(ws, message) {
    // Send text
    ws.send("Hello");

    // Send JSON
    ws.send(JSON.stringify({ type: "greeting", data: "Hello" }));

    // Send binary
    ws.send(new Uint8Array([1, 2, 3]));
    ws.send(Buffer.from("binary data"));

    // Send with compression
    ws.send("compressed message", true);

    // Check if buffer is full
    const bufferedAmount = ws.send("data");
    if (bufferedAmount > 1024 * 1024) {
      console.log("Buffer getting full");
    }
  },
}

为连接附加数据

interface UserData {
  id: string;
  name: string;
  joinedAt: Date;
}

Bun.serve<UserData>({
  fetch(req, server) {
    const url = new URL(req.url);
    const userId = url.searchParams.get("userId");

    // Attach data during upgrade
    server.upgrade(req, {
      data: {
        id: userId,
        name: "User " + userId,
        joinedAt: new Date(),
      },
    });
  },
  websocket: {
    open(ws) {
      // Access attached data
      console.log(`${ws.data.name} connected`);
    },
    message(ws, message) {
      console.log(`${ws.data.name}: ${message}`);
    },
  },
});

发布/订阅（主题）

Bun.serve({
  fetch(req, server) {
    const url = new URL(req.url);
    const room = url.searchParams.get("room") || "general";

    server.upgrade(req, {
      data: { room },
    });
  },
  websocket: {
    open(ws) {
      // Subscribe to a topic
      ws.subscribe(ws.data.room);

      // Publish to topic (excludes sender)
      ws.publish(ws.data.room, `User joined ${ws.data.room}`);
    },
    message(ws, message) {
      // Broadcast to all in room (excludes sender)
      ws.publish(ws.data.room, message);
    },
    close(ws) {
      // Unsubscribe (automatic on close)
      ws.unsubscribe(ws.data.room);
      ws.publish(ws.data.room, "User left");
    },
  },
});

向所有客户端广播

Bun.serve({
  fetch(req, server) {
    server.upgrade(req);
  },
  websocket: {
    open(ws) {
      // Subscribe to global topic
      ws.subscribe("global");
    },
    message(ws, message) {
      // Broadcast to ALL clients including sender
      server.publish("global", message);
    },
  },
});

服务端级别的发布

const server = Bun.serve({
  fetch(req, server) {
    const url = new URL(req.url);

    // HTTP endpoint to publish
    if (url.pathname === "/broadcast") {
      const message = url.searchParams.get("msg");
      server.publish("global", message);
      return new Response("Broadcasted");
    }

    server.upgrade(req);
  },
  websocket: {
    open(ws) {
      ws.subscribe("global");
    },
  },
});

// Can also publish from outside fetch
setInterval(() => {
  server.publish("global", `Server time: ${new Date().toISOString()}`);
}, 5000);

WebSocket 选项

Bun.serve({
  websocket: {
    // Max message size (default 16MB)
    maxPayloadLength: 1024 * 1024, // 1MB

    // Idle timeout in seconds (default 120)
    idleTimeout: 60,

    // Backpressure limit
    backpressureLimit: 1024 * 1024,

    // Enable compression
    perMessageDeflate: true,
    // Or with options
    perMessageDeflate: {
      compress: "shared",
      decompress: "shared",
    },

    // Send/receive pings
    sendPings: true,

    // Handlers
    open(ws) {},
    message(ws, message) {},
    close(ws) {},
  },
});

客户端连接

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

ws.onopen = () => {
  ws.send("Hello Server!");
};

ws.onmessage = (event) => {
  console.log("Received:", event.data);
};

ws.onclose = () => {
  console.log("Disconnected");
};

身份验证

Bun.serve({
  fetch(req, server) {
    // Verify auth before upgrade
    const token = req.headers.get("Authorization");

    if (!verifyToken(token)) {
      return new Response("Unauthorized", { status: 401 });
    }

    const user = decodeToken(token);
    server.upgrade(req, {
      data: { userId: user.id },
    });
  },
  websocket: {
    open(ws) {
      console.log(`Authenticated user ${ws.data.userId} connected`);
    },
  },
});

常见错误

错误	原因	解决方法
Upgrade failed	请求无效	检查升级请求头
Connection closed	客户端断开连接	在关闭处理程序中处理
Message too large	超出 maxPayloadLength	提高限制或分块传输数据
Backpressure	客户端速度较慢	检查缓冲区，并等待排空

常见模式

聊天室

Bun.serve({
  fetch(req, server) {
    const url = new URL(req.url);
    const username = url.searchParams.get("user") || "Anonymous";

    server.upgrade(req, {
      data: { username },
    });
  },
  websocket: {
    open(ws) {
      ws.subscribe("chat");
      ws.publish("chat", `${ws.data.username} joined`);
    },
    message(ws, message) {
      ws.publish("chat", `${ws.data.username}: ${message}`);
    },
    close(ws) {
      ws.publish("chat", `${ws.data.username} left`);
    },
  },
});

何时加载参考资料

在以下情况下加载 references/compression.md：

• perMessageDeflate 配置
• 压缩调优
• 二进制消息处理

在以下情况下加载 references/scaling.md：

• 多个服务端实例
• Redis 发布/订阅集成
• 水平扩展