WebSocketRoute
每当通过 page.route_web_socket() 或 browser_context.route_web_socket() 设置 WebSocket
路由时,WebSocketRoute
对象允许你像真实服务器一样处理 WebSocket。
模拟(Mocking)
默认情况下,被路由的 WebSocket 不会连接到服务器。这样,你可以完全模拟 WebSocket 的通信。下面是一个示例,当收到 "request"
时响应 "response"
。
- 同步
- 异步
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)
))
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)
))
由于我们没有在 WebSocket 路由处理函数中调用 web_socket_route.connect_to_server,Playwright 会认为 WebSocket 将被模拟,并自动在页面内打开 WebSocket。
下面是另一个处理 JSON 消息的示例:
- 同步
- 异步
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" }))
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]):
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)
))
拦截(Intercepting)
另外,你也可以选择连接到真实服务器,但在中间拦截消息,对其进行修改或阻止。调用 web_socket_route.connect_to_server 会返回一个服务端的 WebSocketRoute
实例,你可以向其发送消息,或处理接收到的消息。
下面是一个示例,对页面发送到服务器的部分消息进行了修改。从服务器发送到页面的消息保持不变,依赖默认的转发。
- 同步
- 异步
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)
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)
连接到服务器后,页面与服务器之间的所有消息默认都会被自动转发。
但是,如果你在原始路由上调用 web_socket_route.on_message(),则页面到服务器的消息将不会再被自动转发,而需要由 handler 处理。
同样地,在服务端 WebSocket 上调用 web_socket_route.on_message() 会停止服务器到页面的消息自动转发,需要由 handler 进行处理。
下面的示例会阻止双向的部分消息。由于在两个方向都调用了 web_socket_route.on_message(),因此不会有任何自动转发。
- 同步
- 异步
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))
page.route_web_socket("/ws", handler)
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)
方法
close
新增于: v1.48关闭 WebSocket 连接的一侧。
用法
web_socket_route.close()
web_socket_route.close(**kwargs)
参数
返回值
on_close
新增于: v1.48允许处理 WebSocket.close
。
默认情况下,无论是在页面还是在服务器端关闭连接的一侧,都会关闭另一侧。但是,当设置了 web_socket_route.on_close() 处理函数后,默认的关闭转发将被禁用,需要由处理函数自行处理。
用法
web_socket_route.on_close(handler)
参数
on_message
新增于: v1.48该方法允许处理由 WebSocket 发送的消息,无论是来自页面还是来自服务器。
当在原始 WebSocket 路由上调用时,该方法处理从页面发送的消息。你可以通过 web_socket_route.send() 回复这些消息,或者将其转发到通过 web_socket_route.connect_to_server 返回的服务端连接,或进行其他操作。
一旦调用该方法,消息将不会被自动转发到服务器或页面——你需要通过调用 web_socket_route.send() 手动转发。更多细节请参考顶部的示例。
再次调用该方法会用新的处理函数覆盖之前的处理函数。
用法
web_socket_route.on_message(handler)
参数
send
新增于: v1.48向 WebSocket 发送消息。在原始 WebSocket 上调用时,消息会发送到页面。在 web_socket_route.connect_to_server 的返回结果上调用时,消息会发送到服务器。更多细节请参考顶部的示例。
用法
web_socket_route.send(message)
参数
属性
connect_to_server
新增于: v1.48默认情况下,被路由的 WebSocket 不会连接到服务器,因此你可以完全模拟 WebSocket 通信。该方法会连接到实际的 WebSocket 服务器,并返回服务端的 WebSocketRoute 实例,使你能够从服务器发送和接收消息。
连接到服务器后:
- 从服务器接收到的消息会自动转发到页面中的 WebSocket,除非在服务端
WebSocketRoute
上调用了 web_socket_route.on_message()。 - 页面中通过
WebSocket.send()
发送的消息会自动转发到服务器,除非在原始WebSocketRoute
上调用了 web_socket_route.on_message()。
更多细节请参考顶部的示例。
用法
web_socket_route.connect_to_server
返回值
url
新增于: v1.48页面中创建的 WebSocket 的 URL。
用法
web_socket_route.url
返回值