WebSocketRoute
WebSocket
ルートが page.route_web_socket() または browser_context.route_web_socket() で設定されるたびに、WebSocketRoute
オブジェクトは、実際のサーバーが行うように、WebSocket を処理できます。
モック
デフォルトでは、ルーティングされた WebSocket はサーバーに接続しません。これにより、WebSocket を介した通信全体をモックできます。以下は、"request"
に "response"
で応答する例です。
- Sync
- Async
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 メッセージを処理する別の例を次に示します。
- Sync
- Async
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)
))
インターセプト
あるいは、実際のサーバーに接続し、途中でメッセージをインターセプトして変更またはブロックすることもできます。web_socket_route.connect_to_server を呼び出すと、メッセージを送信したり、受信メッセージを処理したりできるサーバー側の WebSocketRoute
インスタンスが返されます。
以下は、ページからサーバーに送信された一部のメッセージを変更する例です。サーバーからページに送信されたメッセージは、デフォルトの転送に依存して、そのまま残されます。
- Sync
- Async
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() を呼び出すため、自動転送はまったくありません。
- Sync
- Async
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.48WebSocket 接続の片側を閉じます。
使用法
web_socket_route.close()
web_socket_route.close(**kwargs)
引数
戻り値
on_close
追加: v1.48WebSocket.close
を処理できます。
デフォルトでは、ページまたはサーバーのいずれかで接続の片側を閉じると、もう一方の側も閉じられます。ただし、web_socket_route.on_close() ハンドラーが設定されている場合、クロージャーのデフォルトの転送は無効になり、ハンドラーがそれを処理する必要があります。
使用法
web_socket_route.on_close(handler)
引数
-
handler
Callable[int | [undefined]]:Promise[Any] | Any#WebSocket のクローズを処理する関数。オプションの close コード とオプションの close 理由 を受け取ります。
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.48WebSocket にメッセージを送信します。元の WebSocket で呼び出されると、メッセージをページに送信します。web_socket_route.connect_to_server の結果で呼び出されると、メッセージをサーバーに送信します。詳細については、上部の例を参照してください。
使用法
web_socket_route.send(message)
引数
プロパティ
connect_to_server
追加: v1.48デフォルトでは、ルーティングされた WebSocket はサーバーに接続しないため、WebSocket 通信全体をモックできます。このメソッドは、実際の WebSocket サーバーに接続し、サーバー側の WebSocketRoute インスタンスを返し、サーバーとの間でメッセージを送信および受信できるようにします。
サーバーに接続すると
- サーバーから受信したメッセージは、サーバー側の
WebSocketRoute
で web_socket_route.on_message() が呼び出されない限り、ページ内の WebSocket に自動的に転送されます。 - ページ内の
WebSocket.send()
呼び出しによって送信されたメッセージは、元のWebSocketRoute
で web_socket_route.on_message() が呼び出されない限り、サーバーに自動的に転送されます。
詳細については、上部の例を参照してください。
使用法
web_socket_route.connect_to_server
戻り値
url
追加: v1.48ページで作成された WebSocket の URL。
使用法
web_socket_route.url
戻り値