We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
TypeError when connection is closed in connect()
I have the same problem then #185 but I think, I can reproduce it.

When you close a connection in the connect() method of a WebsocketConsumer, it works fine with daphne but uvicorn raises a TypeError 

```
Traceback (most recent call last):
  File "/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/uvicorn/protocols/websockets/websockets_impl.py", line 140, in run_asgi
    result = await asgi(self.asgi_receive, self.asgi_send)
  File "/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/channels/sessions.py", line 179, in __call__
    return await self.inner(receive, self.send)
  File "/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/channels/middleware.py", line 41, in coroutine_call
    await inner_instance(receive, send)
  File "/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/channels/consumer.py", line 59, in __call__
    [receive, self.channel_receive], self.dispatch
  File "/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/channels/utils.py", line 59, in await_many_dispatch
    await task
  File "/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/channels/utils.py", line 51, in await_many_dispatch
    result = task.result()
  File "/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/uvicorn/protocols/websockets/websockets_impl.py", line 222, in asgi_receive
    data = await self.recv()
  File "/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/websockets/protocol.py", line 419, in recv
    return_when=asyncio.FIRST_COMPLETED,
  File "/usr/lib/python3.7/asyncio/tasks.py", line 361, in wait
    fs = {ensure_future(f, loop=loop) for f in set(fs)}
  File "/usr/lib/python3.7/asyncio/tasks.py", line 361, in <setcomp>
    fs = {ensure_future(f, loop=loop) for f in set(fs)}
  File "/usr/lib/python3.7/asyncio/tasks.py", line 592, in ensure_future
    raise TypeError('An asyncio.Future, a coroutine or an awaitable is '
TypeError: An asyncio.Future, a coroutine or an awaitable is required
```

I created this branch on the django-channels-example repository: https://github.com/ostcar/channels-examples/tree/uvicorn-test

All I changed was, that the websocket connection is always closed: https://github.com/andrewgodwin/channels-examples/compare/master...ostcar:uvicorn-test

I think the problem is, that you call `websockets.WebSocketServerProtocol.recv()` before the websocket connection is open so `websockets.WebSocketServerProtocol.connection_open()` was not called. In this case the attribute `transfer_data_task` is still None. So when `websockets.WebSocketServerProtocol.recv()` calls 

```python
yield from asyncio.wait(
                    [pop_message_waiter, self.transfer_data_task],
                    loop=self.loop,
                    return_when=asyncio.FIRST_COMPLETED,
                )
```
it tries to await `None` what causes the exception.

I don't know, if this is a but in the websockets package or if you should check `self.closed` before calling `self.recv()`


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `uvicorn/protocols/websockets/websockets_impl.py`
Content:
```
1 import asyncio
2 import http
3 import logging
4 from urllib.parse import unquote
5 
6 import websockets
7 
8 from uvicorn.protocols.utils import get_local_addr, get_remote_addr, is_ssl
9 
10 
11 class Server:
12     closing = False
13 
14     def register(self, ws):
15         pass
16 
17     def unregister(self, ws):
18         pass
19 
20     def is_serving(self):
21         return not self.closing
22 
23 
24 class WebSocketProtocol(websockets.WebSocketServerProtocol):
25     def __init__(self, config, server_state, _loop=None):
26         if not config.loaded:
27             config.load()
28 
29         self.config = config
30         self.app = config.loaded_app
31         self.loop = _loop or asyncio.get_event_loop()
32         self.logger = logging.getLogger("uvicorn.error")
33         self.root_path = config.root_path
34 
35         # Shared server state
36         self.connections = server_state.connections
37         self.tasks = server_state.tasks
38 
39         # Connection state
40         self.transport = None
41         self.server = None
42         self.client = None
43         self.scheme = None
44 
45         # Connection events
46         self.scope = None
47         self.handshake_started_event = asyncio.Event()
48         self.handshake_completed_event = asyncio.Event()
49         self.closed_event = asyncio.Event()
50         self.initial_response = None
51         self.connect_sent = False
52         self.accepted_subprotocol = None
53         self.transfer_data_task = None
54 
55         self.ws_server = Server()
56 
57         super().__init__(ws_handler=self.ws_handler, ws_server=self.ws_server)
58 
59     def connection_made(self, transport):
60         self.connections.add(self)
61         self.transport = transport
62         self.server = get_local_addr(transport)
63         self.client = get_remote_addr(transport)
64         self.scheme = "wss" if is_ssl(transport) else "ws"
65         super().connection_made(transport)
66 
67     def connection_lost(self, exc):
68         self.connections.remove(self)
69         self.handshake_completed_event.set()
70         super().connection_lost(exc)
71 
72     def shutdown(self):
73         self.ws_server.closing = True
74         self.transport.close()
75 
76     def on_task_complete(self, task):
77         self.tasks.discard(task)
78 
79     async def process_request(self, path, headers):
80         """
81         This hook is called to determine if the websocket should return
82         an HTTP response and close.
83 
84         Our behavior here is to start the ASGI application, and then wait
85         for either `accept` or `close` in order to determine if we should
86         close the connection.
87         """
88         path_portion, _, query_string = path.partition("?")
89 
90         websockets.handshake.check_request(headers)
91 
92         subprotocols = []
93         for header in headers.get_all("Sec-WebSocket-Protocol"):
94             subprotocols.extend([token.strip() for token in header.split(",")])
95 
96         asgi_headers = [
97             (name.encode("ascii"), value.encode("ascii"))
98             for name, value in headers.raw_items()
99         ]
100 
101         self.scope = {
102             "type": "websocket",
103             "asgi": {"version": self.config.asgi_version, "spec_version": "2.1"},
104             "scheme": self.scheme,
105             "server": self.server,
106             "client": self.client,
107             "root_path": self.root_path,
108             "path": unquote(path_portion),
109             "raw_path": path_portion,
110             "query_string": query_string.encode("ascii"),
111             "headers": asgi_headers,
112             "subprotocols": subprotocols,
113         }
114         task = self.loop.create_task(self.run_asgi())
115         task.add_done_callback(self.on_task_complete)
116         self.tasks.add(task)
117         await self.handshake_started_event.wait()
118         return self.initial_response
119 
120     def process_subprotocol(self, headers, available_subprotocols):
121         """
122         We override the standard 'process_subprotocol' behavior here so that
123         we return whatever subprotocol is sent in the 'accept' message.
124         """
125         return self.accepted_subprotocol
126 
127     def send_500_response(self):
128         msg = b"Internal Server Error"
129         content = [
130             b"HTTP/1.1 500 Internal Server Error\r\n"
131             b"content-type: text/plain; charset=utf-8\r\n",
132             b"content-length: " + str(len(msg)).encode("ascii") + b"\r\n",
133             b"connection: close\r\n",
134             b"\r\n",
135             msg,
136         ]
137         self.transport.write(b"".join(content))
138 
139     async def ws_handler(self, protocol, path):
140         """
141         This is the main handler function for the 'websockets' implementation
142         to call into. We just wait for close then return, and instead allow
143         'send' and 'receive' events to drive the flow.
144         """
145         self.handshake_completed_event.set()
146         await self.closed_event.wait()
147 
148     async def run_asgi(self):
149         """
150         Wrapper around the ASGI callable, handling exceptions and unexpected
151         termination states.
152         """
153         try:
154             result = await self.app(self.scope, self.asgi_receive, self.asgi_send)
155         except BaseException as exc:
156             self.closed_event.set()
157             msg = "Exception in ASGI application\n"
158             self.logger.error(msg, exc_info=exc)
159             if not self.handshake_started_event.is_set():
160                 self.send_500_response()
161             else:
162                 await self.handshake_completed_event.wait()
163             self.transport.close()
164         else:
165             self.closed_event.set()
166             if not self.handshake_started_event.is_set():
167                 msg = "ASGI callable returned without sending handshake."
168                 self.logger.error(msg)
169                 self.send_500_response()
170                 self.transport.close()
171             elif result is not None:
172                 msg = "ASGI callable should return None, but returned '%s'."
173                 self.logger.error(msg, result)
174                 await self.handshake_completed_event.wait()
175                 self.transport.close()
176 
177     async def asgi_send(self, message):
178         message_type = message["type"]
179 
180         if not self.handshake_started_event.is_set():
181             if message_type == "websocket.accept":
182                 self.logger.info(
183                     '%s - "WebSocket %s" [accepted]',
184                     self.scope["client"],
185                     self.scope["root_path"] + self.scope["path"],
186                 )
187                 self.initial_response = None
188                 self.accepted_subprotocol = message.get("subprotocol")
189                 self.handshake_started_event.set()
190 
191             elif message_type == "websocket.close":
192                 self.logger.info(
193                     '%s - "WebSocket %s" 403',
194                     self.scope["client"],
195                     self.scope["root_path"] + self.scope["path"],
196                 )
197                 self.initial_response = (http.HTTPStatus.FORBIDDEN, [], b"")
198                 self.handshake_started_event.set()
199                 self.closed_event.set()
200 
201             else:
202                 msg = (
203                     "Expected ASGI message 'websocket.accept' or 'websocket.close', "
204                     "but got '%s'."
205                 )
206                 raise RuntimeError(msg % message_type)
207 
208         elif not self.closed_event.is_set():
209             await self.handshake_completed_event.wait()
210 
211             if message_type == "websocket.send":
212                 bytes_data = message.get("bytes")
213                 text_data = message.get("text")
214                 data = text_data if bytes_data is None else bytes_data
215                 await self.send(data)
216 
217             elif message_type == "websocket.close":
218                 code = message.get("code", 1000)
219                 self.close_code = code  # for WebSocketServerProtocol
220                 await self.close(code)
221                 self.closed_event.set()
222 
223             else:
224                 msg = (
225                     "Expected ASGI message 'websocket.send' or 'websocket.close',"
226                     " but got '%s'."
227                 )
228                 raise RuntimeError(msg % message_type)
229 
230         else:
231             msg = "Unexpected ASGI message '%s', after sending 'websocket.close'."
232             raise RuntimeError(msg % message_type)
233 
234     async def asgi_receive(self):
235         if not self.connect_sent:
236             self.connect_sent = True
237             return {"type": "websocket.connect"}
238 
239         await self.handshake_completed_event.wait()
240 
241         if self.closed_event.is_set():
242             # If the client disconnected: WebSocketServerProtocol set self.close_code.
243             # If the handshake failed or the app closed before handshake completion,
244             # use 1006 Abnormal Closure.
245             code = getattr(self, "close_code", 1006)
246             return {"type": "websocket.disconnect", "code": code}
247 
248         try:
249             data = await self.recv()
250         except websockets.ConnectionClosed as exc:
251             return {"type": "websocket.disconnect", "code": exc.code}
252 
253         msg = {"type": "websocket.receive"}
254 
255         if isinstance(data, str):
256             msg["text"] = data
257         else:
258             msg["bytes"] = data
259 
260         return msg
261
```
--- 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/uvicorn/protocols/websockets/websockets_impl.py b/uvicorn/protocols/websockets/websockets_impl.py
--- a/uvicorn/protocols/websockets/websockets_impl.py
+++ b/uvicorn/protocols/websockets/websockets_impl.py
@@ -216,7 +216,6 @@
 
             elif message_type == "websocket.close":
                 code = message.get("code", 1000)
-                self.close_code = code  # for WebSocketServerProtocol
                 await self.close(code)
                 self.closed_event.set()
 
@@ -237,15 +236,8 @@
             return {"type": "websocket.connect"}
 
         await self.handshake_completed_event.wait()
-
-        if self.closed_event.is_set():
-            # If the client disconnected: WebSocketServerProtocol set self.close_code.
-            # If the handshake failed or the app closed before handshake completion,
-            # use 1006 Abnormal Closure.
-            code = getattr(self, "close_code", 1006)
-            return {"type": "websocket.disconnect", "code": code}
-
         try:
+            await self.ensure_open()
             data = await self.recv()
         except websockets.ConnectionClosed as exc:
             return {"type": "websocket.disconnect", "code": exc.code}

{"golden_diff": "diff --git a/uvicorn/protocols/websockets/websockets_impl.py b/uvicorn/protocols/websockets/websockets_impl.py\n--- a/uvicorn/protocols/websockets/websockets_impl.py\n+++ b/uvicorn/protocols/websockets/websockets_impl.py\n@@ -216,7 +216,6 @@\n \n             elif message_type == \"websocket.close\":\n                 code = message.get(\"code\", 1000)\n-                self.close_code = code  # for WebSocketServerProtocol\n                 await self.close(code)\n                 self.closed_event.set()\n \n@@ -237,15 +236,8 @@\n             return {\"type\": \"websocket.connect\"}\n \n         await self.handshake_completed_event.wait()\n-\n-        if self.closed_event.is_set():\n-            # If the client disconnected: WebSocketServerProtocol set self.close_code.\n-            # If the handshake failed or the app closed before handshake completion,\n-            # use 1006 Abnormal Closure.\n-            code = getattr(self, \"close_code\", 1006)\n-            return {\"type\": \"websocket.disconnect\", \"code\": code}\n-\n         try:\n+            await self.ensure_open()\n             data = await self.recv()\n         except websockets.ConnectionClosed as exc:\n             return {\"type\": \"websocket.disconnect\", \"code\": exc.code}\n", "issue": "TypeError when connection is closed in connect()\nI have the same problem then #185 but I think, I can reproduce it.\r\n\r\nWhen you close a connection in the connect() method of a WebsocketConsumer, it works fine with daphne but uvicorn raises a TypeError \r\n\r\n```\r\nTraceback (most recent call last):\r\n  File \"/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/uvicorn/protocols/websockets/websockets_impl.py\", line 140, in run_asgi\r\n    result = await asgi(self.asgi_receive, self.asgi_send)\r\n  File \"/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/channels/sessions.py\", line 179, in __call__\r\n    return await self.inner(receive, self.send)\r\n  File \"/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/channels/middleware.py\", line 41, in coroutine_call\r\n    await inner_instance(receive, send)\r\n  File \"/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/channels/consumer.py\", line 59, in __call__\r\n    [receive, self.channel_receive], self.dispatch\r\n  File \"/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/channels/utils.py\", line 59, in await_many_dispatch\r\n    await task\r\n  File \"/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/channels/utils.py\", line 51, in await_many_dispatch\r\n    result = task.result()\r\n  File \"/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/uvicorn/protocols/websockets/websockets_impl.py\", line 222, in asgi_receive\r\n    data = await self.recv()\r\n  File \"/home/ossi/src/channels-examples/.venv/lib/python3.7/site-packages/websockets/protocol.py\", line 419, in recv\r\n    return_when=asyncio.FIRST_COMPLETED,\r\n  File \"/usr/lib/python3.7/asyncio/tasks.py\", line 361, in wait\r\n    fs = {ensure_future(f, loop=loop) for f in set(fs)}\r\n  File \"/usr/lib/python3.7/asyncio/tasks.py\", line 361, in <setcomp>\r\n    fs = {ensure_future(f, loop=loop) for f in set(fs)}\r\n  File \"/usr/lib/python3.7/asyncio/tasks.py\", line 592, in ensure_future\r\n    raise TypeError('An asyncio.Future, a coroutine or an awaitable is '\r\nTypeError: An asyncio.Future, a coroutine or an awaitable is required\r\n```\r\n\r\nI created this branch on the django-channels-example repository: https://github.com/ostcar/channels-examples/tree/uvicorn-test\r\n\r\nAll I changed was, that the websocket connection is always closed: https://github.com/andrewgodwin/channels-examples/compare/master...ostcar:uvicorn-test\r\n\r\nI think the problem is, that you call `websockets.WebSocketServerProtocol.recv()` before the websocket connection is open so `websockets.WebSocketServerProtocol.connection_open()` was not called. In this case the attribute `transfer_data_task` is still None. So when `websockets.WebSocketServerProtocol.recv()` calls \r\n\r\n```python\r\nyield from asyncio.wait(\r\n                    [pop_message_waiter, self.transfer_data_task],\r\n                    loop=self.loop,\r\n                    return_when=asyncio.FIRST_COMPLETED,\r\n                )\r\n```\r\nit tries to await `None` what causes the exception.\r\n\r\nI don't know, if this is a but in the websockets package or if you should check `self.closed` before calling `self.recv()`\r\n\n", "before_files": [{"content": "import asyncio\nimport http\nimport logging\nfrom urllib.parse import unquote\n\nimport websockets\n\nfrom uvicorn.protocols.utils import get_local_addr, get_remote_addr, is_ssl\n\n\nclass Server:\n    closing = False\n\n    def register(self, ws):\n        pass\n\n    def unregister(self, ws):\n        pass\n\n    def is_serving(self):\n        return not self.closing\n\n\nclass WebSocketProtocol(websockets.WebSocketServerProtocol):\n    def __init__(self, config, server_state, _loop=None):\n        if not config.loaded:\n            config.load()\n\n        self.config = config\n        self.app = config.loaded_app\n        self.loop = _loop or asyncio.get_event_loop()\n        self.logger = logging.getLogger(\"uvicorn.error\")\n        self.root_path = config.root_path\n\n        # Shared server state\n        self.connections = server_state.connections\n        self.tasks = server_state.tasks\n\n        # Connection state\n        self.transport = None\n        self.server = None\n        self.client = None\n        self.scheme = None\n\n        # Connection events\n        self.scope = None\n        self.handshake_started_event = asyncio.Event()\n        self.handshake_completed_event = asyncio.Event()\n        self.closed_event = asyncio.Event()\n        self.initial_response = None\n        self.connect_sent = False\n        self.accepted_subprotocol = None\n        self.transfer_data_task = None\n\n        self.ws_server = Server()\n\n        super().__init__(ws_handler=self.ws_handler, ws_server=self.ws_server)\n\n    def connection_made(self, transport):\n        self.connections.add(self)\n        self.transport = transport\n        self.server = get_local_addr(transport)\n        self.client = get_remote_addr(transport)\n        self.scheme = \"wss\" if is_ssl(transport) else \"ws\"\n        super().connection_made(transport)\n\n    def connection_lost(self, exc):\n        self.connections.remove(self)\n        self.handshake_completed_event.set()\n        super().connection_lost(exc)\n\n    def shutdown(self):\n        self.ws_server.closing = True\n        self.transport.close()\n\n    def on_task_complete(self, task):\n        self.tasks.discard(task)\n\n    async def process_request(self, path, headers):\n        \"\"\"\n        This hook is called to determine if the websocket should return\n        an HTTP response and close.\n\n        Our behavior here is to start the ASGI application, and then wait\n        for either `accept` or `close` in order to determine if we should\n        close the connection.\n        \"\"\"\n        path_portion, _, query_string = path.partition(\"?\")\n\n        websockets.handshake.check_request(headers)\n\n        subprotocols = []\n        for header in headers.get_all(\"Sec-WebSocket-Protocol\"):\n            subprotocols.extend([token.strip() for token in header.split(\",\")])\n\n        asgi_headers = [\n            (name.encode(\"ascii\"), value.encode(\"ascii\"))\n            for name, value in headers.raw_items()\n        ]\n\n        self.scope = {\n            \"type\": \"websocket\",\n            \"asgi\": {\"version\": self.config.asgi_version, \"spec_version\": \"2.1\"},\n            \"scheme\": self.scheme,\n            \"server\": self.server,\n            \"client\": self.client,\n            \"root_path\": self.root_path,\n            \"path\": unquote(path_portion),\n            \"raw_path\": path_portion,\n            \"query_string\": query_string.encode(\"ascii\"),\n            \"headers\": asgi_headers,\n            \"subprotocols\": subprotocols,\n        }\n        task = self.loop.create_task(self.run_asgi())\n        task.add_done_callback(self.on_task_complete)\n        self.tasks.add(task)\n        await self.handshake_started_event.wait()\n        return self.initial_response\n\n    def process_subprotocol(self, headers, available_subprotocols):\n        \"\"\"\n        We override the standard 'process_subprotocol' behavior here so that\n        we return whatever subprotocol is sent in the 'accept' message.\n        \"\"\"\n        return self.accepted_subprotocol\n\n    def send_500_response(self):\n        msg = b\"Internal Server Error\"\n        content = [\n            b\"HTTP/1.1 500 Internal Server Error\\r\\n\"\n            b\"content-type: text/plain; charset=utf-8\\r\\n\",\n            b\"content-length: \" + str(len(msg)).encode(\"ascii\") + b\"\\r\\n\",\n            b\"connection: close\\r\\n\",\n            b\"\\r\\n\",\n            msg,\n        ]\n        self.transport.write(b\"\".join(content))\n\n    async def ws_handler(self, protocol, path):\n        \"\"\"\n        This is the main handler function for the 'websockets' implementation\n        to call into. We just wait for close then return, and instead allow\n        'send' and 'receive' events to drive the flow.\n        \"\"\"\n        self.handshake_completed_event.set()\n        await self.closed_event.wait()\n\n    async def run_asgi(self):\n        \"\"\"\n        Wrapper around the ASGI callable, handling exceptions and unexpected\n        termination states.\n        \"\"\"\n        try:\n            result = await self.app(self.scope, self.asgi_receive, self.asgi_send)\n        except BaseException as exc:\n            self.closed_event.set()\n            msg = \"Exception in ASGI application\\n\"\n            self.logger.error(msg, exc_info=exc)\n            if not self.handshake_started_event.is_set():\n                self.send_500_response()\n            else:\n                await self.handshake_completed_event.wait()\n            self.transport.close()\n        else:\n            self.closed_event.set()\n            if not self.handshake_started_event.is_set():\n                msg = \"ASGI callable returned without sending handshake.\"\n                self.logger.error(msg)\n                self.send_500_response()\n                self.transport.close()\n            elif result is not None:\n                msg = \"ASGI callable should return None, but returned '%s'.\"\n                self.logger.error(msg, result)\n                await self.handshake_completed_event.wait()\n                self.transport.close()\n\n    async def asgi_send(self, message):\n        message_type = message[\"type\"]\n\n        if not self.handshake_started_event.is_set():\n            if message_type == \"websocket.accept\":\n                self.logger.info(\n                    '%s - \"WebSocket %s\" [accepted]',\n                    self.scope[\"client\"],\n                    self.scope[\"root_path\"] + self.scope[\"path\"],\n                )\n                self.initial_response = None\n                self.accepted_subprotocol = message.get(\"subprotocol\")\n                self.handshake_started_event.set()\n\n            elif message_type == \"websocket.close\":\n                self.logger.info(\n                    '%s - \"WebSocket %s\" 403',\n                    self.scope[\"client\"],\n                    self.scope[\"root_path\"] + self.scope[\"path\"],\n                )\n                self.initial_response = (http.HTTPStatus.FORBIDDEN, [], b\"\")\n                self.handshake_started_event.set()\n                self.closed_event.set()\n\n            else:\n                msg = (\n                    \"Expected ASGI message 'websocket.accept' or 'websocket.close', \"\n                    \"but got '%s'.\"\n                )\n                raise RuntimeError(msg % message_type)\n\n        elif not self.closed_event.is_set():\n            await self.handshake_completed_event.wait()\n\n            if message_type == \"websocket.send\":\n                bytes_data = message.get(\"bytes\")\n                text_data = message.get(\"text\")\n                data = text_data if bytes_data is None else bytes_data\n                await self.send(data)\n\n            elif message_type == \"websocket.close\":\n                code = message.get(\"code\", 1000)\n                self.close_code = code  # for WebSocketServerProtocol\n                await self.close(code)\n                self.closed_event.set()\n\n            else:\n                msg = (\n                    \"Expected ASGI message 'websocket.send' or 'websocket.close',\"\n                    \" but got '%s'.\"\n                )\n                raise RuntimeError(msg % message_type)\n\n        else:\n            msg = \"Unexpected ASGI message '%s', after sending 'websocket.close'.\"\n            raise RuntimeError(msg % message_type)\n\n    async def asgi_receive(self):\n        if not self.connect_sent:\n            self.connect_sent = True\n            return {\"type\": \"websocket.connect\"}\n\n        await self.handshake_completed_event.wait()\n\n        if self.closed_event.is_set():\n            # If the client disconnected: WebSocketServerProtocol set self.close_code.\n            # If the handshake failed or the app closed before handshake completion,\n            # use 1006 Abnormal Closure.\n            code = getattr(self, \"close_code\", 1006)\n            return {\"type\": \"websocket.disconnect\", \"code\": code}\n\n        try:\n            data = await self.recv()\n        except websockets.ConnectionClosed as exc:\n            return {\"type\": \"websocket.disconnect\", \"code\": exc.code}\n\n        msg = {\"type\": \"websocket.receive\"}\n\n        if isinstance(data, str):\n            msg[\"text\"] = data\n        else:\n            msg[\"bytes\"] = data\n\n        return msg\n", "path": "uvicorn/protocols/websockets/websockets_impl.py"}], "after_files": [{"content": "import asyncio\nimport http\nimport logging\nfrom urllib.parse import unquote\n\nimport websockets\n\nfrom uvicorn.protocols.utils import get_local_addr, get_remote_addr, is_ssl\n\n\nclass Server:\n    closing = False\n\n    def register(self, ws):\n        pass\n\n    def unregister(self, ws):\n        pass\n\n    def is_serving(self):\n        return not self.closing\n\n\nclass WebSocketProtocol(websockets.WebSocketServerProtocol):\n    def __init__(self, config, server_state, _loop=None):\n        if not config.loaded:\n            config.load()\n\n        self.config = config\n        self.app = config.loaded_app\n        self.loop = _loop or asyncio.get_event_loop()\n        self.logger = logging.getLogger(\"uvicorn.error\")\n        self.root_path = config.root_path\n\n        # Shared server state\n        self.connections = server_state.connections\n        self.tasks = server_state.tasks\n\n        # Connection state\n        self.transport = None\n        self.server = None\n        self.client = None\n        self.scheme = None\n\n        # Connection events\n        self.scope = None\n        self.handshake_started_event = asyncio.Event()\n        self.handshake_completed_event = asyncio.Event()\n        self.closed_event = asyncio.Event()\n        self.initial_response = None\n        self.connect_sent = False\n        self.accepted_subprotocol = None\n        self.transfer_data_task = None\n\n        self.ws_server = Server()\n\n        super().__init__(ws_handler=self.ws_handler, ws_server=self.ws_server)\n\n    def connection_made(self, transport):\n        self.connections.add(self)\n        self.transport = transport\n        self.server = get_local_addr(transport)\n        self.client = get_remote_addr(transport)\n        self.scheme = \"wss\" if is_ssl(transport) else \"ws\"\n        super().connection_made(transport)\n\n    def connection_lost(self, exc):\n        self.connections.remove(self)\n        self.handshake_completed_event.set()\n        super().connection_lost(exc)\n\n    def shutdown(self):\n        self.ws_server.closing = True\n        self.transport.close()\n\n    def on_task_complete(self, task):\n        self.tasks.discard(task)\n\n    async def process_request(self, path, headers):\n        \"\"\"\n        This hook is called to determine if the websocket should return\n        an HTTP response and close.\n\n        Our behavior here is to start the ASGI application, and then wait\n        for either `accept` or `close` in order to determine if we should\n        close the connection.\n        \"\"\"\n        path_portion, _, query_string = path.partition(\"?\")\n\n        websockets.handshake.check_request(headers)\n\n        subprotocols = []\n        for header in headers.get_all(\"Sec-WebSocket-Protocol\"):\n            subprotocols.extend([token.strip() for token in header.split(\",\")])\n\n        asgi_headers = [\n            (name.encode(\"ascii\"), value.encode(\"ascii\"))\n            for name, value in headers.raw_items()\n        ]\n\n        self.scope = {\n            \"type\": \"websocket\",\n            \"asgi\": {\"version\": self.config.asgi_version, \"spec_version\": \"2.1\"},\n            \"scheme\": self.scheme,\n            \"server\": self.server,\n            \"client\": self.client,\n            \"root_path\": self.root_path,\n            \"path\": unquote(path_portion),\n            \"raw_path\": path_portion,\n            \"query_string\": query_string.encode(\"ascii\"),\n            \"headers\": asgi_headers,\n            \"subprotocols\": subprotocols,\n        }\n        task = self.loop.create_task(self.run_asgi())\n        task.add_done_callback(self.on_task_complete)\n        self.tasks.add(task)\n        await self.handshake_started_event.wait()\n        return self.initial_response\n\n    def process_subprotocol(self, headers, available_subprotocols):\n        \"\"\"\n        We override the standard 'process_subprotocol' behavior here so that\n        we return whatever subprotocol is sent in the 'accept' message.\n        \"\"\"\n        return self.accepted_subprotocol\n\n    def send_500_response(self):\n        msg = b\"Internal Server Error\"\n        content = [\n            b\"HTTP/1.1 500 Internal Server Error\\r\\n\"\n            b\"content-type: text/plain; charset=utf-8\\r\\n\",\n            b\"content-length: \" + str(len(msg)).encode(\"ascii\") + b\"\\r\\n\",\n            b\"connection: close\\r\\n\",\n            b\"\\r\\n\",\n            msg,\n        ]\n        self.transport.write(b\"\".join(content))\n\n    async def ws_handler(self, protocol, path):\n        \"\"\"\n        This is the main handler function for the 'websockets' implementation\n        to call into. We just wait for close then return, and instead allow\n        'send' and 'receive' events to drive the flow.\n        \"\"\"\n        self.handshake_completed_event.set()\n        await self.closed_event.wait()\n\n    async def run_asgi(self):\n        \"\"\"\n        Wrapper around the ASGI callable, handling exceptions and unexpected\n        termination states.\n        \"\"\"\n        try:\n            result = await self.app(self.scope, self.asgi_receive, self.asgi_send)\n        except BaseException as exc:\n            self.closed_event.set()\n            msg = \"Exception in ASGI application\\n\"\n            self.logger.error(msg, exc_info=exc)\n            if not self.handshake_started_event.is_set():\n                self.send_500_response()\n            else:\n                await self.handshake_completed_event.wait()\n            self.transport.close()\n        else:\n            self.closed_event.set()\n            if not self.handshake_started_event.is_set():\n                msg = \"ASGI callable returned without sending handshake.\"\n                self.logger.error(msg)\n                self.send_500_response()\n                self.transport.close()\n            elif result is not None:\n                msg = \"ASGI callable should return None, but returned '%s'.\"\n                self.logger.error(msg, result)\n                await self.handshake_completed_event.wait()\n                self.transport.close()\n\n    async def asgi_send(self, message):\n        message_type = message[\"type\"]\n\n        if not self.handshake_started_event.is_set():\n            if message_type == \"websocket.accept\":\n                self.logger.info(\n                    '%s - \"WebSocket %s\" [accepted]',\n                    self.scope[\"client\"],\n                    self.scope[\"root_path\"] + self.scope[\"path\"],\n                )\n                self.initial_response = None\n                self.accepted_subprotocol = message.get(\"subprotocol\")\n                self.handshake_started_event.set()\n\n            elif message_type == \"websocket.close\":\n                self.logger.info(\n                    '%s - \"WebSocket %s\" 403',\n                    self.scope[\"client\"],\n                    self.scope[\"root_path\"] + self.scope[\"path\"],\n                )\n                self.initial_response = (http.HTTPStatus.FORBIDDEN, [], b\"\")\n                self.handshake_started_event.set()\n                self.closed_event.set()\n\n            else:\n                msg = (\n                    \"Expected ASGI message 'websocket.accept' or 'websocket.close', \"\n                    \"but got '%s'.\"\n                )\n                raise RuntimeError(msg % message_type)\n\n        elif not self.closed_event.is_set():\n            await self.handshake_completed_event.wait()\n\n            if message_type == \"websocket.send\":\n                bytes_data = message.get(\"bytes\")\n                text_data = message.get(\"text\")\n                data = text_data if bytes_data is None else bytes_data\n                await self.send(data)\n\n            elif message_type == \"websocket.close\":\n                code = message.get(\"code\", 1000)\n                await self.close(code)\n                self.closed_event.set()\n\n            else:\n                msg = (\n                    \"Expected ASGI message 'websocket.send' or 'websocket.close',\"\n                    \" but got '%s'.\"\n                )\n                raise RuntimeError(msg % message_type)\n\n        else:\n            msg = \"Unexpected ASGI message '%s', after sending 'websocket.close'.\"\n            raise RuntimeError(msg % message_type)\n\n    async def asgi_receive(self):\n        if not self.connect_sent:\n            self.connect_sent = True\n            return {\"type\": \"websocket.connect\"}\n\n        await self.handshake_completed_event.wait()\n        try:\n            await self.ensure_open()\n            data = await self.recv()\n        except websockets.ConnectionClosed as exc:\n            return {\"type\": \"websocket.disconnect\", \"code\": exc.code}\n\n        msg = {\"type\": \"websocket.receive\"}\n\n        if isinstance(data, str):\n            msg[\"text\"] = data\n        else:\n            msg[\"bytes\"] = data\n\n        return msg\n", "path": "uvicorn/protocols/websockets/websockets_impl.py"}]}