WebSockets
Bun 中的服务器端 WebSockets
Bun.serve() 支持服务器端 WebSockets，具备实时压缩、TLS 支持以及 Bun 原生的发布-订阅 API。
启动 WebSocket 服务器
以下服务器使用 Bun.serve 构建，会在 fetch 处理程序中将每个传入请求升级为 WebSocket 连接。你可以在 websocket 参数中声明套接字处理程序。
server.tsBun.serve({
fetch(req, server) {
// 将请求升级为 WebSocket
if (server.upgrade(req)) {
return; // 不返回 Response
}
return new Response("升级失败", { status: 500 });
},
websocket: {}, // 处理程序
});
Bun 支持以下 WebSocket 事件处理程序：
server.tsBun.serve({
fetch(req, server) {}, // 升级逻辑
websocket: {
message(ws, message) {}, // 接收到消息
open(ws) {}, // 连接打开
close(ws, code, message) {}, // 连接关闭
drain(ws) {}, // 套接字准备好接收更多数据
},
});
为速度设计的 API
在 Bun 中，你只需为每个服务器声明一次处理程序，而不是为每个套接字分别声明。
你可以将单个 WebSocketHandler 对象传递给 Bun.serve()，其中包含 open、message、close、drain 和 error 方法。这种设计不同于客户端的 WebSocket 类，后者扩展了 EventTarget（onmessage、onopen、onclose）。
客户端通常只会打开少量套接字连接，因此基于事件的 API 在客户端很适用。
但服务器端往往会打开 大量 连接，这意味着：
- 为每个连接添加/移除事件监听器所花费的时间会累积起来
- 为每个连接存储回调函数引用会占用额外内存
- 通常，人们会为每个连接创建新函数，这同样意味着更多的内存开销
在所有连接之间重复使用同一个处理程序对象，可以避免这两项开销。
每个处理程序的第一个参数都是负责处理该事件的 ServerWebSocket 实例。ServerWebSocket 类是一个快速的、原生 Bun 实现的 WebSocket，并提供了一些额外功能。
server.tsBun.serve({
fetch(req, server) {}, // 升级逻辑
websocket: {
message(ws, message) {
ws.send(message); // 将消息原样返回
},
},
});
发送消息
每个 ServerWebSocket 实例都有 .send() 方法用于向客户端发送消息，支持多种输入类型。
server.tsBun.serve({
fetch(req, server) {}, // 升级逻辑
websocket: {
message(ws, message) {
ws.send("Hello world"); // 字符串
ws.send(response.arrayBuffer()); // ArrayBuffer
ws.send(new Uint8Array([1, 2, 3])); // TypedArray | DataView
ws.send(new Blob(["binary"])); // Blob
},
},
});
头信息
升级成功后，Bun 会根据规范发送 101 Switching Protocols 响应。要为此 Response 附加额外的 headers，请将它们传递给 server.upgrade()。
server.tsBun.serve({
fetch(req, server) {
const sessionId = await generateSessionId();
server.upgrade(req, {
headers: { // [!code ++]
"Set-Cookie": `SessionId=${sessionId}`, // [!code ++]
}, // [!code ++]
});
},
websocket: {}, // 处理程序
});
上下文数据
在 .upgrade() 调用中为新的 WebSocket 附加上下文 data。它可在 WebSocket 处理程序中的 ws.data 属性上访问。
要为 ws.data 进行强类型定义，可在 websocket 处理对象里添加 data 属性，从而为所有生命周期钩子中的 ws.data 指定类型。
server.tstype WebSocketData = {
createdAt: number;
channelId: string;
authToken: string;
};
Bun.serve({
fetch(req, server) {
const cookies = new Bun.CookieMap(req.headers.get("cookie")!);
server.upgrade(req, {
// 此对象必须符合 WebSocketData 类型
data: {
createdAt: Date.now(),
channelId: new URL(req.url).searchParams.get("channelId"),
authToken: cookies.get("X-Token"),
},
});
return undefined;
},
websocket: {
// TypeScript：如此指定 ws.data 类型
data: {} as WebSocketData,
// 收到消息时调用的处理函数
async message(ws, message) {
// ws.data 现已正确类型为 WebSocketData
const user = getUserFromToken(ws.data.authToken);
await saveMessageToDatabase({
channel: ws.data.channelId,
message: String(message),
userId: user.id,
});
},
},
});
要从浏览器连接到此服务器，可以创建一个新的 WebSocket。
const socket = new WebSocket("ws://localhost:3000/chat");
socket.addEventListener("message", event => {
console.log(event.data);
});
发布/订阅
Bun 的 ServerWebSocket 包含用于基于主题进行广播的原生发布-订阅 API。你可以使用字符串标识符指定主题。单个套接字可以通过 .subscribe() 订阅主题，并通过 .publish() 向该主题的所有其他订阅者发送消息（不包括自身）。这种基于主题的广播 API 类似于 MQTT 和 Redis Pub/Sub。
server.tsconst server = Bun.serve({
fetch(req, server) {
const url = new URL(req.url);
if (url.pathname === "/chat") {
console.log(`开始升级!`);
const username = getUsernameFromReq(req);
const success = server.upgrade(req, { data: { username } });
return success ? undefined : new Response("WebSocket 升级错误", { status: 400 });
}
return new Response("Hello world");
},
websocket: {
// TypeScript：如此指定 ws.data 类型
data: {} as { username: string },
open(ws) {
const msg = `${ws.data.username} 加入了聊天室`;
ws.subscribe("the-group-chat");
server.publish("the-group-chat", msg);
},
message(ws, message) {
// 群聊功能
// 服务器将收到的消息转发给所有人
server.publish("the-group-chat", `${ws.data.username}: ${message}`);
// 查看当前订阅
console.log(ws.subscriptions); // ["the-group-chat"]
},
close(ws) {
const msg = `${ws.data.username} 离开了聊天室`;
ws.unsubscribe("the-group-chat");
server.publish("the-group-chat", msg);
},
},
});
console.log(`监听在 ${server.hostname}:${server.port}`);
调用 .publish(topic, data) 会向主题的所有订阅者发送消息，_但不包括_调用 .publish() 的套接字。要向主题的所有订阅者发送消息，请使用 Server 实例上的 .publish() 方法。
const server = Bun.serve({
websocket: {
// ...
},
});
// 监听某个外部事件
server.publish("the-group-chat", "Hello world");
压缩
使用 perMessageDeflate 参数启用逐消息压缩。
server.tsBun.serve({
websocket: {
perMessageDeflate: true, // [!code ++]
},
});
要压缩单条消息，请将一个 boolean 作为第二个参数传递给 .send()。
ws.send("Hello world", true);
要对压缩特性进行精细控制，请参考参考文档。
背压管理
ServerWebSocket 的 .send(message) 方法返回一个 number 表示操作结果。
-1— 消息已入队，但存在背压（backpressure）0— 由于连接问题，消息已丢弃1+— 发送的字节数
超时和限制
默认情况下，Bun 会关闭处于空闲状态超过 120 秒的 WebSocket 连接。使用 idleTimeout 参数进行配置。
Bun.serve({
fetch(req, server) {}, // 升级逻辑
websocket: {
idleTimeout: 60, // 60 秒 // [!code ++]
},
});
如果收到的消息大于 16 MB，Bun 也会关闭 WebSocket 连接。使用 maxPayloadLength 参数进行配置。
Bun.serve({
fetch(req, server) {}, // 升级逻辑
websocket: {
maxPayloadLength: 1024 * 1024, // 1 MB // [!code ++]
},
});
连接到 WebSocket 服务器
Bun 实现了 WebSocket 类。创建连接至 ws:// 或 wss:// 服务器的 WebSocket 客户端实例，类似于浏览器中的用法。
const socket = new WebSocket("ws://localhost:3000");
// 支持子协议协商
const socket2 = new WebSocket("ws://localhost:3000", ["soap", "wamp"]);
浏览器会将页面上设置的 Cookie 与 WebSocket 升级请求一同发送。这是 WebSocket API 的标准特性。
在 Bun 中，你还可以直接在构造函数中设置自定义请求头。这是 Bun 对 WebSocket 标准的特有扩展。在浏览器中无法使用。
const socket = new WebSocket("ws://localhost:3000", {
headers: {
/* 自定义头部 */
}, // [!code ++]
});
给套接字添加事件监听器的方法：
// 收到消息
socket.addEventListener("message", event => {});
// 连接打开
socket.addEventListener("open", event => {});
// 连接关闭
socket.addEventListener("close", event => {});
// 错误处理
socket.addEventListener("error", event => {});
背压管理
浏览器的 WebSocket API 无法减慢发送速度超过你处理速度的对端：传入的消息会无限制地缓存在内存中。Bun 添加了 pause() 和 resume()，用于暂停和恢复从底层套接字读取数据，从而让发送方感受到 TCP 背压。这是 Bun 特有的扩展。在浏览器中无法使用。
const socket = new WebSocket("ws://localhost:3000");
socket.addEventListener("message", event => {
if (!file.write(event.data)) {
// The file's buffer is full: stop reading until it drains.
socket.pause();
file.once("drain", () => socket.resume());
}
});
pause() 和 resume() 在操作生效时（或连接中的套接字打开后生效时）返回 true，没有可操作的套接字时则返回 false；socket.isPaused 用于报告当前状态。调用 pause() 时已经解码的消息仍可能被传递。ws 包中的 pause()、resume() 和 isPaused 对应相同的方法。
参考
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; // 默认值：16 * 1024 * 1024 = 16 MB
idleTimeout?: number; // 默认值：120（秒）
backpressureLimit?: number; // 默认值：16 * 1024 * 1024 = 16 MB
closeOnBackpressureLimit?: boolean; // 默认值：false
sendPings?: boolean; // 默认值：true
publishToSelf?: boolean; // 默认值：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 | Blob, 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 | Blob, compress?: boolean): number;
close(code?: number, reason?: string): void;
subscribe(topic: string): boolean;
unsubscribe(topic: string): boolean;
publish(topic: string, message: string | ArrayBuffer | Uint8Array | Blob): void;
isSubscribed(topic: string): boolean;
cork(cb: (ws: ServerWebSocket) => void): void;
}