WebSocketRoute

WebSocketルートがpage.route_web_socket()またはbrowser_context.route_web_socket()で設定されるたびに、WebSocketRouteオブジェクトは、実際のサーバーが行うようにWebSocketを処理できます。

モック

デフォルトでは、ルーティングされた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)
))

インターセプト

あるいは、実際のサーバーに接続しながら、メッセージを途中でインターセプトして変更またはブロックしたい場合があるかもしれません。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()を呼び出すと、ページからサーバーへのメッセージは**転送されなくなります**。代わりに、ハンドラーによって処理される必要があります。

同様に、サーバー側のWebSocketでweb_socket_route.on_message()を呼び出すと、サーバーからページへの**メッセージの転送が停止され**、ハンドラーがそれらを処理する必要があります。

次の例では、両方向のメッセージの一部をブロックします。両方向で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 webSocketRoute.close

WebSocket接続の一方を閉じます。

使用方法

web_socket_route.close()
web_socket_route.close(**kwargs)

引数

• code int (オプション)#

  オプションの閉じるコード。

• reason str (オプション)#

  オプションの閉じる理由。

戻り値

• NoneType#

---

on_close

追加バージョン: v1.48 webSocketRoute.on_close

WebSocket.closeを処理できます。

デフォルトでは、ページまたはサーバーのいずれかで接続の一方を閉じると、他方も閉じられます。しかし、web_socket_route.on_close()ハンドラーが設定されている場合、クロージャのデフォルト転送は無効になり、ハンドラーがそれを処理する必要があります。

使用方法

web_socket_route.on_close(handler)

引数

• handler Callable[int | undefined]:Promise[Any] | Any#

  WebSocketのクロージャを処理する関数。オプションの閉じるコードとオプションの閉じる理由を受け取ります。

---

on_message

追加バージョン: v1.48 webSocketRoute.on_message

このメソッドは、WebSocketによって、ページから、またはサーバーから送信されるメッセージを処理できます。

元のWebSocketルートで呼び出された場合、このメソッドはページから送信されたメッセージを処理します。これらのメッセージは、web_socket_route.send()で応答したり、web_socket_route.connect_to_serverによって返されるサーバー側接続に転送したり、その他の方法で処理できます。

このメソッドが呼び出されると、メッセージは自動的にサーバーまたはページに転送されません。`web_socket_route.send()`を呼び出して手動で転送する必要があります。詳細については、冒頭の例を参照してください。

このメソッドを再度呼び出すと、ハンドラーが新しいものに上書きされます。

使用方法

web_socket_route.on_message(handler)

引数

• handler Callable[str]:Promise[Any] | Any#

  メッセージを処理する関数。

---

send

追加バージョン: v1.48 webSocketRoute.send

WebSocketにメッセージを送信します。元のWebSocketで呼び出された場合、メッセージをページに送信します。web_socket_route.connect_to_serverの結果に対して呼び出された場合、メッセージをサーバーに送信します。詳細については、冒頭の例を参照してください。

使用方法

web_socket_route.send(message)

引数

• message str | bytes#

  送信するメッセージ。

---

プロパティ

connect_to_server

追加バージョン: v1.48 webSocketRoute.connect_to_server

デフォルトでは、ルーティングされた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

戻り値

• WebSocketRoute#

---

url

追加バージョン: v1.48 webSocketRoute.url

ページで作成されたWebSocketのURL。

使用方法

web_socket_route.url

戻り値

• str#