We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Deleting generator in subscription clean up will not call a 'finally' statement
<!-- Provide a general summary of the bug in the title above. -->

<!--- This template is entirely optional and can be removed, but is here to help both you and us. -->
<!--- Anything on lines wrapped in comments like these will not show up in the final text. -->

## Describe the Bug
I have found an issue with the way subscriptions are cleaned up when using the graphql-transport-ws protocol. 

My subscription returns a AsyncIterator which look something like this:
```
async def get_value() -> AsyncIterator[int]:
    try:
        while True:
            # Call some other code to 
            # get the value from another object
    finally:
        # Do some important cleanup here.
```
I have multiple subscriptions, which are being updated at 10Hz. When I cancel the subscriptions (for example, close the webpage calling subscribe), I get a `ConnectionResetError: "Cannot write to closing transport"` on the server side, presumably as the task has not yet been cancelled and is trying to send an updated value while the websocket is closing. Looking at the Strawberry code that handles this (subscriptions/protocols/graphql_transport_ws/handlers.py), I can see that this error is caught as a `BaseException` ([here](https://github.com/strawberry-graphql/strawberry/blob/fe9737b9d218f437500db215525a147de2ea1d64/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py#L218)) and so this task and generator are deleted from the internal dictionaries. 

What I have found however, is that when the generator is simply deleted in this clean up, the 'finally' statement in my try-finally implementation does not get executed and so I cannot perform any clean up. I have seen this lead to a memory leak, which is only stopped when the garbage collector comes round and runs the 'finally' statement. 

To ensure the generator runs to completion and executes the 'finally' statement, I wonder whether the generator should be explicitly closed first, before deleting it from the internal subscriptions list, much like is done in the cleanup_operation() method ([here](https://github.com/strawberry-graphql/strawberry/blob/fe9737b9d218f437500db215525a147de2ea1d64/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py#L282?)).

## System Information

 - Operating system: Centos 7
 - Strawberry version (if applicable): 0.121.1
 - Python version: 3.8.12

## Additional Context

<!-- Add any other relevant information about the problem here. -->


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py`
Content:
```
1 import asyncio
2 from abc import ABC, abstractmethod
3 from contextlib import suppress
4 from datetime import timedelta
5 from typing import Any, AsyncGenerator, Callable, Dict, List, Optional
6 
7 from graphql import (
8     ExecutionResult as GraphQLExecutionResult,
9     GraphQLError,
10     GraphQLSyntaxError,
11     parse,
12 )
13 from graphql.error.graphql_error import format_error as format_graphql_error
14 
15 from strawberry.schema import BaseSchema
16 from strawberry.subscriptions.protocols.graphql_transport_ws.types import (
17     CompleteMessage,
18     ConnectionAckMessage,
19     ConnectionInitMessage,
20     ErrorMessage,
21     GraphQLTransportMessage,
22     NextMessage,
23     PingMessage,
24     PongMessage,
25     SubscribeMessage,
26     SubscribeMessagePayload,
27 )
28 from strawberry.types.graphql import OperationType
29 from strawberry.utils.debug import pretty_print_graphql_operation
30 from strawberry.utils.operation import get_operation_type
31 
32 
33 class BaseGraphQLTransportWSHandler(ABC):
34     def __init__(
35         self,
36         schema: BaseSchema,
37         debug: bool,
38         connection_init_wait_timeout: timedelta,
39     ):
40         self.schema = schema
41         self.debug = debug
42         self.connection_init_wait_timeout = connection_init_wait_timeout
43         self.connection_init_timeout_task: Optional[asyncio.Task] = None
44         self.connection_init_received = False
45         self.connection_acknowledged = False
46         self.subscriptions: Dict[str, AsyncGenerator] = {}
47         self.tasks: Dict[str, asyncio.Task] = {}
48         self.completed_tasks: List[asyncio.Task] = []
49 
50     @abstractmethod
51     async def get_context(self) -> Any:
52         """Return the operations context"""
53 
54     @abstractmethod
55     async def get_root_value(self) -> Any:
56         """Return the schemas root value"""
57 
58     @abstractmethod
59     async def send_json(self, data: dict) -> None:
60         """Send the data JSON encoded to the WebSocket client"""
61 
62     @abstractmethod
63     async def close(self, code: int, reason: str) -> None:
64         """Close the WebSocket with the passed code and reason"""
65 
66     @abstractmethod
67     async def handle_request(self) -> Any:
68         """Handle the request this instance was created for"""
69 
70     async def handle(self) -> Any:
71         timeout_handler = self.handle_connection_init_timeout()
72         self.connection_init_timeout_task = asyncio.create_task(timeout_handler)
73         return await self.handle_request()
74 
75     async def handle_connection_init_timeout(self):
76         delay = self.connection_init_wait_timeout.total_seconds()
77         await asyncio.sleep(delay=delay)
78 
79         if self.connection_init_received:
80             return
81 
82         reason = "Connection initialisation timeout"
83         await self.close(code=4408, reason=reason)
84 
85     async def handle_message(self, message: dict):
86         handler: Callable
87         handler_arg: Any
88         try:
89             message_type = message.pop("type")
90 
91             if message_type == ConnectionInitMessage.type:
92                 handler = self.handle_connection_init
93                 handler_arg = ConnectionInitMessage(**message)
94 
95             elif message_type == PingMessage.type:
96                 handler = self.handle_ping
97                 handler_arg = PingMessage(**message)
98 
99             elif message_type == PongMessage.type:
100                 handler = self.handle_pong
101                 handler_arg = PongMessage(**message)
102 
103             elif message_type == SubscribeMessage.type:
104                 handler = self.handle_subscribe
105                 payload = SubscribeMessagePayload(**message.pop("payload"))
106                 handler_arg = SubscribeMessage(payload=payload, **message)
107 
108             elif message_type == CompleteMessage.type:
109                 handler = self.handle_complete
110                 handler_arg = CompleteMessage(**message)
111 
112             else:
113                 handler = self.handle_invalid_message
114                 handler_arg = f"Unknown message type: {message_type}"
115 
116         except (KeyError, TypeError):
117             handler = self.handle_invalid_message
118             handler_arg = "Failed to parse message"
119 
120         await handler(handler_arg)
121         await self.reap_completed_tasks()
122 
123     async def handle_connection_init(self, message: ConnectionInitMessage) -> None:
124         if self.connection_init_received:
125             reason = "Too many initialisation requests"
126             await self.close(code=4429, reason=reason)
127             return
128 
129         self.connection_init_received = True
130         await self.send_message(ConnectionAckMessage())
131         self.connection_acknowledged = True
132 
133     async def handle_ping(self, message: PingMessage) -> None:
134         await self.send_message(PongMessage())
135 
136     async def handle_pong(self, message: PongMessage) -> None:
137         pass
138 
139     async def handle_subscribe(self, message: SubscribeMessage) -> None:
140         if not self.connection_acknowledged:
141             await self.close(code=4401, reason="Unauthorized")
142             return
143 
144         try:
145             graphql_document = parse(message.payload.query)
146         except GraphQLSyntaxError as exc:
147             await self.close(code=4400, reason=exc.message)
148             return
149 
150         try:
151             operation_type = get_operation_type(
152                 graphql_document, message.payload.operationName
153             )
154         except RuntimeError:
155             await self.close(code=4400, reason="Can't get GraphQL operation type")
156             return
157 
158         if message.id in self.subscriptions:
159             reason = f"Subscriber for {message.id} already exists"
160             await self.close(code=4409, reason=reason)
161             return
162 
163         if self.debug:  # pragma: no cover
164             pretty_print_graphql_operation(
165                 message.payload.operationName,
166                 message.payload.query,
167                 message.payload.variables,
168             )
169 
170         context = await self.get_context()
171         root_value = await self.get_root_value()
172 
173         # Get an AsyncGenerator yielding the results
174         if operation_type == OperationType.SUBSCRIPTION:
175             result_source = await self.schema.subscribe(
176                 query=message.payload.query,
177                 variable_values=message.payload.variables,
178                 operation_name=message.payload.operationName,
179                 context_value=context,
180                 root_value=root_value,
181             )
182         else:
183             # create AsyncGenerator returning a single result
184             async def get_result_source():
185                 yield await self.schema.execute(
186                     query=message.payload.query,
187                     variable_values=message.payload.variables,
188                     context_value=context,
189                     root_value=root_value,
190                     operation_name=message.payload.operationName,
191                 )
192 
193             result_source = get_result_source()
194 
195         # Handle initial validation errors
196         if isinstance(result_source, GraphQLExecutionResult):
197             assert result_source.errors
198             payload = [format_graphql_error(result_source.errors[0])]
199             await self.send_message(ErrorMessage(id=message.id, payload=payload))
200             self.schema.process_errors(result_source.errors)
201             return
202 
203         # Create task to handle this subscription, reserve the operation ID
204         self.subscriptions[message.id] = result_source
205         self.tasks[message.id] = asyncio.create_task(
206             self.operation_task(result_source, message.id)
207         )
208 
209     async def operation_task(
210         self, result_source: AsyncGenerator, operation_id: str
211     ) -> None:
212         """
213         Operation task top level method.  Cleans up and de-registers the operation
214         once it is done.
215         """
216         try:
217             await self.handle_async_results(result_source, operation_id)
218         except BaseException:  # pragma: no cover
219             # cleanup in case of something really unexpected
220             del self.subscriptions[operation_id]
221             del self.tasks[operation_id]
222             raise
223         else:
224             # de-register the operation _before_ sending the `Complete` message
225             # to make the `operation_id` immediately available for re-use
226             del self.subscriptions[operation_id]
227             del self.tasks[operation_id]
228             await self.send_message(CompleteMessage(id=operation_id))
229         finally:
230             # add this task to a list to be reaped later
231             task = asyncio.current_task()
232             assert task is not None
233             self.completed_tasks.append(task)
234 
235     async def handle_async_results(
236         self,
237         result_source: AsyncGenerator,
238         operation_id: str,
239     ) -> None:
240         try:
241             async for result in result_source:
242                 if result.errors:
243                     error_payload = [format_graphql_error(err) for err in result.errors]
244                     error_message = ErrorMessage(id=operation_id, payload=error_payload)
245                     await self.send_message(error_message)
246                     self.schema.process_errors(result.errors)
247                     return
248                 else:
249                     next_payload = {"data": result.data}
250                     next_message = NextMessage(id=operation_id, payload=next_payload)
251                     await self.send_message(next_message)
252         except asyncio.CancelledError:
253             # CancelledErrors are expected during task cleanup.
254             return
255         except Exception as error:
256             # GraphQLErrors are handled by graphql-core and included in the
257             # ExecutionResult
258             error = GraphQLError(str(error), original_error=error)
259             error_payload = [format_graphql_error(error)]
260             error_message = ErrorMessage(id=operation_id, payload=error_payload)
261             await self.send_message(error_message)
262             self.schema.process_errors([error])
263             return
264 
265     async def handle_complete(self, message: CompleteMessage) -> None:
266         await self.cleanup_operation(operation_id=message.id)
267 
268     async def handle_invalid_message(self, error_message: str) -> None:
269         await self.close(code=4400, reason=error_message)
270 
271     async def send_message(self, message: GraphQLTransportMessage) -> None:
272         data = message.as_dict()
273         await self.send_json(data)
274 
275     async def cleanup_operation(self, operation_id: str) -> None:
276         if operation_id not in self.subscriptions:
277             return
278         generator = self.subscriptions.pop(operation_id)
279         task = self.tasks.pop(operation_id)
280         # since python 3.8, generators cannot be reliably closed
281         with suppress(RuntimeError):
282             await generator.aclose()
283         task.cancel()
284         with suppress(BaseException):
285             await task
286 
287     async def reap_completed_tasks(self) -> None:
288         """
289         Await tasks that have completed
290         """
291         tasks, self.completed_tasks = self.completed_tasks, []
292         for task in tasks:
293             with suppress(BaseException):
294                 await task
295
```
--- 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/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py b/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py
--- a/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py
+++ b/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py
@@ -217,6 +217,11 @@
             await self.handle_async_results(result_source, operation_id)
         except BaseException:  # pragma: no cover
             # cleanup in case of something really unexpected
+            # wait for generator to be closed to ensure that any existing
+            # 'finally' statement is called
+            result_source = self.subscriptions[operation_id]
+            with suppress(RuntimeError):
+                await result_source.aclose()
             del self.subscriptions[operation_id]
             del self.tasks[operation_id]
             raise
@@ -275,14 +280,14 @@
     async def cleanup_operation(self, operation_id: str) -> None:
         if operation_id not in self.subscriptions:
             return
-        generator = self.subscriptions.pop(operation_id)
+        result_source = self.subscriptions.pop(operation_id)
         task = self.tasks.pop(operation_id)
-        # since python 3.8, generators cannot be reliably closed
-        with suppress(RuntimeError):
-            await generator.aclose()
         task.cancel()
         with suppress(BaseException):
             await task
+        # since python 3.8, generators cannot be reliably closed
+        with suppress(RuntimeError):
+            await result_source.aclose()
 
     async def reap_completed_tasks(self) -> None:
         """

{"golden_diff": "diff --git a/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py b/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py\n--- a/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py\n+++ b/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py\n@@ -217,6 +217,11 @@\n             await self.handle_async_results(result_source, operation_id)\n         except BaseException:  # pragma: no cover\n             # cleanup in case of something really unexpected\n+            # wait for generator to be closed to ensure that any existing\n+            # 'finally' statement is called\n+            result_source = self.subscriptions[operation_id]\n+            with suppress(RuntimeError):\n+                await result_source.aclose()\n             del self.subscriptions[operation_id]\n             del self.tasks[operation_id]\n             raise\n@@ -275,14 +280,14 @@\n     async def cleanup_operation(self, operation_id: str) -> None:\n         if operation_id not in self.subscriptions:\n             return\n-        generator = self.subscriptions.pop(operation_id)\n+        result_source = self.subscriptions.pop(operation_id)\n         task = self.tasks.pop(operation_id)\n-        # since python 3.8, generators cannot be reliably closed\n-        with suppress(RuntimeError):\n-            await generator.aclose()\n         task.cancel()\n         with suppress(BaseException):\n             await task\n+        # since python 3.8, generators cannot be reliably closed\n+        with suppress(RuntimeError):\n+            await result_source.aclose()\n \n     async def reap_completed_tasks(self) -> None:\n         \"\"\"\n", "issue": "Deleting generator in subscription clean up will not call a 'finally' statement\n<!-- Provide a general summary of the bug in the title above. -->\r\n\r\n<!--- This template is entirely optional and can be removed, but is here to help both you and us. -->\r\n<!--- Anything on lines wrapped in comments like these will not show up in the final text. -->\r\n\r\n## Describe the Bug\r\nI have found an issue with the way subscriptions are cleaned up when using the graphql-transport-ws protocol. \r\n\r\nMy subscription returns a AsyncIterator which look something like this:\r\n```\r\nasync def get_value() -> AsyncIterator[int]:\r\n    try:\r\n        while True:\r\n            # Call some other code to \r\n            # get the value from another object\r\n    finally:\r\n        # Do some important cleanup here.\r\n```\r\nI have multiple subscriptions, which are being updated at 10Hz. When I cancel the subscriptions (for example, close the webpage calling subscribe), I get a `ConnectionResetError: \"Cannot write to closing transport\"` on the server side, presumably as the task has not yet been cancelled and is trying to send an updated value while the websocket is closing. Looking at the Strawberry code that handles this (subscriptions/protocols/graphql_transport_ws/handlers.py), I can see that this error is caught as a `BaseException` ([here](https://github.com/strawberry-graphql/strawberry/blob/fe9737b9d218f437500db215525a147de2ea1d64/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py#L218)) and so this task and generator are deleted from the internal dictionaries. \r\n\r\nWhat I have found however, is that when the generator is simply deleted in this clean up, the 'finally' statement in my try-finally implementation does not get executed and so I cannot perform any clean up. I have seen this lead to a memory leak, which is only stopped when the garbage collector comes round and runs the 'finally' statement. \r\n\r\nTo ensure the generator runs to completion and executes the 'finally' statement, I wonder whether the generator should be explicitly closed first, before deleting it from the internal subscriptions list, much like is done in the cleanup_operation() method ([here](https://github.com/strawberry-graphql/strawberry/blob/fe9737b9d218f437500db215525a147de2ea1d64/strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py#L282?)).\r\n\r\n## System Information\r\n\r\n - Operating system: Centos 7\r\n - Strawberry version (if applicable): 0.121.1\r\n - Python version: 3.8.12\r\n\r\n## Additional Context\r\n\r\n<!-- Add any other relevant information about the problem here. -->\r\n\n", "before_files": [{"content": "import asyncio\nfrom abc import ABC, abstractmethod\nfrom contextlib import suppress\nfrom datetime import timedelta\nfrom typing import Any, AsyncGenerator, Callable, Dict, List, Optional\n\nfrom graphql import (\n    ExecutionResult as GraphQLExecutionResult,\n    GraphQLError,\n    GraphQLSyntaxError,\n    parse,\n)\nfrom graphql.error.graphql_error import format_error as format_graphql_error\n\nfrom strawberry.schema import BaseSchema\nfrom strawberry.subscriptions.protocols.graphql_transport_ws.types import (\n    CompleteMessage,\n    ConnectionAckMessage,\n    ConnectionInitMessage,\n    ErrorMessage,\n    GraphQLTransportMessage,\n    NextMessage,\n    PingMessage,\n    PongMessage,\n    SubscribeMessage,\n    SubscribeMessagePayload,\n)\nfrom strawberry.types.graphql import OperationType\nfrom strawberry.utils.debug import pretty_print_graphql_operation\nfrom strawberry.utils.operation import get_operation_type\n\n\nclass BaseGraphQLTransportWSHandler(ABC):\n    def __init__(\n        self,\n        schema: BaseSchema,\n        debug: bool,\n        connection_init_wait_timeout: timedelta,\n    ):\n        self.schema = schema\n        self.debug = debug\n        self.connection_init_wait_timeout = connection_init_wait_timeout\n        self.connection_init_timeout_task: Optional[asyncio.Task] = None\n        self.connection_init_received = False\n        self.connection_acknowledged = False\n        self.subscriptions: Dict[str, AsyncGenerator] = {}\n        self.tasks: Dict[str, asyncio.Task] = {}\n        self.completed_tasks: List[asyncio.Task] = []\n\n    @abstractmethod\n    async def get_context(self) -> Any:\n        \"\"\"Return the operations context\"\"\"\n\n    @abstractmethod\n    async def get_root_value(self) -> Any:\n        \"\"\"Return the schemas root value\"\"\"\n\n    @abstractmethod\n    async def send_json(self, data: dict) -> None:\n        \"\"\"Send the data JSON encoded to the WebSocket client\"\"\"\n\n    @abstractmethod\n    async def close(self, code: int, reason: str) -> None:\n        \"\"\"Close the WebSocket with the passed code and reason\"\"\"\n\n    @abstractmethod\n    async def handle_request(self) -> Any:\n        \"\"\"Handle the request this instance was created for\"\"\"\n\n    async def handle(self) -> Any:\n        timeout_handler = self.handle_connection_init_timeout()\n        self.connection_init_timeout_task = asyncio.create_task(timeout_handler)\n        return await self.handle_request()\n\n    async def handle_connection_init_timeout(self):\n        delay = self.connection_init_wait_timeout.total_seconds()\n        await asyncio.sleep(delay=delay)\n\n        if self.connection_init_received:\n            return\n\n        reason = \"Connection initialisation timeout\"\n        await self.close(code=4408, reason=reason)\n\n    async def handle_message(self, message: dict):\n        handler: Callable\n        handler_arg: Any\n        try:\n            message_type = message.pop(\"type\")\n\n            if message_type == ConnectionInitMessage.type:\n                handler = self.handle_connection_init\n                handler_arg = ConnectionInitMessage(**message)\n\n            elif message_type == PingMessage.type:\n                handler = self.handle_ping\n                handler_arg = PingMessage(**message)\n\n            elif message_type == PongMessage.type:\n                handler = self.handle_pong\n                handler_arg = PongMessage(**message)\n\n            elif message_type == SubscribeMessage.type:\n                handler = self.handle_subscribe\n                payload = SubscribeMessagePayload(**message.pop(\"payload\"))\n                handler_arg = SubscribeMessage(payload=payload, **message)\n\n            elif message_type == CompleteMessage.type:\n                handler = self.handle_complete\n                handler_arg = CompleteMessage(**message)\n\n            else:\n                handler = self.handle_invalid_message\n                handler_arg = f\"Unknown message type: {message_type}\"\n\n        except (KeyError, TypeError):\n            handler = self.handle_invalid_message\n            handler_arg = \"Failed to parse message\"\n\n        await handler(handler_arg)\n        await self.reap_completed_tasks()\n\n    async def handle_connection_init(self, message: ConnectionInitMessage) -> None:\n        if self.connection_init_received:\n            reason = \"Too many initialisation requests\"\n            await self.close(code=4429, reason=reason)\n            return\n\n        self.connection_init_received = True\n        await self.send_message(ConnectionAckMessage())\n        self.connection_acknowledged = True\n\n    async def handle_ping(self, message: PingMessage) -> None:\n        await self.send_message(PongMessage())\n\n    async def handle_pong(self, message: PongMessage) -> None:\n        pass\n\n    async def handle_subscribe(self, message: SubscribeMessage) -> None:\n        if not self.connection_acknowledged:\n            await self.close(code=4401, reason=\"Unauthorized\")\n            return\n\n        try:\n            graphql_document = parse(message.payload.query)\n        except GraphQLSyntaxError as exc:\n            await self.close(code=4400, reason=exc.message)\n            return\n\n        try:\n            operation_type = get_operation_type(\n                graphql_document, message.payload.operationName\n            )\n        except RuntimeError:\n            await self.close(code=4400, reason=\"Can't get GraphQL operation type\")\n            return\n\n        if message.id in self.subscriptions:\n            reason = f\"Subscriber for {message.id} already exists\"\n            await self.close(code=4409, reason=reason)\n            return\n\n        if self.debug:  # pragma: no cover\n            pretty_print_graphql_operation(\n                message.payload.operationName,\n                message.payload.query,\n                message.payload.variables,\n            )\n\n        context = await self.get_context()\n        root_value = await self.get_root_value()\n\n        # Get an AsyncGenerator yielding the results\n        if operation_type == OperationType.SUBSCRIPTION:\n            result_source = await self.schema.subscribe(\n                query=message.payload.query,\n                variable_values=message.payload.variables,\n                operation_name=message.payload.operationName,\n                context_value=context,\n                root_value=root_value,\n            )\n        else:\n            # create AsyncGenerator returning a single result\n            async def get_result_source():\n                yield await self.schema.execute(\n                    query=message.payload.query,\n                    variable_values=message.payload.variables,\n                    context_value=context,\n                    root_value=root_value,\n                    operation_name=message.payload.operationName,\n                )\n\n            result_source = get_result_source()\n\n        # Handle initial validation errors\n        if isinstance(result_source, GraphQLExecutionResult):\n            assert result_source.errors\n            payload = [format_graphql_error(result_source.errors[0])]\n            await self.send_message(ErrorMessage(id=message.id, payload=payload))\n            self.schema.process_errors(result_source.errors)\n            return\n\n        # Create task to handle this subscription, reserve the operation ID\n        self.subscriptions[message.id] = result_source\n        self.tasks[message.id] = asyncio.create_task(\n            self.operation_task(result_source, message.id)\n        )\n\n    async def operation_task(\n        self, result_source: AsyncGenerator, operation_id: str\n    ) -> None:\n        \"\"\"\n        Operation task top level method.  Cleans up and de-registers the operation\n        once it is done.\n        \"\"\"\n        try:\n            await self.handle_async_results(result_source, operation_id)\n        except BaseException:  # pragma: no cover\n            # cleanup in case of something really unexpected\n            del self.subscriptions[operation_id]\n            del self.tasks[operation_id]\n            raise\n        else:\n            # de-register the operation _before_ sending the `Complete` message\n            # to make the `operation_id` immediately available for re-use\n            del self.subscriptions[operation_id]\n            del self.tasks[operation_id]\n            await self.send_message(CompleteMessage(id=operation_id))\n        finally:\n            # add this task to a list to be reaped later\n            task = asyncio.current_task()\n            assert task is not None\n            self.completed_tasks.append(task)\n\n    async def handle_async_results(\n        self,\n        result_source: AsyncGenerator,\n        operation_id: str,\n    ) -> None:\n        try:\n            async for result in result_source:\n                if result.errors:\n                    error_payload = [format_graphql_error(err) for err in result.errors]\n                    error_message = ErrorMessage(id=operation_id, payload=error_payload)\n                    await self.send_message(error_message)\n                    self.schema.process_errors(result.errors)\n                    return\n                else:\n                    next_payload = {\"data\": result.data}\n                    next_message = NextMessage(id=operation_id, payload=next_payload)\n                    await self.send_message(next_message)\n        except asyncio.CancelledError:\n            # CancelledErrors are expected during task cleanup.\n            return\n        except Exception as error:\n            # GraphQLErrors are handled by graphql-core and included in the\n            # ExecutionResult\n            error = GraphQLError(str(error), original_error=error)\n            error_payload = [format_graphql_error(error)]\n            error_message = ErrorMessage(id=operation_id, payload=error_payload)\n            await self.send_message(error_message)\n            self.schema.process_errors([error])\n            return\n\n    async def handle_complete(self, message: CompleteMessage) -> None:\n        await self.cleanup_operation(operation_id=message.id)\n\n    async def handle_invalid_message(self, error_message: str) -> None:\n        await self.close(code=4400, reason=error_message)\n\n    async def send_message(self, message: GraphQLTransportMessage) -> None:\n        data = message.as_dict()\n        await self.send_json(data)\n\n    async def cleanup_operation(self, operation_id: str) -> None:\n        if operation_id not in self.subscriptions:\n            return\n        generator = self.subscriptions.pop(operation_id)\n        task = self.tasks.pop(operation_id)\n        # since python 3.8, generators cannot be reliably closed\n        with suppress(RuntimeError):\n            await generator.aclose()\n        task.cancel()\n        with suppress(BaseException):\n            await task\n\n    async def reap_completed_tasks(self) -> None:\n        \"\"\"\n        Await tasks that have completed\n        \"\"\"\n        tasks, self.completed_tasks = self.completed_tasks, []\n        for task in tasks:\n            with suppress(BaseException):\n                await task\n", "path": "strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py"}], "after_files": [{"content": "import asyncio\nfrom abc import ABC, abstractmethod\nfrom contextlib import suppress\nfrom datetime import timedelta\nfrom typing import Any, AsyncGenerator, Callable, Dict, List, Optional\n\nfrom graphql import (\n    ExecutionResult as GraphQLExecutionResult,\n    GraphQLError,\n    GraphQLSyntaxError,\n    parse,\n)\nfrom graphql.error.graphql_error import format_error as format_graphql_error\n\nfrom strawberry.schema import BaseSchema\nfrom strawberry.subscriptions.protocols.graphql_transport_ws.types import (\n    CompleteMessage,\n    ConnectionAckMessage,\n    ConnectionInitMessage,\n    ErrorMessage,\n    GraphQLTransportMessage,\n    NextMessage,\n    PingMessage,\n    PongMessage,\n    SubscribeMessage,\n    SubscribeMessagePayload,\n)\nfrom strawberry.types.graphql import OperationType\nfrom strawberry.utils.debug import pretty_print_graphql_operation\nfrom strawberry.utils.operation import get_operation_type\n\n\nclass BaseGraphQLTransportWSHandler(ABC):\n    def __init__(\n        self,\n        schema: BaseSchema,\n        debug: bool,\n        connection_init_wait_timeout: timedelta,\n    ):\n        self.schema = schema\n        self.debug = debug\n        self.connection_init_wait_timeout = connection_init_wait_timeout\n        self.connection_init_timeout_task: Optional[asyncio.Task] = None\n        self.connection_init_received = False\n        self.connection_acknowledged = False\n        self.subscriptions: Dict[str, AsyncGenerator] = {}\n        self.tasks: Dict[str, asyncio.Task] = {}\n        self.completed_tasks: List[asyncio.Task] = []\n\n    @abstractmethod\n    async def get_context(self) -> Any:\n        \"\"\"Return the operations context\"\"\"\n\n    @abstractmethod\n    async def get_root_value(self) -> Any:\n        \"\"\"Return the schemas root value\"\"\"\n\n    @abstractmethod\n    async def send_json(self, data: dict) -> None:\n        \"\"\"Send the data JSON encoded to the WebSocket client\"\"\"\n\n    @abstractmethod\n    async def close(self, code: int, reason: str) -> None:\n        \"\"\"Close the WebSocket with the passed code and reason\"\"\"\n\n    @abstractmethod\n    async def handle_request(self) -> Any:\n        \"\"\"Handle the request this instance was created for\"\"\"\n\n    async def handle(self) -> Any:\n        timeout_handler = self.handle_connection_init_timeout()\n        self.connection_init_timeout_task = asyncio.create_task(timeout_handler)\n        return await self.handle_request()\n\n    async def handle_connection_init_timeout(self):\n        delay = self.connection_init_wait_timeout.total_seconds()\n        await asyncio.sleep(delay=delay)\n\n        if self.connection_init_received:\n            return\n\n        reason = \"Connection initialisation timeout\"\n        await self.close(code=4408, reason=reason)\n\n    async def handle_message(self, message: dict):\n        handler: Callable\n        handler_arg: Any\n        try:\n            message_type = message.pop(\"type\")\n\n            if message_type == ConnectionInitMessage.type:\n                handler = self.handle_connection_init\n                handler_arg = ConnectionInitMessage(**message)\n\n            elif message_type == PingMessage.type:\n                handler = self.handle_ping\n                handler_arg = PingMessage(**message)\n\n            elif message_type == PongMessage.type:\n                handler = self.handle_pong\n                handler_arg = PongMessage(**message)\n\n            elif message_type == SubscribeMessage.type:\n                handler = self.handle_subscribe\n                payload = SubscribeMessagePayload(**message.pop(\"payload\"))\n                handler_arg = SubscribeMessage(payload=payload, **message)\n\n            elif message_type == CompleteMessage.type:\n                handler = self.handle_complete\n                handler_arg = CompleteMessage(**message)\n\n            else:\n                handler = self.handle_invalid_message\n                handler_arg = f\"Unknown message type: {message_type}\"\n\n        except (KeyError, TypeError):\n            handler = self.handle_invalid_message\n            handler_arg = \"Failed to parse message\"\n\n        await handler(handler_arg)\n        await self.reap_completed_tasks()\n\n    async def handle_connection_init(self, message: ConnectionInitMessage) -> None:\n        if self.connection_init_received:\n            reason = \"Too many initialisation requests\"\n            await self.close(code=4429, reason=reason)\n            return\n\n        self.connection_init_received = True\n        await self.send_message(ConnectionAckMessage())\n        self.connection_acknowledged = True\n\n    async def handle_ping(self, message: PingMessage) -> None:\n        await self.send_message(PongMessage())\n\n    async def handle_pong(self, message: PongMessage) -> None:\n        pass\n\n    async def handle_subscribe(self, message: SubscribeMessage) -> None:\n        if not self.connection_acknowledged:\n            await self.close(code=4401, reason=\"Unauthorized\")\n            return\n\n        try:\n            graphql_document = parse(message.payload.query)\n        except GraphQLSyntaxError as exc:\n            await self.close(code=4400, reason=exc.message)\n            return\n\n        try:\n            operation_type = get_operation_type(\n                graphql_document, message.payload.operationName\n            )\n        except RuntimeError:\n            await self.close(code=4400, reason=\"Can't get GraphQL operation type\")\n            return\n\n        if message.id in self.subscriptions:\n            reason = f\"Subscriber for {message.id} already exists\"\n            await self.close(code=4409, reason=reason)\n            return\n\n        if self.debug:  # pragma: no cover\n            pretty_print_graphql_operation(\n                message.payload.operationName,\n                message.payload.query,\n                message.payload.variables,\n            )\n\n        context = await self.get_context()\n        root_value = await self.get_root_value()\n\n        # Get an AsyncGenerator yielding the results\n        if operation_type == OperationType.SUBSCRIPTION:\n            result_source = await self.schema.subscribe(\n                query=message.payload.query,\n                variable_values=message.payload.variables,\n                operation_name=message.payload.operationName,\n                context_value=context,\n                root_value=root_value,\n            )\n        else:\n            # create AsyncGenerator returning a single result\n            async def get_result_source():\n                yield await self.schema.execute(\n                    query=message.payload.query,\n                    variable_values=message.payload.variables,\n                    context_value=context,\n                    root_value=root_value,\n                    operation_name=message.payload.operationName,\n                )\n\n            result_source = get_result_source()\n\n        # Handle initial validation errors\n        if isinstance(result_source, GraphQLExecutionResult):\n            assert result_source.errors\n            payload = [format_graphql_error(result_source.errors[0])]\n            await self.send_message(ErrorMessage(id=message.id, payload=payload))\n            self.schema.process_errors(result_source.errors)\n            return\n\n        # Create task to handle this subscription, reserve the operation ID\n        self.subscriptions[message.id] = result_source\n        self.tasks[message.id] = asyncio.create_task(\n            self.operation_task(result_source, message.id)\n        )\n\n    async def operation_task(\n        self, result_source: AsyncGenerator, operation_id: str\n    ) -> None:\n        \"\"\"\n        Operation task top level method.  Cleans up and de-registers the operation\n        once it is done.\n        \"\"\"\n        try:\n            await self.handle_async_results(result_source, operation_id)\n        except BaseException:  # pragma: no cover\n            # cleanup in case of something really unexpected\n            # wait for generator to be closed to ensure that any existing\n            # 'finally' statement is called\n            result_source = self.subscriptions[operation_id]\n            with suppress(RuntimeError):\n                await result_source.aclose()\n            del self.subscriptions[operation_id]\n            del self.tasks[operation_id]\n            raise\n        else:\n            # de-register the operation _before_ sending the `Complete` message\n            # to make the `operation_id` immediately available for re-use\n            del self.subscriptions[operation_id]\n            del self.tasks[operation_id]\n            await self.send_message(CompleteMessage(id=operation_id))\n        finally:\n            # add this task to a list to be reaped later\n            task = asyncio.current_task()\n            assert task is not None\n            self.completed_tasks.append(task)\n\n    async def handle_async_results(\n        self,\n        result_source: AsyncGenerator,\n        operation_id: str,\n    ) -> None:\n        try:\n            async for result in result_source:\n                if result.errors:\n                    error_payload = [format_graphql_error(err) for err in result.errors]\n                    error_message = ErrorMessage(id=operation_id, payload=error_payload)\n                    await self.send_message(error_message)\n                    self.schema.process_errors(result.errors)\n                    return\n                else:\n                    next_payload = {\"data\": result.data}\n                    next_message = NextMessage(id=operation_id, payload=next_payload)\n                    await self.send_message(next_message)\n        except asyncio.CancelledError:\n            # CancelledErrors are expected during task cleanup.\n            return\n        except Exception as error:\n            # GraphQLErrors are handled by graphql-core and included in the\n            # ExecutionResult\n            error = GraphQLError(str(error), original_error=error)\n            error_payload = [format_graphql_error(error)]\n            error_message = ErrorMessage(id=operation_id, payload=error_payload)\n            await self.send_message(error_message)\n            self.schema.process_errors([error])\n            return\n\n    async def handle_complete(self, message: CompleteMessage) -> None:\n        await self.cleanup_operation(operation_id=message.id)\n\n    async def handle_invalid_message(self, error_message: str) -> None:\n        await self.close(code=4400, reason=error_message)\n\n    async def send_message(self, message: GraphQLTransportMessage) -> None:\n        data = message.as_dict()\n        await self.send_json(data)\n\n    async def cleanup_operation(self, operation_id: str) -> None:\n        if operation_id not in self.subscriptions:\n            return\n        result_source = self.subscriptions.pop(operation_id)\n        task = self.tasks.pop(operation_id)\n        task.cancel()\n        with suppress(BaseException):\n            await task\n        # since python 3.8, generators cannot be reliably closed\n        with suppress(RuntimeError):\n            await result_source.aclose()\n\n    async def reap_completed_tasks(self) -> None:\n        \"\"\"\n        Await tasks that have completed\n        \"\"\"\n        tasks, self.completed_tasks = self.completed_tasks, []\n        for task in tasks:\n            with suppress(BaseException):\n                await task\n", "path": "strawberry/subscriptions/protocols/graphql_transport_ws/handlers.py"}]}