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