首页
/ Playwright WebSocketRoute 深度指南：WebSocket 路由、Mock、拦截与源码实现

Playwright WebSocketRoute 深度指南：WebSocket 路由、Mock、拦截与源码实现

2026-09-06 12:05:44作者：董斯意
playwright
Playwright is a framework for Web Testing and Automation. It allows testing Chromium, Firefox and WebKit with a single API.
项目地址：https://gitcode.com/GitHub_Trending/pl/playwright

本文围绕 Playwright 的 WebSocketRoute 类 API（自 v1.48 引入）展开，系统讲解如何通过 page.routeWebSocket() / browserContext.routeWebSocket() 对页面中的 WebSocket 连接进行完整 Mock、消息拦截与双向转发控制，并结合 WebSocketRoute 客户端实现、WebSocketRouteDispatcher 分发器实现 与 route-web-socket 测试套件 剖析其底层事件通道、二进制消息编解码与连接生命周期管理。读完本文，你可以独立搭建可复现的 WebSocket 服务端 Mock，并在真实服务器上实现消息改写、阻断与协议协商控制。

一、WebSocketRoute 是什么：路由设置后的“服务端替身”

在 Playwright 中，每当通过 Page.routeWebSocket 或 BrowserContext.routeWebSocket 设置了一条 WebSocket 路由，对应的 WebSocketRoute 对象就会允许你像真正的服务端一样处理该 WebSocket 连接。原始 API 文档见 class-websocketroute.md。

两条入口方法的签名在源码中定义如下：

await page.routeWebSocket(url, handler);
// url: URLMatch（string 支持 glob / RegExp / function）
// handler: (ws: WebSocketRoute) => void | Promise<void>
  • Java：page.routeWebSocket(url, handler)；Python：page.route_web_socket(url, handler)；.NET：page.RouteWebSocketAsync(url, handler)。

从源码结构看，routeWebSocket 的核心行为是：将新的 WebSocketRouteHandler 以 unshift 插入路由列表头部（后注册的规则优先匹配），随后调用 _updateWebSocketInterceptionPatterns 把 URL 匹配模式下发到浏览器端（见 page.ts 与 browserContext.ts）。BrowserContext 级别的路由则作用于该上下文内所有页面。

二、Mock：不连接真实服务器，模拟整条 WebSocket 通信

默认情况下，被路由的 WebSocket 不会连接到真实服务器。因此你可以完整地 Mock 掉 WebSocket 通信。下面是一个响应 "request" 消息并回复 "response" 的完整示例（继承自原文档，覆盖四种语言）：

await page.routeWebSocket('wss://example.com/ws', ws => {
  ws.onMessage(message => {
    if (message === 'request')
      ws.send('response');
  });
});

page.routeWebSocket("wss://example.com/ws", ws -> {
  ws.onMessage(frame -> {
    if ("request".equals(frame.text()))
      ws.send("response");
  });
});

def message_handler(ws: WebSocketRoute, message: Union[str, bytes]):
  if message == "request":
    ws.send("response")

await page.route_web_socket("wss://example.com/ws", lambda ws: ws.on_message(
    lambda message: message_handler(ws, message)
))

def message_handler(ws: WebSocketRoute, message: Union[str, bytes]):
  if message == "request":
    ws.send("response")

page.route_web_socket("wss://example.com/ws", lambda ws: ws.on_message(
    lambda message: message_handler(ws, message)
))

await page.RouteWebSocketAsync("wss://example.com/ws", ws => {
  ws.OnMessage(frame => {
    if (frame.Text == "request")
      ws.Send("response");
  });
});

为什么不调用 connectToServer 就是 Mock？ 在 network.ts 中可以看到，路由处理函数返回后，客户端会执行 _afterHandle()：只要没有 connectToServer（this._connected 为 false），就调用 ensureOpened 通道命令，让注入到页面里的 mock 脚本把 WebSocket 直接标记为“已打开”状态，从而保证页面里的 ws.send() 不会因没有真实连接而抛错：

async _afterHandle() {
  if (this._connected)
    return;
  // Ensure that websocket is "open" and can send messages without an actual server connection.
  // If this happens after the page has been closed, ignore the error.
  await this._channel.ensureOpened({}, kNoTimeout).catch(() => {});
}

这也是官方文档所述“Playwright assumes that WebSocket will be mocked, and opens the WebSocket inside the page automatically”的实现依据。

处理 JSON 消息

Mock 场景中最常见的是结构化消息，原文档给出了 JSON 处理示例：

await page.routeWebSocket('wss://example.com/ws', ws => {
  ws.onMessage(message => {
    const json = JSON.parse(message);
    if (json.request === 'question')
      ws.send(JSON.stringify({ response: 'answer' }));
  });
});

page.routeWebSocket("wss://example.com/ws", ws -> {
  ws.onMessage(frame -> {
    JsonObject json = new JsonParser().parse(frame.text()).getAsJsonObject();
    if ("question".equals(json.get("request").getAsString())) {
      Map<String, String> result = new HashMap();
      result.put("response", "answer");
      ws.send(gson.toJson(result));
    }
  });
});

def message_handler(ws: WebSocketRoute, message: Union[str, bytes]):
  json_message = json.loads(message)
  if json_message["request"] == "question":
    ws.send(json.dumps({ "response": "answer" }))

await page.route_web_socket("wss://example.com/ws", lambda ws: ws.on_message(
    lambda message: message_handler(ws, message)
))

await page.RouteWebSocketAsync("wss://example.com/ws", ws => {
  ws.OnMessage(frame => {
    using var jsonDoc = JsonDocument.Parse(frame.Text);
    JsonElement root = jsonDoc.RootElement;
    if (root.TryGetProperty("request", out JsonElement requestElement) && requestElement.GetString() == "question")
    {
      var response = new Dictionary<string, string> { ["response"] = "answer" };
      string jsonResponse = JsonSerializer.Serialize(response);
      ws.Send(jsonResponse);
    }
  });
});

语言差异提示：JS 与 Python 的 onMessage handler 直接接收原始消息（string | bytes/Buffer）；而 Java 与 C# 的 handler 接收的是 WebSocketFrame 对象，需要调用 frame.text() / frame.Text 取文本载荷、frame.binary() / frame.Binary 取二进制载荷，其定义见 class-websocketframe.md（该类自 v1.9 起存在，仅用于 C# 与 Java）。

三、Intercepting：connectToServer 之后的双向转发与拦截

如果你希望连接真实服务器，但想在中间截获并修改或阻断消息，可以调用 WebSocketRoute.connectToServer。它返回一个“服务端侧”的 WebSocketRoute 实例，你可以向它发消息或处理来自服务器的消息。

修改页面发给服务器的消息

以下示例改写页面上行消息，而服务器下行消息保持默认转发：

await page.routeWebSocket('/ws', ws => {
  const server = ws.connectToServer();
  ws.onMessage(message => {
    if (message === 'request')
      server.send('request2');
    else
      server.send(message);
  });
});

page.routeWebSocket("/ws", ws -> {
  WebSocketRoute server = ws.connectToServer();
  ws.onMessage(frame -> {
    if ("request".equals(frame.text()))
      server.send("request2");
    else
      server.send(frame.text());
  });
});

def message_handler(server: WebSocketRoute, message: Union[str, bytes]):
  if message == "request":
    server.send("request2")
  else:
    server.send(message)

def handler(ws: WebSocketRoute):
  server = ws.connect_to_server()
  ws.on_message(lambda message: message_handler(server, message))

await page.route_web_socket("/ws", handler)

def message_handler(server: WebSocketRoute, message: Union[str, bytes]):
  if message == "request":
    server.send("request2")
  else:
    server.send(message)

def handler(ws: WebSocketRoute):
  server = ws.connect_to_server()
  ws.on_message(lambda message: message_handler(server, message))

page.route_web_socket("/ws", handler)

await page.RouteWebSocketAsync("/ws", ws => {
  var server = ws.ConnectToServer();
  ws.OnMessage(frame => {
    if (frame.Text == "request")
      server.Send("request2");
    else
      server.Send(frame.Text);
  });
});

转发语义：onMessage 会“接管”某个方向

连接服务器之后，所有消息默认在页面与服务器之间自动双向转发。但存在两条明确的规则（原文档核心要点，也与源码实现一一对应）：

  1. 在原始路由上调用 onMessage 后，页面到服务器的消息不再自动转发，必须由 handler 自行处理（例如调用 server.send 显式转发）；
  2. 在服务端路由上调用 onMessage 后，服务器到页面的消息不再自动转发，必须由 handler 接管。

这两条规则可以直接在 network.ts 的通道事件处理中得到印证：

this._channel.on('messageFromPage', ({ message, isBase64 }) => {
  if (this._onPageMessage)
    this._onPageMessage(isBase64 ? Buffer.from(message, 'base64') : message);
  else if (this._connected)
    this._channel.sendToServer({ message, isBase64 }, kNoTimeout).catch(() => {});
});

this._channel.on('messageFromServer', ({ message, isBase64 }) => {
  if (this._onServerMessage)
    this._onServerMessage(isBase64 ? Buffer.from(message, 'base64') : message);
  else
    this._channel.sendToPage({ message, isBase64 }, kNoTimeout).catch(() => {});
});

即：有 handler 走 handler，没有 handler 且已连接则自动转发到对端。关闭事件（closePage/closeServer）也遵循同样的“handler 优先，否则自动向对端传播”模式。

双向阻断消息

下面这个示例在两个方向都调用 onMessage，因此完全不保留自动转发，用于阻断特定消息：

await page.routeWebSocket('/ws', ws => {
  const server = ws.connectToServer();
  ws.onMessage(message => {
    if (message !== 'blocked-from-the-page')
      server.send(message);
  });
  server.onMessage(message => {
    if (message !== 'blocked-from-the-server')
      ws.send(message);
  });
});

page.routeWebSocket("/ws", ws -> {
  WebSocketRoute server = ws.connectToServer();
  ws.onMessage(frame -> {
    if (!"blocked-from-the-page".equals(frame.text()))
      server.send(frame.text());
  });
  server.onMessage(frame -> {
    if (!"blocked-from-the-server".equals(frame.text()))
      ws.send(frame.text());
  });
});

def ws_message_handler(server: WebSocketRoute, message: Union[str, bytes]):
  if message != "blocked-from-the-page":
    server.send(message)

def server_message_handler(ws: WebSocketRoute, message: Union[str, bytes]):
  if message != "blocked-from-the-server":
    ws.send(message)

def handler(ws: WebSocketRoute):
  server = ws.connect_to_server()
  ws.on_message(lambda message: ws_message_handler(server, message))
  server.on_message(lambda message: server_message_handler(ws, message))

await page.route_web_socket("/ws", handler)

await page.RouteWebSocketAsync("/ws", ws => {
  var server = ws.ConnectToServer();
  ws.OnMessage(frame => {
    if (frame.Text != "blocked-from-the-page")
      server.Send(frame.Text);
  });
  server.OnMessage(frame => {
    if (frame.Text != "blocked-from-the-server")
      ws.Send(frame.Text);
  });
});

注意 Python 侧的函数参数命名要刻意区分两个 WebSocketRoute（原始路由与服务端路由），否则闭包内会引用错对象。

四、API 方法逐项详解

以下各方法均自 v1.48 引入（protocols 除外，见第六节），与 class-websocketroute.md 保持一一对应。

close(options)：关闭 WebSocket 连接的一侧

WebSocketRoute.close 关闭 WebSocket 连接的一侧，接受两个可选参数：

参数 类型 说明
code int 可选的 WebSocket 关闭码
reason string 可选的关闭原因文本

从 源码 看，在原始路由（页面侧）上调用 close() 实际会向 closePage 通道发送 { code, reason, wasClean: true }；而在服务端路由上调用则走 closeServer 通道。

connectToServer()：建立到真实服务器的连接

  • 返回：服务端侧的 WebSocketRoute 实例，可向它发消息或处理来自服务器的消息。
  • 默认被路由的 WebSocket 不连接服务器，以便 Mock 整条通信；调用此方法后即连接真实服务器。
  • 连接后的转发行为：服务器消息自动转发到页面（除非在服务端路由上调用了 onMessage）；页面 WebSocket.send() 的消息自动转发到服务器（除非在原始路由上调用了 onMessage）。
  • 重复调用会抛错：源码 中 connectToServer() 先检查 this._connected，若已连接则抛出 Already connected to the server，对应测试用例 should throw when connecting twice（route-web-socket.spec.ts）。
  • 服务端路由上的 connectToServer 不可用，客户端会直接抛出 connectToServer must be called on the page-side WebSocketRoute（见 network.ts），这是对 API 使用方向的源码级保护。

onMessage(handler)：接管某一方向的消息

  • 在原始路由上调用：处理页面发出的消息。你可以用 send 直接应答页面、用 connectToServer 返回的服务端连接转发、或做其他处理。
  • 一旦调用，消息不再自动转发到服务器或页面，需要手动调用 send。
  • 重复调用会覆盖旧 handler（源码中 handler 以单一字段 _onPageMessage / _onServerMessage 存储，直接赋值覆盖，见 network.ts）。
  • handler 签名按语言区分：
    • JS / Python：handler(message: string | Buffer) => any，直接接收原始消息；
    • C# / Java：handler(frame: WebSocketFrame)，通过 text() / binary() 区分文本帧与二进制帧。

send(message)：向 WebSocket 发送消息

message 类型为 string | Buffer。在原始路由上调用会把消息发给页面；在 connectToServer 返回的服务端路由上调用则发给服务器。

二进制编解码在 源码 中处理：字符串消息以 isBase64: false 直接透传；Buffer 消息先 toString('base64') 再发送，接收侧再按 isBase64 标志还原为 Buffer。这意味着 send 天然支持文本帧与二进制帧两种载荷。

onClose(handler)：处理 WebSocket 关闭

onClose 用于处理 WebSocket.close 事件。默认行为是：连接任一侧（页面或服务器）关闭时，另一侧也会被自动关闭；一旦设置了 onClose handler，默认的关闭转发被禁用，由 handler 自行决定是否向对端传播关闭。handler 接收可选的关闭码与关闭原因：

  • JS / Python：(code?: int, reason?: string) => any；
  • Java：(Integer code, String reason)（可能为 null）；
  • C#：(int? code, string? reason)。

url()：页面中创建的 WebSocket 的 URL

返回字符串，即路由命中的 WebSocket 实际 URL。实现上直接读取初始化数据 this._initializer.url（见 network.ts）。

五、路由匹配机制与底层注入原理

客户端：WebSocketRouteHandler 与模式匹配

WebSocketRouteHandler 封装了 routeWebSocket 的 URL 与 handler：

  • glob 模式提前校验：构造函数中若 url 是字符串，会立即调用 resolveGlobToRegexPattern 验证，使非法模式在 page.routeWebSocket() 调用点就抛错，而不是延迟到连接时才失败（源码注释明确说明了这一设计意图）；
  • baseURL 支持：相对 URL（如 /ws）会结合 context 的 baseURL 解析，测试用例 should work with relative WebSocket URL 与 should work with baseURL 覆盖了这一行为；
  • 模式序列化下发：prepareInterceptionPatterns 把所有 handler 的 URL 模式序列化为浏览器端拦截模式；若任一 handler 的模式为空（匹配全部），则整体下发 **/*。

服务端：WebSocketRouteDispatcher 与页面注入

webSocketRouteDispatcher.ts 揭示了完整的拦截链路：

  1. 安装拦截（install 静态方法）：在 BrowserContext 上通过 exposeBinding('__pwWebSocketBinding') 注入页面通信入口，并通过 addInitScript 注入由 @injected/webSocketMock 生成的 mock 脚本（即页面内对 WebSocket 构造函数的劫持）。页面中的 WebSocket 创建事件会携带 id、url、protocols 上报。
  2. 路由归属判断：收到 onCreate 后，服务端先用 matchesPattern 依次尝试页面级与上下文级的拦截模式（_webSocketInterceptionPatterns）；命中则创建 WebSocketRouteDispatcher 并通知客户端派发 handler；未命中则下发 passthrough，让页面直连真实服务器。
  3. 消息转发：页面消息（onMessageFromPage）、服务器消息（onMessageFromServer）、两侧关闭事件（onClosePage / onCloseServer）都通过 binding 上报到 dispatcher，再由客户端 WebSocketRoute 按“handler 优先、否则自动转发”的规则处理。
  4. 执行上下文失效保护：dispatcher 监听页面/框架的 InternalFrameNavigatedToNewDocument、FrameDetached、Page.Close、Page.Crash 事件，一旦发生，认为 mock WebSocket 已无通信能力，会派发 closePage / closeServer（wasClean: true），对应测试 should emit close upon frame navigation、should emit close upon frame detach 与 should not throw after page closure。
  5. 单客户端约束：install 中若发现同一 context 已有其他连接在路由 WebSocket，会抛出 Another client is already routing WebSockets——即一个浏览器上下文的 WebSocket 路由只能由一个 Playwright 连接管理。

六、protocols()：自 v1.60 起的子协议协商

WebSocketRoute.protocols 自 v1.60 引入，返回页面通过 WebSocket 构造函数 第二个参数请求的子协议列表，对应 Sec-WebSocket-Protocol 请求头；未指定任何协议时返回空数组。服务端路由上同样可以读取（源码中 protocols() 在原始路由与服务端侧对象上均实现了相同的 [...this._initializer.protocols] 逻辑，见 network.ts 与 L545-L547）。

典型用法是根据子协议选择处理策略，不支持则用标准关闭码 1002（Protocol Error）关闭：

await page.routeWebSocket('wss://example.com/ws', ws => {
  if (ws.protocols().includes('chat.v2'))
    ws.onMessage(message => ws.send(JSON.stringify({ version: 2, echo: message })));
  else
    ws.close({ code: 1002, reason: 'Unsupported protocol' });
});

page.routeWebSocket("wss://example.com/ws", ws -> {
  if (ws.protocols().contains("chat.v2")) {
    ws.onMessage(frame -> ws.send("v2:" + frame.text()));
  } else {
    ws.close(1002, "Unsupported protocol");
  }
});

async def handler(ws: WebSocketRoute):
  if "chat.v2" in ws.protocols:
    ws.on_message(lambda message: ws.send(f"v2:{message}"))
  else:
    await ws.close(code=1002, reason="Unsupported protocol")

await page.route_web_socket("wss://example.com/ws", handler)

def handler(ws: WebSocketRoute):
  if "chat.v2" in ws.protocols:
    ws.on_message(lambda message: ws.send(f"v2:{message}"))
  else:
    ws.close(code=1002, reason="Unsupported protocol")

page.route_web_socket("wss://example.com/ws", handler)

await page.RouteWebSocketAsync("wss://example.com/ws", ws => {
  if (ws.Protocols.Contains("chat.v2"))
    ws.OnMessage(frame => ws.Send($"v2:{frame.Text}"));
  else
    ws.CloseAsync(new() { Code = 1002, Reason = "Unsupported protocol" });
});

注意 Python 的 protocols 是属性（ws.protocols）而非方法，与其他语言的方法调用形式不同；真实连接时，测试用例 should pass through the required protocol 验证了协议会在握手时正确透传给服务器。

七、测试用例佐证的行为边界

tests/library/route-web-socket.spec.ts 为上述文档语义提供了完整的行为验证，可以据此确认适用边界：

测试用例 验证的语义
should work with text message 文本消息往返
should work with binaryType=blob / arraybuffer 页面不同 binaryType 下二进制消息正确到达 handler
should work when connection errors out 上游连接失败时的行为
should work with client-side close 页面侧 ws.close 的关闭传播
should observe upstream handshake failure when connectToServer is used 拦截模式下握手失败可被观察
should observe multiple concurrent routed WebSockets with connectToServer 多个并发路由连接互不串扰（依赖 dispatcher 的 _idToDispatcher 映射）
should pattern match URL 模式匹配（glob）
should emit close upon frame navigation / frame detach 帧导航/分离时自动派发 close
should route on context BrowserContext.routeWebSocket 生效于上下文内页面
should not throw when connecting twice 的对应用例 二次 connectToServer 抛错
should work with baseURL / baseURL regardless of scheme casing 相对 URL 与大小写不敏感匹配
should expose protocols to the route handler / on server-side route protocols 在两侧均可用

此外，tests/library/multiclient.spec.ts 验证了多客户端场景下路由的共存约束，与 dispatcher 中 “Another client is already routing WebSockets” 的源码约束相呼应。

八、实用建议与限制总结

  • Mock 与拦截的切换点只有一个：是否在 handler 中调用 connectToServer。不调用则 Playwright 自动把页面内 WebSocket 置为打开态（ensureOpened），整条链路脱离真实服务器；调用了则进入真实握手，默认双向转发，onMessage / onClose 按方向逐步接管。
  • handler 必须显式转发：一旦在某一方向设置了 onMessage，该方向消息即停止自动转发，忘记调用对端 send 会导致消息静默丢失（send 内部 .catch(() => {}) 吞掉了传输错误，不会主动报错）。
  • 适用前提：routeWebSocket 需要 v1.48+；protocols 需要 v1.60+；Java/C# 的消息载荷需经 WebSocketFrame 间接读取；同一浏览器上下文的 WebSocket 路由仅能由一个 Playwright 客户端连接建立。
  • 生命周期注意：页面导航、帧分离或页面关闭都会触发路由侧的关闭派发，编写测试时不应假设导航后 mock 连接仍然存活。

综合来看，WebSocketRoute 以“页面侧路由 + 服务端侧路由”的双实例模型，配合按方向可接管的转发规则，覆盖了从纯 Mock 到真实链路上的细粒度拦截的完整光谱，是 Playwright 处理实时通信类页面测试与自动化（如聊天协议、行情推送、协同编辑同步）的核心能力。

playwright
Playwright is a framework for Web Testing and Automation. It allows testing Chromium, Firefox and WebKit with a single API.
项目地址：https://gitcode.com/GitHub_Trending/pl/playwright
登录后查看全文
热门项目推荐
相关项目推荐

项目优选

收起
kernelkernel
deepin linux kernel
C
33
18
pytorchpytorch
作为 Ascend for PyTorch 社区的核心组件，TorchNPU 是昇腾专为 PyTorch 打造的深度学习适配插件，使 PyTorch 框架能够直接调用昇腾 NPU，为开发者提供昇腾 AI 处理器的超强算力。
Python
856
1.35 K
kernelkernel
openEuler内核是openEuler操作系统的核心，既是系统性能与稳定性的基石，也是连接处理器、设备与服务的桥梁。
C
529
592
leetcodeleetcode
🔥LeetCode solutions in any programming language | 多种编程语言实现 LeetCode、《剑指 Offer（第 2 版）》、《程序员面试金典（第 6 版）》题解
Markdown
77
23
ops-transformerops-transformer
本项目是CANN提供的transformer类大模型算子库，实现网络在NPU上加速计算。
C++
1.13 K
2.74 K
docsdocs
暂无描述
Markdown
897
5.79 K
jiuwenswarmjiuwenswarm
JiuwenSwarm 是一款基于openJiuwen开发的智能AI Agent，它能够将大语言模型的强大能力，通过你日常使用的各类通讯应用，直接延伸至你的指尖。
Python
3.58 K
1.01 K
cann-learning-hubcann-learning-hub
CANN 学习中心仓，支持在线互动运行、边学边练，提供教程、示例与优化方案，一站式助力昇腾开发者快速上手。
Jupyter Notebook
1.01 K
513
ops-nnops-nn
本项目是CANN提供的神经网络类计算算子库，实现网络在NPU上加速计算。
C++
913
1.83 K
openGauss-serveropenGauss-server
openGauss kernel ~ openGauss is an open source relational database management system
C++
213
314