We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
fix: handle cancelled tasks
prevents `InvalidStateError` when the task has been cancelled, here:

```python
callback.future.set_exception(parsed_error)
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `playwright/_impl/_connection.py`
Content:
```
1 # Copyright (c) Microsoft Corporation.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import asyncio
16 import sys
17 import traceback
18 from pathlib import Path
19 from typing import Any, Callable, Dict, List, Optional, Union
20 
21 from greenlet import greenlet
22 from pyee import AsyncIOEventEmitter
23 
24 from playwright._impl._helper import ParsedMessagePayload, parse_error
25 from playwright._impl._transport import Transport
26 
27 
28 class Channel(AsyncIOEventEmitter):
29     def __init__(self, connection: "Connection", guid: str) -> None:
30         super().__init__()
31         self._connection: Connection = connection
32         self._guid = guid
33         self._object: Optional[ChannelOwner] = None
34 
35     async def send(self, method: str, params: Dict = None) -> Any:
36         return await self.inner_send(method, params, False)
37 
38     async def send_return_as_dict(self, method: str, params: Dict = None) -> Any:
39         return await self.inner_send(method, params, True)
40 
41     async def inner_send(
42         self, method: str, params: Optional[Dict], return_as_dict: bool
43     ) -> Any:
44         if params is None:
45             params = {}
46         callback = self._connection._send_message_to_server(self._guid, method, params)
47         result = await callback.future
48         # Protocol now has named return values, assume result is one level deeper unless
49         # there is explicit ambiguity.
50         if not result:
51             return None
52         assert isinstance(result, dict)
53         if return_as_dict:
54             return result
55         if len(result) == 0:
56             return None
57         assert len(result) == 1
58         key = next(iter(result))
59         return result[key]
60 
61     def send_no_reply(self, method: str, params: Dict = None) -> None:
62         if params is None:
63             params = {}
64         self._connection._send_message_to_server(self._guid, method, params)
65 
66 
67 class ChannelOwner(AsyncIOEventEmitter):
68     def __init__(
69         self,
70         parent: Union["ChannelOwner", "Connection"],
71         type: str,
72         guid: str,
73         initializer: Dict,
74     ) -> None:
75         super().__init__(loop=parent._loop)
76         self._loop: asyncio.AbstractEventLoop = parent._loop
77         self._dispatcher_fiber: Any = parent._dispatcher_fiber
78         self._type = type
79         self._guid = guid
80         self._connection: Connection = (
81             parent._connection if isinstance(parent, ChannelOwner) else parent
82         )
83         self._parent: Optional[ChannelOwner] = (
84             parent if isinstance(parent, ChannelOwner) else None
85         )
86         self._objects: Dict[str, "ChannelOwner"] = {}
87         self._channel = Channel(self._connection, guid)
88         self._channel._object = self
89         self._initializer = initializer
90 
91         self._connection._objects[guid] = self
92         if self._parent:
93             self._parent._objects[guid] = self
94 
95     def _wait_for_event_info_before(self, wait_id: str, name: str) -> None:
96         self._connection._send_message_to_server(
97             self._guid,
98             "waitForEventInfo",
99             {
100                 "info": {
101                     "name": name,
102                     "waitId": wait_id,
103                     "phase": "before",
104                     "stack": capture_call_stack(),
105                 }
106             },
107         )
108 
109     def _wait_for_event_info_after(
110         self, wait_id: str, exception: Exception = None
111     ) -> None:
112         info = {"waitId": wait_id, "phase": "after"}
113         if exception:
114             info["error"] = str(exception)
115         self._connection._send_message_to_server(
116             self._guid,
117             "waitForEventInfo",
118             {"info": info},
119         )
120 
121     def _dispose(self) -> None:
122         # Clean up from parent and connection.
123         if self._parent:
124             del self._parent._objects[self._guid]
125         del self._connection._objects[self._guid]
126 
127         # Dispose all children.
128         for object in list(self._objects.values()):
129             object._dispose()
130         self._objects.clear()
131 
132 
133 class ProtocolCallback:
134     def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
135         self.stack_trace: traceback.StackSummary = traceback.StackSummary()
136         self.future = loop.create_future()
137 
138 
139 class RootChannelOwner(ChannelOwner):
140     def __init__(self, connection: "Connection") -> None:
141         super().__init__(connection, "", "", {})
142 
143 
144 class Connection:
145     def __init__(
146         self, dispatcher_fiber: Any, object_factory: Any, driver_executable: Path
147     ) -> None:
148         self._dispatcher_fiber: Any = dispatcher_fiber
149         self._transport = Transport(driver_executable)
150         self._transport.on_message = lambda msg: self._dispatch(msg)
151         self._waiting_for_object: Dict[str, Any] = {}
152         self._last_id = 0
153         self._objects: Dict[str, ChannelOwner] = {}
154         self._callbacks: Dict[int, ProtocolCallback] = {}
155         self._object_factory = object_factory
156         self._is_sync = False
157         self._api_name = ""
158 
159     async def run_as_sync(self) -> None:
160         self._is_sync = True
161         await self.run()
162 
163     async def run(self) -> None:
164         self._loop = asyncio.get_running_loop()
165         self._root_object = RootChannelOwner(self)
166         await self._transport.run()
167 
168     def stop_sync(self) -> None:
169         self._transport.stop()
170         self._dispatcher_fiber.switch()
171 
172     async def stop_async(self) -> None:
173         self._transport.stop()
174         await self._transport.wait_until_stopped()
175 
176     async def wait_for_object_with_known_name(self, guid: str) -> Any:
177         if guid in self._objects:
178             return self._objects[guid]
179         callback = self._loop.create_future()
180 
181         def callback_wrapper(result: Any) -> None:
182             callback.set_result(result)
183 
184         self._waiting_for_object[guid] = callback_wrapper
185         return await callback
186 
187     def call_on_object_with_known_name(
188         self, guid: str, callback: Callable[[Any], None]
189     ) -> None:
190         self._waiting_for_object[guid] = callback
191 
192     def _send_message_to_server(
193         self, guid: str, method: str, params: Dict
194     ) -> ProtocolCallback:
195         self._last_id += 1
196         id = self._last_id
197         callback = ProtocolCallback(self._loop)
198         task = asyncio.current_task(self._loop)
199         callback.stack_trace = getattr(task, "__pw_stack_trace__", None)
200         if not callback.stack_trace:
201             callback.stack_trace = traceback.extract_stack()
202 
203         metadata = {"stack": serialize_call_stack(callback.stack_trace)}
204         api_name = getattr(task, "__pw_api_name__", None)
205         if api_name:
206             metadata["apiName"] = api_name
207 
208         message = dict(
209             id=id,
210             guid=guid,
211             method=method,
212             params=self._replace_channels_with_guids(params, "params"),
213             metadata=metadata,
214         )
215         self._transport.send(message)
216         self._callbacks[id] = callback
217         return callback
218 
219     def _dispatch(self, msg: ParsedMessagePayload) -> None:
220         id = msg.get("id")
221         if id:
222             callback = self._callbacks.pop(id)
223             error = msg.get("error")
224             if error:
225                 parsed_error = parse_error(error["error"])  # type: ignore
226                 parsed_error.stack = "".join(
227                     traceback.format_list(callback.stack_trace)[-10:]
228                 )
229                 callback.future.set_exception(parsed_error)
230             else:
231                 result = self._replace_guids_with_channels(msg.get("result"))
232                 callback.future.set_result(result)
233             return
234 
235         guid = msg["guid"]
236         method = msg.get("method")
237         params = msg["params"]
238         if method == "__create__":
239             parent = self._objects[guid]
240             self._create_remote_object(
241                 parent, params["type"], params["guid"], params["initializer"]
242             )
243             return
244         if method == "__dispose__":
245             self._objects[guid]._dispose()
246             return
247 
248         object = self._objects[guid]
249         try:
250             if self._is_sync:
251                 for listener in object._channel.listeners(method):
252                     g = greenlet(listener)
253                     g.switch(self._replace_guids_with_channels(params))
254             else:
255                 object._channel.emit(method, self._replace_guids_with_channels(params))
256         except Exception:
257             print(
258                 "Error dispatching the event",
259                 "".join(traceback.format_exception(*sys.exc_info())),
260             )
261 
262     def _create_remote_object(
263         self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
264     ) -> Any:
265         result: ChannelOwner
266         initializer = self._replace_guids_with_channels(initializer)
267         result = self._object_factory(parent, type, guid, initializer)
268         if guid in self._waiting_for_object:
269             self._waiting_for_object.pop(guid)(result)
270         return result
271 
272     def _replace_channels_with_guids(self, payload: Any, param_name: str) -> Any:
273         if payload is None:
274             return payload
275         if isinstance(payload, Path):
276             return str(payload)
277         if isinstance(payload, list):
278             return list(
279                 map(lambda p: self._replace_channels_with_guids(p, "index"), payload)
280             )
281         if isinstance(payload, Channel):
282             return dict(guid=payload._guid)
283         if isinstance(payload, dict):
284             result = {}
285             for key in payload:
286                 result[key] = self._replace_channels_with_guids(payload[key], key)
287             return result
288         return payload
289 
290     def _replace_guids_with_channels(self, payload: Any) -> Any:
291         if payload is None:
292             return payload
293         if isinstance(payload, list):
294             return list(map(lambda p: self._replace_guids_with_channels(p), payload))
295         if isinstance(payload, dict):
296             if payload.get("guid") in self._objects:
297                 return self._objects[payload["guid"]]._channel
298             result = {}
299             for key in payload:
300                 result[key] = self._replace_guids_with_channels(payload[key])
301             return result
302         return payload
303 
304 
305 def from_channel(channel: Channel) -> Any:
306     return channel._object
307 
308 
309 def from_nullable_channel(channel: Optional[Channel]) -> Optional[Any]:
310     return channel._object if channel else None
311 
312 
313 def serialize_call_stack(stack_trace: traceback.StackSummary) -> List[Dict]:
314     stack: List[Dict] = []
315     for frame in stack_trace:
316         if "_generated.py" in frame.filename:
317             break
318         stack.append(
319             {"file": frame.filename, "line": frame.lineno, "function": frame.name}
320         )
321     stack.reverse()
322     return stack
323 
324 
325 def capture_call_stack() -> List[Dict]:
326     return serialize_call_stack(traceback.extract_stack())
327
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/playwright/_impl/_connection.py b/playwright/_impl/_connection.py
--- a/playwright/_impl/_connection.py
+++ b/playwright/_impl/_connection.py
@@ -220,6 +220,8 @@
         id = msg.get("id")
         if id:
             callback = self._callbacks.pop(id)
+            if callback.future.cancelled():
+                return
             error = msg.get("error")
             if error:
                 parsed_error = parse_error(error["error"])  # type: ignore

{"golden_diff": "diff --git a/playwright/_impl/_connection.py b/playwright/_impl/_connection.py\n--- a/playwright/_impl/_connection.py\n+++ b/playwright/_impl/_connection.py\n@@ -220,6 +220,8 @@\n         id = msg.get(\"id\")\n         if id:\n             callback = self._callbacks.pop(id)\n+            if callback.future.cancelled():\n+                return\n             error = msg.get(\"error\")\n             if error:\n                 parsed_error = parse_error(error[\"error\"])  # type: ignore\n", "issue": "fix: handle cancelled tasks\nprevents `InvalidStateError` when the task has been cancelled, here:\r\n\r\n```python\r\ncallback.future.set_exception(parsed_error)\r\n```\n", "before_files": [{"content": "# Copyright (c) Microsoft Corporation.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport asyncio\nimport sys\nimport traceback\nfrom pathlib import Path\nfrom typing import Any, Callable, Dict, List, Optional, Union\n\nfrom greenlet import greenlet\nfrom pyee import AsyncIOEventEmitter\n\nfrom playwright._impl._helper import ParsedMessagePayload, parse_error\nfrom playwright._impl._transport import Transport\n\n\nclass Channel(AsyncIOEventEmitter):\n    def __init__(self, connection: \"Connection\", guid: str) -> None:\n        super().__init__()\n        self._connection: Connection = connection\n        self._guid = guid\n        self._object: Optional[ChannelOwner] = None\n\n    async def send(self, method: str, params: Dict = None) -> Any:\n        return await self.inner_send(method, params, False)\n\n    async def send_return_as_dict(self, method: str, params: Dict = None) -> Any:\n        return await self.inner_send(method, params, True)\n\n    async def inner_send(\n        self, method: str, params: Optional[Dict], return_as_dict: bool\n    ) -> Any:\n        if params is None:\n            params = {}\n        callback = self._connection._send_message_to_server(self._guid, method, params)\n        result = await callback.future\n        # Protocol now has named return values, assume result is one level deeper unless\n        # there is explicit ambiguity.\n        if not result:\n            return None\n        assert isinstance(result, dict)\n        if return_as_dict:\n            return result\n        if len(result) == 0:\n            return None\n        assert len(result) == 1\n        key = next(iter(result))\n        return result[key]\n\n    def send_no_reply(self, method: str, params: Dict = None) -> None:\n        if params is None:\n            params = {}\n        self._connection._send_message_to_server(self._guid, method, params)\n\n\nclass ChannelOwner(AsyncIOEventEmitter):\n    def __init__(\n        self,\n        parent: Union[\"ChannelOwner\", \"Connection\"],\n        type: str,\n        guid: str,\n        initializer: Dict,\n    ) -> None:\n        super().__init__(loop=parent._loop)\n        self._loop: asyncio.AbstractEventLoop = parent._loop\n        self._dispatcher_fiber: Any = parent._dispatcher_fiber\n        self._type = type\n        self._guid = guid\n        self._connection: Connection = (\n            parent._connection if isinstance(parent, ChannelOwner) else parent\n        )\n        self._parent: Optional[ChannelOwner] = (\n            parent if isinstance(parent, ChannelOwner) else None\n        )\n        self._objects: Dict[str, \"ChannelOwner\"] = {}\n        self._channel = Channel(self._connection, guid)\n        self._channel._object = self\n        self._initializer = initializer\n\n        self._connection._objects[guid] = self\n        if self._parent:\n            self._parent._objects[guid] = self\n\n    def _wait_for_event_info_before(self, wait_id: str, name: str) -> None:\n        self._connection._send_message_to_server(\n            self._guid,\n            \"waitForEventInfo\",\n            {\n                \"info\": {\n                    \"name\": name,\n                    \"waitId\": wait_id,\n                    \"phase\": \"before\",\n                    \"stack\": capture_call_stack(),\n                }\n            },\n        )\n\n    def _wait_for_event_info_after(\n        self, wait_id: str, exception: Exception = None\n    ) -> None:\n        info = {\"waitId\": wait_id, \"phase\": \"after\"}\n        if exception:\n            info[\"error\"] = str(exception)\n        self._connection._send_message_to_server(\n            self._guid,\n            \"waitForEventInfo\",\n            {\"info\": info},\n        )\n\n    def _dispose(self) -> None:\n        # Clean up from parent and connection.\n        if self._parent:\n            del self._parent._objects[self._guid]\n        del self._connection._objects[self._guid]\n\n        # Dispose all children.\n        for object in list(self._objects.values()):\n            object._dispose()\n        self._objects.clear()\n\n\nclass ProtocolCallback:\n    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:\n        self.stack_trace: traceback.StackSummary = traceback.StackSummary()\n        self.future = loop.create_future()\n\n\nclass RootChannelOwner(ChannelOwner):\n    def __init__(self, connection: \"Connection\") -> None:\n        super().__init__(connection, \"\", \"\", {})\n\n\nclass Connection:\n    def __init__(\n        self, dispatcher_fiber: Any, object_factory: Any, driver_executable: Path\n    ) -> None:\n        self._dispatcher_fiber: Any = dispatcher_fiber\n        self._transport = Transport(driver_executable)\n        self._transport.on_message = lambda msg: self._dispatch(msg)\n        self._waiting_for_object: Dict[str, Any] = {}\n        self._last_id = 0\n        self._objects: Dict[str, ChannelOwner] = {}\n        self._callbacks: Dict[int, ProtocolCallback] = {}\n        self._object_factory = object_factory\n        self._is_sync = False\n        self._api_name = \"\"\n\n    async def run_as_sync(self) -> None:\n        self._is_sync = True\n        await self.run()\n\n    async def run(self) -> None:\n        self._loop = asyncio.get_running_loop()\n        self._root_object = RootChannelOwner(self)\n        await self._transport.run()\n\n    def stop_sync(self) -> None:\n        self._transport.stop()\n        self._dispatcher_fiber.switch()\n\n    async def stop_async(self) -> None:\n        self._transport.stop()\n        await self._transport.wait_until_stopped()\n\n    async def wait_for_object_with_known_name(self, guid: str) -> Any:\n        if guid in self._objects:\n            return self._objects[guid]\n        callback = self._loop.create_future()\n\n        def callback_wrapper(result: Any) -> None:\n            callback.set_result(result)\n\n        self._waiting_for_object[guid] = callback_wrapper\n        return await callback\n\n    def call_on_object_with_known_name(\n        self, guid: str, callback: Callable[[Any], None]\n    ) -> None:\n        self._waiting_for_object[guid] = callback\n\n    def _send_message_to_server(\n        self, guid: str, method: str, params: Dict\n    ) -> ProtocolCallback:\n        self._last_id += 1\n        id = self._last_id\n        callback = ProtocolCallback(self._loop)\n        task = asyncio.current_task(self._loop)\n        callback.stack_trace = getattr(task, \"__pw_stack_trace__\", None)\n        if not callback.stack_trace:\n            callback.stack_trace = traceback.extract_stack()\n\n        metadata = {\"stack\": serialize_call_stack(callback.stack_trace)}\n        api_name = getattr(task, \"__pw_api_name__\", None)\n        if api_name:\n            metadata[\"apiName\"] = api_name\n\n        message = dict(\n            id=id,\n            guid=guid,\n            method=method,\n            params=self._replace_channels_with_guids(params, \"params\"),\n            metadata=metadata,\n        )\n        self._transport.send(message)\n        self._callbacks[id] = callback\n        return callback\n\n    def _dispatch(self, msg: ParsedMessagePayload) -> None:\n        id = msg.get(\"id\")\n        if id:\n            callback = self._callbacks.pop(id)\n            error = msg.get(\"error\")\n            if error:\n                parsed_error = parse_error(error[\"error\"])  # type: ignore\n                parsed_error.stack = \"\".join(\n                    traceback.format_list(callback.stack_trace)[-10:]\n                )\n                callback.future.set_exception(parsed_error)\n            else:\n                result = self._replace_guids_with_channels(msg.get(\"result\"))\n                callback.future.set_result(result)\n            return\n\n        guid = msg[\"guid\"]\n        method = msg.get(\"method\")\n        params = msg[\"params\"]\n        if method == \"__create__\":\n            parent = self._objects[guid]\n            self._create_remote_object(\n                parent, params[\"type\"], params[\"guid\"], params[\"initializer\"]\n            )\n            return\n        if method == \"__dispose__\":\n            self._objects[guid]._dispose()\n            return\n\n        object = self._objects[guid]\n        try:\n            if self._is_sync:\n                for listener in object._channel.listeners(method):\n                    g = greenlet(listener)\n                    g.switch(self._replace_guids_with_channels(params))\n            else:\n                object._channel.emit(method, self._replace_guids_with_channels(params))\n        except Exception:\n            print(\n                \"Error dispatching the event\",\n                \"\".join(traceback.format_exception(*sys.exc_info())),\n            )\n\n    def _create_remote_object(\n        self, parent: ChannelOwner, type: str, guid: str, initializer: Dict\n    ) -> Any:\n        result: ChannelOwner\n        initializer = self._replace_guids_with_channels(initializer)\n        result = self._object_factory(parent, type, guid, initializer)\n        if guid in self._waiting_for_object:\n            self._waiting_for_object.pop(guid)(result)\n        return result\n\n    def _replace_channels_with_guids(self, payload: Any, param_name: str) -> Any:\n        if payload is None:\n            return payload\n        if isinstance(payload, Path):\n            return str(payload)\n        if isinstance(payload, list):\n            return list(\n                map(lambda p: self._replace_channels_with_guids(p, \"index\"), payload)\n            )\n        if isinstance(payload, Channel):\n            return dict(guid=payload._guid)\n        if isinstance(payload, dict):\n            result = {}\n            for key in payload:\n                result[key] = self._replace_channels_with_guids(payload[key], key)\n            return result\n        return payload\n\n    def _replace_guids_with_channels(self, payload: Any) -> Any:\n        if payload is None:\n            return payload\n        if isinstance(payload, list):\n            return list(map(lambda p: self._replace_guids_with_channels(p), payload))\n        if isinstance(payload, dict):\n            if payload.get(\"guid\") in self._objects:\n                return self._objects[payload[\"guid\"]]._channel\n            result = {}\n            for key in payload:\n                result[key] = self._replace_guids_with_channels(payload[key])\n            return result\n        return payload\n\n\ndef from_channel(channel: Channel) -> Any:\n    return channel._object\n\n\ndef from_nullable_channel(channel: Optional[Channel]) -> Optional[Any]:\n    return channel._object if channel else None\n\n\ndef serialize_call_stack(stack_trace: traceback.StackSummary) -> List[Dict]:\n    stack: List[Dict] = []\n    for frame in stack_trace:\n        if \"_generated.py\" in frame.filename:\n            break\n        stack.append(\n            {\"file\": frame.filename, \"line\": frame.lineno, \"function\": frame.name}\n        )\n    stack.reverse()\n    return stack\n\n\ndef capture_call_stack() -> List[Dict]:\n    return serialize_call_stack(traceback.extract_stack())\n", "path": "playwright/_impl/_connection.py"}], "after_files": [{"content": "# Copyright (c) Microsoft Corporation.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport asyncio\nimport sys\nimport traceback\nfrom pathlib import Path\nfrom typing import Any, Callable, Dict, List, Optional, Union\n\nfrom greenlet import greenlet\nfrom pyee import AsyncIOEventEmitter\n\nfrom playwright._impl._helper import ParsedMessagePayload, parse_error\nfrom playwright._impl._transport import Transport\n\n\nclass Channel(AsyncIOEventEmitter):\n    def __init__(self, connection: \"Connection\", guid: str) -> None:\n        super().__init__()\n        self._connection: Connection = connection\n        self._guid = guid\n        self._object: Optional[ChannelOwner] = None\n\n    async def send(self, method: str, params: Dict = None) -> Any:\n        return await self.inner_send(method, params, False)\n\n    async def send_return_as_dict(self, method: str, params: Dict = None) -> Any:\n        return await self.inner_send(method, params, True)\n\n    async def inner_send(\n        self, method: str, params: Optional[Dict], return_as_dict: bool\n    ) -> Any:\n        if params is None:\n            params = {}\n        callback = self._connection._send_message_to_server(self._guid, method, params)\n        result = await callback.future\n        # Protocol now has named return values, assume result is one level deeper unless\n        # there is explicit ambiguity.\n        if not result:\n            return None\n        assert isinstance(result, dict)\n        if return_as_dict:\n            return result\n        if len(result) == 0:\n            return None\n        assert len(result) == 1\n        key = next(iter(result))\n        return result[key]\n\n    def send_no_reply(self, method: str, params: Dict = None) -> None:\n        if params is None:\n            params = {}\n        self._connection._send_message_to_server(self._guid, method, params)\n\n\nclass ChannelOwner(AsyncIOEventEmitter):\n    def __init__(\n        self,\n        parent: Union[\"ChannelOwner\", \"Connection\"],\n        type: str,\n        guid: str,\n        initializer: Dict,\n    ) -> None:\n        super().__init__(loop=parent._loop)\n        self._loop: asyncio.AbstractEventLoop = parent._loop\n        self._dispatcher_fiber: Any = parent._dispatcher_fiber\n        self._type = type\n        self._guid = guid\n        self._connection: Connection = (\n            parent._connection if isinstance(parent, ChannelOwner) else parent\n        )\n        self._parent: Optional[ChannelOwner] = (\n            parent if isinstance(parent, ChannelOwner) else None\n        )\n        self._objects: Dict[str, \"ChannelOwner\"] = {}\n        self._channel = Channel(self._connection, guid)\n        self._channel._object = self\n        self._initializer = initializer\n\n        self._connection._objects[guid] = self\n        if self._parent:\n            self._parent._objects[guid] = self\n\n    def _wait_for_event_info_before(self, wait_id: str, name: str) -> None:\n        self._connection._send_message_to_server(\n            self._guid,\n            \"waitForEventInfo\",\n            {\n                \"info\": {\n                    \"name\": name,\n                    \"waitId\": wait_id,\n                    \"phase\": \"before\",\n                    \"stack\": capture_call_stack(),\n                }\n            },\n        )\n\n    def _wait_for_event_info_after(\n        self, wait_id: str, exception: Exception = None\n    ) -> None:\n        info = {\"waitId\": wait_id, \"phase\": \"after\"}\n        if exception:\n            info[\"error\"] = str(exception)\n        self._connection._send_message_to_server(\n            self._guid,\n            \"waitForEventInfo\",\n            {\"info\": info},\n        )\n\n    def _dispose(self) -> None:\n        # Clean up from parent and connection.\n        if self._parent:\n            del self._parent._objects[self._guid]\n        del self._connection._objects[self._guid]\n\n        # Dispose all children.\n        for object in list(self._objects.values()):\n            object._dispose()\n        self._objects.clear()\n\n\nclass ProtocolCallback:\n    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:\n        self.stack_trace: traceback.StackSummary = traceback.StackSummary()\n        self.future = loop.create_future()\n\n\nclass RootChannelOwner(ChannelOwner):\n    def __init__(self, connection: \"Connection\") -> None:\n        super().__init__(connection, \"\", \"\", {})\n\n\nclass Connection:\n    def __init__(\n        self, dispatcher_fiber: Any, object_factory: Any, driver_executable: Path\n    ) -> None:\n        self._dispatcher_fiber: Any = dispatcher_fiber\n        self._transport = Transport(driver_executable)\n        self._transport.on_message = lambda msg: self._dispatch(msg)\n        self._waiting_for_object: Dict[str, Any] = {}\n        self._last_id = 0\n        self._objects: Dict[str, ChannelOwner] = {}\n        self._callbacks: Dict[int, ProtocolCallback] = {}\n        self._object_factory = object_factory\n        self._is_sync = False\n        self._api_name = \"\"\n\n    async def run_as_sync(self) -> None:\n        self._is_sync = True\n        await self.run()\n\n    async def run(self) -> None:\n        self._loop = asyncio.get_running_loop()\n        self._root_object = RootChannelOwner(self)\n        await self._transport.run()\n\n    def stop_sync(self) -> None:\n        self._transport.stop()\n        self._dispatcher_fiber.switch()\n\n    async def stop_async(self) -> None:\n        self._transport.stop()\n        await self._transport.wait_until_stopped()\n\n    async def wait_for_object_with_known_name(self, guid: str) -> Any:\n        if guid in self._objects:\n            return self._objects[guid]\n        callback = self._loop.create_future()\n\n        def callback_wrapper(result: Any) -> None:\n            callback.set_result(result)\n\n        self._waiting_for_object[guid] = callback_wrapper\n        return await callback\n\n    def call_on_object_with_known_name(\n        self, guid: str, callback: Callable[[Any], None]\n    ) -> None:\n        self._waiting_for_object[guid] = callback\n\n    def _send_message_to_server(\n        self, guid: str, method: str, params: Dict\n    ) -> ProtocolCallback:\n        self._last_id += 1\n        id = self._last_id\n        callback = ProtocolCallback(self._loop)\n        task = asyncio.current_task(self._loop)\n        callback.stack_trace = getattr(task, \"__pw_stack_trace__\", None)\n        if not callback.stack_trace:\n            callback.stack_trace = traceback.extract_stack()\n\n        metadata = {\"stack\": serialize_call_stack(callback.stack_trace)}\n        api_name = getattr(task, \"__pw_api_name__\", None)\n        if api_name:\n            metadata[\"apiName\"] = api_name\n\n        message = dict(\n            id=id,\n            guid=guid,\n            method=method,\n            params=self._replace_channels_with_guids(params, \"params\"),\n            metadata=metadata,\n        )\n        self._transport.send(message)\n        self._callbacks[id] = callback\n        return callback\n\n    def _dispatch(self, msg: ParsedMessagePayload) -> None:\n        id = msg.get(\"id\")\n        if id:\n            callback = self._callbacks.pop(id)\n            if callback.future.cancelled():\n                return\n            error = msg.get(\"error\")\n            if error:\n                parsed_error = parse_error(error[\"error\"])  # type: ignore\n                parsed_error.stack = \"\".join(\n                    traceback.format_list(callback.stack_trace)[-10:]\n                )\n                callback.future.set_exception(parsed_error)\n            else:\n                result = self._replace_guids_with_channels(msg.get(\"result\"))\n                callback.future.set_result(result)\n            return\n\n        guid = msg[\"guid\"]\n        method = msg.get(\"method\")\n        params = msg[\"params\"]\n        if method == \"__create__\":\n            parent = self._objects[guid]\n            self._create_remote_object(\n                parent, params[\"type\"], params[\"guid\"], params[\"initializer\"]\n            )\n            return\n        if method == \"__dispose__\":\n            self._objects[guid]._dispose()\n            return\n\n        object = self._objects[guid]\n        try:\n            if self._is_sync:\n                for listener in object._channel.listeners(method):\n                    g = greenlet(listener)\n                    g.switch(self._replace_guids_with_channels(params))\n            else:\n                object._channel.emit(method, self._replace_guids_with_channels(params))\n        except Exception:\n            print(\n                \"Error dispatching the event\",\n                \"\".join(traceback.format_exception(*sys.exc_info())),\n            )\n\n    def _create_remote_object(\n        self, parent: ChannelOwner, type: str, guid: str, initializer: Dict\n    ) -> Any:\n        result: ChannelOwner\n        initializer = self._replace_guids_with_channels(initializer)\n        result = self._object_factory(parent, type, guid, initializer)\n        if guid in self._waiting_for_object:\n            self._waiting_for_object.pop(guid)(result)\n        return result\n\n    def _replace_channels_with_guids(self, payload: Any, param_name: str) -> Any:\n        if payload is None:\n            return payload\n        if isinstance(payload, Path):\n            return str(payload)\n        if isinstance(payload, list):\n            return list(\n                map(lambda p: self._replace_channels_with_guids(p, \"index\"), payload)\n            )\n        if isinstance(payload, Channel):\n            return dict(guid=payload._guid)\n        if isinstance(payload, dict):\n            result = {}\n            for key in payload:\n                result[key] = self._replace_channels_with_guids(payload[key], key)\n            return result\n        return payload\n\n    def _replace_guids_with_channels(self, payload: Any) -> Any:\n        if payload is None:\n            return payload\n        if isinstance(payload, list):\n            return list(map(lambda p: self._replace_guids_with_channels(p), payload))\n        if isinstance(payload, dict):\n            if payload.get(\"guid\") in self._objects:\n                return self._objects[payload[\"guid\"]]._channel\n            result = {}\n            for key in payload:\n                result[key] = self._replace_guids_with_channels(payload[key])\n            return result\n        return payload\n\n\ndef from_channel(channel: Channel) -> Any:\n    return channel._object\n\n\ndef from_nullable_channel(channel: Optional[Channel]) -> Optional[Any]:\n    return channel._object if channel else None\n\n\ndef serialize_call_stack(stack_trace: traceback.StackSummary) -> List[Dict]:\n    stack: List[Dict] = []\n    for frame in stack_trace:\n        if \"_generated.py\" in frame.filename:\n            break\n        stack.append(\n            {\"file\": frame.filename, \"line\": frame.lineno, \"function\": frame.name}\n        )\n    stack.reverse()\n    return stack\n\n\ndef capture_call_stack() -> List[Dict]:\n    return serialize_call_stack(traceback.extract_stack())\n", "path": "playwright/_impl/_connection.py"}]}