We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
websocket.disconnect events are not sent to applications when clients don’t close connections properly (wsproto only)
### Checklist

<!-- Please make sure you check all these items before submitting your bug report. -->

- [x] The bug is reproducible against the latest release and/or `master`.
- [x] There are no similar issues or pull requests to fix it yet.

### Describe the bug

When WebSocket connections are not properly closed by clients, uvicorn + wsproto leaves one zombie asyncio tasks running per connection and do not emit disconnect events to applications.

### To reproduce
1. Start any application with uvicorn using wsproto;
2. Create a websocket connection;
3. Do not close the websocket properly;
4. Observe that no `websocket.disconnect` events have been sent to your application;
5. Check for all asyncio tasks (`asyncio.all_tasks()`): there should be a zombie `run_asgi` task for each non properly closed connection.

I’ve setup this repo with instructions to make it easier to reproduce this issue: https://github.com/sephioh/uvicorn-wsproto-issue

### Expected behavior

A [disconnect](https://asgi.readthedocs.io/en/latest/specs/www.html#disconnect-receive-event-ws) event should be sent to applications. 

### Actual behavior

Disconnect event are not being sent to apps when client socket loss happens. 

### Environment

Running uvicorn 0.13.4 with CPython 3.9.0 on Linux
Must be executed with wsproto (—ws wsproto).

### Additional context

https://docs.python.org/3/library/asyncio-protocol.html#asyncio.BaseProtocol.connection_lost

No problems are found when running with websockets impl (—ws websockets).

I’ve opened a PR with a possible fix: https://github.com/encode/uvicorn/pull/996

--- 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/wsproto_impl.py`
Content:
```
1 import asyncio
2 import logging
3 from urllib.parse import unquote
4 
5 import h11
6 import wsproto
7 from wsproto import ConnectionType, events
8 from wsproto.connection import ConnectionState
9 from wsproto.extensions import PerMessageDeflate
10 from wsproto.utilities import RemoteProtocolError
11 
12 from uvicorn.logging import TRACE_LOG_LEVEL
13 from uvicorn.protocols.utils import (
14     get_local_addr,
15     get_path_with_query_string,
16     get_remote_addr,
17     is_ssl,
18 )
19 
20 
21 class WSProtocol(asyncio.Protocol):
22     def __init__(self, config, server_state, _loop=None):
23         if not config.loaded:
24             config.load()
25 
26         self.config = config
27         self.app = config.loaded_app
28         self.loop = _loop or asyncio.get_event_loop()
29         self.logger = logging.getLogger("uvicorn.error")
30         self.root_path = config.root_path
31 
32         # Shared server state
33         self.connections = server_state.connections
34         self.tasks = server_state.tasks
35         self.default_headers = server_state.default_headers
36 
37         # Connection state
38         self.transport = None
39         self.server = None
40         self.client = None
41         self.scheme = None
42 
43         # WebSocket state
44         self.connect_event = None
45         self.queue = asyncio.Queue()
46         self.handshake_complete = False
47         self.close_sent = False
48 
49         self.conn = wsproto.WSConnection(connection_type=ConnectionType.SERVER)
50 
51         self.read_paused = False
52         self.writable = asyncio.Event()
53         self.writable.set()
54 
55         # Buffers
56         self.bytes = b""
57         self.text = ""
58 
59     # Protocol interface
60 
61     def connection_made(self, transport):
62         self.connections.add(self)
63         self.transport = transport
64         self.server = get_local_addr(transport)
65         self.client = get_remote_addr(transport)
66         self.scheme = "wss" if is_ssl(transport) else "ws"
67 
68         if self.logger.level <= TRACE_LOG_LEVEL:
69             prefix = "%s:%d - " % tuple(self.client) if self.client else ""
70             self.logger.log(TRACE_LOG_LEVEL, "%sWebSocket connection made", prefix)
71 
72     def connection_lost(self, exc):
73         if exc is not None:
74             self.queue.put_nowait({"type": "websocket.disconnect"})
75         self.connections.remove(self)
76 
77         if self.logger.level <= TRACE_LOG_LEVEL:
78             prefix = "%s:%d - " % tuple(self.client) if self.client else ""
79             self.logger.log(TRACE_LOG_LEVEL, "%sWebSocket connection lost", prefix)
80 
81         if exc is None:
82             self.transport.close()
83 
84     def eof_received(self):
85         pass
86 
87     def data_received(self, data):
88         try:
89             self.conn.receive_data(data)
90         except RemoteProtocolError as err:
91             if err.event_hint is not None:
92                 self.transport.write(self.conn.send(err.event_hint))
93                 self.transport.close()
94             else:
95                 self.handle_no_connect(events.CloseConnection())
96         else:
97             self.handle_events()
98 
99     def handle_events(self):
100         for event in self.conn.events():
101             if isinstance(event, events.Request):
102                 self.handle_connect(event)
103             elif isinstance(event, events.TextMessage):
104                 self.handle_text(event)
105             elif isinstance(event, events.BytesMessage):
106                 self.handle_bytes(event)
107             elif isinstance(event, events.RejectConnection):
108                 self.handle_no_connect(event)
109             elif isinstance(event, events.RejectData):
110                 self.handle_no_connect(event)
111             elif isinstance(event, events.CloseConnection):
112                 self.handle_close(event)
113             elif isinstance(event, events.Ping):
114                 self.handle_ping(event)
115 
116     def pause_writing(self):
117         """
118         Called by the transport when the write buffer exceeds the high water mark.
119         """
120         self.writable.clear()
121 
122     def resume_writing(self):
123         """
124         Called by the transport when the write buffer drops below the low water mark.
125         """
126         self.writable.set()
127 
128     def shutdown(self):
129         self.queue.put_nowait({"type": "websocket.disconnect", "code": 1012})
130         output = self.conn.send(wsproto.events.CloseConnection(code=1012))
131         self.transport.write(output)
132         self.transport.close()
133 
134     def on_task_complete(self, task):
135         self.tasks.discard(task)
136 
137     # Event handlers
138 
139     def handle_connect(self, event):
140         self.connect_event = event
141         headers = [(b"host", event.host.encode())]
142         headers += [(key.lower(), value) for key, value in event.extra_headers]
143         raw_path, _, query_string = event.target.partition("?")
144         self.scope = {
145             "type": "websocket",
146             "asgi": {"version": self.config.asgi_version, "spec_version": "2.3"},
147             "http_version": "1.1",
148             "scheme": self.scheme,
149             "server": self.server,
150             "client": self.client,
151             "root_path": self.root_path,
152             "path": unquote(raw_path),
153             "raw_path": raw_path.encode("ascii"),
154             "query_string": query_string.encode("ascii"),
155             "headers": headers,
156             "subprotocols": event.subprotocols,
157         }
158         self.queue.put_nowait({"type": "websocket.connect"})
159         task = self.loop.create_task(self.run_asgi())
160         task.add_done_callback(self.on_task_complete)
161         self.tasks.add(task)
162 
163     def handle_no_connect(self, event):
164         headers = [
165             (b"content-type", b"text/plain; charset=utf-8"),
166             (b"connection", b"close"),
167         ]
168         msg = h11.Response(status_code=400, headers=headers, reason="Bad Request")
169         output = self.conn.send(msg)
170         msg = h11.Data(data=event.reason.encode("utf-8"))
171         output += self.conn.send(msg)
172         msg = h11.EndOfMessage()
173         output += self.conn.send(msg)
174         self.transport.write(output)
175         self.transport.close()
176 
177     def handle_text(self, event):
178         self.text += event.data
179         if event.message_finished:
180             self.queue.put_nowait({"type": "websocket.receive", "text": self.text})
181             self.text = ""
182             if not self.read_paused:
183                 self.read_paused = True
184                 self.transport.pause_reading()
185 
186     def handle_bytes(self, event):
187         self.bytes += event.data
188         # todo: we may want to guard the size of self.bytes and self.text
189         if event.message_finished:
190             self.queue.put_nowait({"type": "websocket.receive", "bytes": self.bytes})
191             self.bytes = b""
192             if not self.read_paused:
193                 self.read_paused = True
194                 self.transport.pause_reading()
195 
196     def handle_close(self, event):
197         if self.conn.state == ConnectionState.REMOTE_CLOSING:
198             self.transport.write(self.conn.send(event.response()))
199         self.queue.put_nowait({"type": "websocket.disconnect", "code": event.code})
200         self.transport.close()
201 
202     def handle_ping(self, event):
203         self.transport.write(self.conn.send(event.response()))
204 
205     def send_500_response(self):
206         headers = [
207             (b"content-type", b"text/plain; charset=utf-8"),
208             (b"connection", b"close"),
209         ]
210         if self.conn.connection is None:
211             output = self.conn.send(wsproto.events.RejectConnection(status_code=500))
212         else:
213             msg = h11.Response(
214                 status_code=500, headers=headers, reason="Internal Server Error"
215             )
216             output = self.conn.send(msg)
217             msg = h11.Data(data=b"Internal Server Error")
218             output += self.conn.send(msg)
219             msg = h11.EndOfMessage()
220             output += self.conn.send(msg)
221         self.transport.write(output)
222 
223     async def run_asgi(self):
224         try:
225             result = await self.app(self.scope, self.receive, self.send)
226         except BaseException as exc:
227             msg = "Exception in ASGI application\n"
228             self.logger.error(msg, exc_info=exc)
229             if not self.handshake_complete:
230                 self.send_500_response()
231             self.transport.close()
232         else:
233             if not self.handshake_complete:
234                 msg = "ASGI callable returned without completing handshake."
235                 self.logger.error(msg)
236                 self.send_500_response()
237                 self.transport.close()
238             elif result is not None:
239                 msg = "ASGI callable should return None, but returned '%s'."
240                 self.logger.error(msg, result)
241                 self.transport.close()
242 
243     async def send(self, message):
244         await self.writable.wait()
245 
246         message_type = message["type"]
247 
248         if not self.handshake_complete:
249             if message_type == "websocket.accept":
250                 self.logger.info(
251                     '%s - "WebSocket %s" [accepted]',
252                     self.scope["client"],
253                     get_path_with_query_string(self.scope),
254                 )
255                 self.handshake_complete = True
256                 subprotocol = message.get("subprotocol")
257                 extra_headers = self.default_headers + list(message.get("headers", []))
258                 extensions = []
259                 if self.config.ws_per_message_deflate:
260                     extensions.append(PerMessageDeflate())
261                 output = self.conn.send(
262                     wsproto.events.AcceptConnection(
263                         subprotocol=subprotocol,
264                         extensions=extensions,
265                         extra_headers=extra_headers,
266                     )
267                 )
268                 self.transport.write(output)
269 
270             elif message_type == "websocket.close":
271                 self.queue.put_nowait({"type": "websocket.disconnect", "code": None})
272                 self.logger.info(
273                     '%s - "WebSocket %s" 403',
274                     self.scope["client"],
275                     get_path_with_query_string(self.scope),
276                 )
277                 self.handshake_complete = True
278                 self.close_sent = True
279                 msg = events.RejectConnection(status_code=403, headers=[])
280                 output = self.conn.send(msg)
281                 self.transport.write(output)
282                 self.transport.close()
283 
284             else:
285                 msg = (
286                     "Expected ASGI message 'websocket.accept' or 'websocket.close', "
287                     "but got '%s'."
288                 )
289                 raise RuntimeError(msg % message_type)
290 
291         elif not self.close_sent:
292             if message_type == "websocket.send":
293                 bytes_data = message.get("bytes")
294                 text_data = message.get("text")
295                 data = text_data if bytes_data is None else bytes_data
296                 output = self.conn.send(wsproto.events.Message(data=data))
297                 if not self.transport.is_closing():
298                     self.transport.write(output)
299 
300             elif message_type == "websocket.close":
301                 self.close_sent = True
302                 code = message.get("code", 1000)
303                 reason = message.get("reason", "") or ""
304                 self.queue.put_nowait({"type": "websocket.disconnect", "code": code})
305                 output = self.conn.send(
306                     wsproto.events.CloseConnection(code=code, reason=reason)
307                 )
308                 if not self.transport.is_closing():
309                     self.transport.write(output)
310                     self.transport.close()
311 
312             else:
313                 msg = (
314                     "Expected ASGI message 'websocket.send' or 'websocket.close',"
315                     " but got '%s'."
316                 )
317                 raise RuntimeError(msg % message_type)
318 
319         else:
320             msg = "Unexpected ASGI message '%s', after sending 'websocket.close'."
321             raise RuntimeError(msg % message_type)
322 
323     async def receive(self):
324         message = await self.queue.get()
325         if self.read_paused and self.queue.empty():
326             self.read_paused = False
327             self.transport.resume_reading()
328         return message
329
```
--- 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/wsproto_impl.py b/uvicorn/protocols/websockets/wsproto_impl.py
--- a/uvicorn/protocols/websockets/wsproto_impl.py
+++ b/uvicorn/protocols/websockets/wsproto_impl.py
@@ -70,8 +70,7 @@
             self.logger.log(TRACE_LOG_LEVEL, "%sWebSocket connection made", prefix)
 
     def connection_lost(self, exc):
-        if exc is not None:
-            self.queue.put_nowait({"type": "websocket.disconnect"})
+        self.queue.put_nowait({"type": "websocket.disconnect"})
         self.connections.remove(self)
 
         if self.logger.level <= TRACE_LOG_LEVEL:

{"golden_diff": "diff --git a/uvicorn/protocols/websockets/wsproto_impl.py b/uvicorn/protocols/websockets/wsproto_impl.py\n--- a/uvicorn/protocols/websockets/wsproto_impl.py\n+++ b/uvicorn/protocols/websockets/wsproto_impl.py\n@@ -70,8 +70,7 @@\n             self.logger.log(TRACE_LOG_LEVEL, \"%sWebSocket connection made\", prefix)\n \n     def connection_lost(self, exc):\n-        if exc is not None:\n-            self.queue.put_nowait({\"type\": \"websocket.disconnect\"})\n+        self.queue.put_nowait({\"type\": \"websocket.disconnect\"})\n         self.connections.remove(self)\n \n         if self.logger.level <= TRACE_LOG_LEVEL:\n", "issue": "websocket.disconnect events are not sent to applications when clients don\u2019t close connections properly (wsproto only)\n### Checklist\r\n\r\n<!-- Please make sure you check all these items before submitting your bug report. -->\r\n\r\n- [x] The bug is reproducible against the latest release and/or `master`.\r\n- [x] There are no similar issues or pull requests to fix it yet.\r\n\r\n### Describe the bug\r\n\r\nWhen WebSocket connections are not properly closed by clients, uvicorn + wsproto leaves one zombie asyncio tasks running per connection and do not emit disconnect events to applications.\r\n\r\n### To reproduce\r\n1. Start any application with uvicorn using wsproto;\r\n2. Create a websocket connection;\r\n3. Do not close the websocket properly;\r\n4. Observe that no `websocket.disconnect` events have been sent to your application;\r\n5. Check for all asyncio tasks (`asyncio.all_tasks()`): there should be a zombie `run_asgi` task for each non properly closed connection.\r\n\r\nI\u2019ve setup this repo with instructions to make it easier to reproduce this issue: https://github.com/sephioh/uvicorn-wsproto-issue\r\n\r\n### Expected behavior\r\n\r\nA [disconnect](https://asgi.readthedocs.io/en/latest/specs/www.html#disconnect-receive-event-ws) event should be sent to applications. \r\n\r\n### Actual behavior\r\n\r\nDisconnect event are not being sent to apps when client socket loss happens. \r\n\r\n### Environment\r\n\r\nRunning uvicorn 0.13.4 with CPython 3.9.0 on Linux\r\nMust be executed with wsproto (\u2014ws wsproto).\r\n\r\n### Additional context\r\n\r\nhttps://docs.python.org/3/library/asyncio-protocol.html#asyncio.BaseProtocol.connection_lost\r\n\r\nNo problems are found when running with websockets impl (\u2014ws websockets).\r\n\r\nI\u2019ve opened a PR with a possible fix: https://github.com/encode/uvicorn/pull/996\n", "before_files": [{"content": "import asyncio\nimport logging\nfrom urllib.parse import unquote\n\nimport h11\nimport wsproto\nfrom wsproto import ConnectionType, events\nfrom wsproto.connection import ConnectionState\nfrom wsproto.extensions import PerMessageDeflate\nfrom wsproto.utilities import RemoteProtocolError\n\nfrom uvicorn.logging import TRACE_LOG_LEVEL\nfrom uvicorn.protocols.utils import (\n    get_local_addr,\n    get_path_with_query_string,\n    get_remote_addr,\n    is_ssl,\n)\n\n\nclass WSProtocol(asyncio.Protocol):\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        self.default_headers = server_state.default_headers\n\n        # Connection state\n        self.transport = None\n        self.server = None\n        self.client = None\n        self.scheme = None\n\n        # WebSocket state\n        self.connect_event = None\n        self.queue = asyncio.Queue()\n        self.handshake_complete = False\n        self.close_sent = False\n\n        self.conn = wsproto.WSConnection(connection_type=ConnectionType.SERVER)\n\n        self.read_paused = False\n        self.writable = asyncio.Event()\n        self.writable.set()\n\n        # Buffers\n        self.bytes = b\"\"\n        self.text = \"\"\n\n    # Protocol interface\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\n        if self.logger.level <= TRACE_LOG_LEVEL:\n            prefix = \"%s:%d - \" % tuple(self.client) if self.client else \"\"\n            self.logger.log(TRACE_LOG_LEVEL, \"%sWebSocket connection made\", prefix)\n\n    def connection_lost(self, exc):\n        if exc is not None:\n            self.queue.put_nowait({\"type\": \"websocket.disconnect\"})\n        self.connections.remove(self)\n\n        if self.logger.level <= TRACE_LOG_LEVEL:\n            prefix = \"%s:%d - \" % tuple(self.client) if self.client else \"\"\n            self.logger.log(TRACE_LOG_LEVEL, \"%sWebSocket connection lost\", prefix)\n\n        if exc is None:\n            self.transport.close()\n\n    def eof_received(self):\n        pass\n\n    def data_received(self, data):\n        try:\n            self.conn.receive_data(data)\n        except RemoteProtocolError as err:\n            if err.event_hint is not None:\n                self.transport.write(self.conn.send(err.event_hint))\n                self.transport.close()\n            else:\n                self.handle_no_connect(events.CloseConnection())\n        else:\n            self.handle_events()\n\n    def handle_events(self):\n        for event in self.conn.events():\n            if isinstance(event, events.Request):\n                self.handle_connect(event)\n            elif isinstance(event, events.TextMessage):\n                self.handle_text(event)\n            elif isinstance(event, events.BytesMessage):\n                self.handle_bytes(event)\n            elif isinstance(event, events.RejectConnection):\n                self.handle_no_connect(event)\n            elif isinstance(event, events.RejectData):\n                self.handle_no_connect(event)\n            elif isinstance(event, events.CloseConnection):\n                self.handle_close(event)\n            elif isinstance(event, events.Ping):\n                self.handle_ping(event)\n\n    def pause_writing(self):\n        \"\"\"\n        Called by the transport when the write buffer exceeds the high water mark.\n        \"\"\"\n        self.writable.clear()\n\n    def resume_writing(self):\n        \"\"\"\n        Called by the transport when the write buffer drops below the low water mark.\n        \"\"\"\n        self.writable.set()\n\n    def shutdown(self):\n        self.queue.put_nowait({\"type\": \"websocket.disconnect\", \"code\": 1012})\n        output = self.conn.send(wsproto.events.CloseConnection(code=1012))\n        self.transport.write(output)\n        self.transport.close()\n\n    def on_task_complete(self, task):\n        self.tasks.discard(task)\n\n    # Event handlers\n\n    def handle_connect(self, event):\n        self.connect_event = event\n        headers = [(b\"host\", event.host.encode())]\n        headers += [(key.lower(), value) for key, value in event.extra_headers]\n        raw_path, _, query_string = event.target.partition(\"?\")\n        self.scope = {\n            \"type\": \"websocket\",\n            \"asgi\": {\"version\": self.config.asgi_version, \"spec_version\": \"2.3\"},\n            \"http_version\": \"1.1\",\n            \"scheme\": self.scheme,\n            \"server\": self.server,\n            \"client\": self.client,\n            \"root_path\": self.root_path,\n            \"path\": unquote(raw_path),\n            \"raw_path\": raw_path.encode(\"ascii\"),\n            \"query_string\": query_string.encode(\"ascii\"),\n            \"headers\": headers,\n            \"subprotocols\": event.subprotocols,\n        }\n        self.queue.put_nowait({\"type\": \"websocket.connect\"})\n        task = self.loop.create_task(self.run_asgi())\n        task.add_done_callback(self.on_task_complete)\n        self.tasks.add(task)\n\n    def handle_no_connect(self, event):\n        headers = [\n            (b\"content-type\", b\"text/plain; charset=utf-8\"),\n            (b\"connection\", b\"close\"),\n        ]\n        msg = h11.Response(status_code=400, headers=headers, reason=\"Bad Request\")\n        output = self.conn.send(msg)\n        msg = h11.Data(data=event.reason.encode(\"utf-8\"))\n        output += self.conn.send(msg)\n        msg = h11.EndOfMessage()\n        output += self.conn.send(msg)\n        self.transport.write(output)\n        self.transport.close()\n\n    def handle_text(self, event):\n        self.text += event.data\n        if event.message_finished:\n            self.queue.put_nowait({\"type\": \"websocket.receive\", \"text\": self.text})\n            self.text = \"\"\n            if not self.read_paused:\n                self.read_paused = True\n                self.transport.pause_reading()\n\n    def handle_bytes(self, event):\n        self.bytes += event.data\n        # todo: we may want to guard the size of self.bytes and self.text\n        if event.message_finished:\n            self.queue.put_nowait({\"type\": \"websocket.receive\", \"bytes\": self.bytes})\n            self.bytes = b\"\"\n            if not self.read_paused:\n                self.read_paused = True\n                self.transport.pause_reading()\n\n    def handle_close(self, event):\n        if self.conn.state == ConnectionState.REMOTE_CLOSING:\n            self.transport.write(self.conn.send(event.response()))\n        self.queue.put_nowait({\"type\": \"websocket.disconnect\", \"code\": event.code})\n        self.transport.close()\n\n    def handle_ping(self, event):\n        self.transport.write(self.conn.send(event.response()))\n\n    def send_500_response(self):\n        headers = [\n            (b\"content-type\", b\"text/plain; charset=utf-8\"),\n            (b\"connection\", b\"close\"),\n        ]\n        if self.conn.connection is None:\n            output = self.conn.send(wsproto.events.RejectConnection(status_code=500))\n        else:\n            msg = h11.Response(\n                status_code=500, headers=headers, reason=\"Internal Server Error\"\n            )\n            output = self.conn.send(msg)\n            msg = h11.Data(data=b\"Internal Server Error\")\n            output += self.conn.send(msg)\n            msg = h11.EndOfMessage()\n            output += self.conn.send(msg)\n        self.transport.write(output)\n\n    async def run_asgi(self):\n        try:\n            result = await self.app(self.scope, self.receive, self.send)\n        except BaseException as exc:\n            msg = \"Exception in ASGI application\\n\"\n            self.logger.error(msg, exc_info=exc)\n            if not self.handshake_complete:\n                self.send_500_response()\n            self.transport.close()\n        else:\n            if not self.handshake_complete:\n                msg = \"ASGI callable returned without completing 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                self.transport.close()\n\n    async def send(self, message):\n        await self.writable.wait()\n\n        message_type = message[\"type\"]\n\n        if not self.handshake_complete:\n            if message_type == \"websocket.accept\":\n                self.logger.info(\n                    '%s - \"WebSocket %s\" [accepted]',\n                    self.scope[\"client\"],\n                    get_path_with_query_string(self.scope),\n                )\n                self.handshake_complete = True\n                subprotocol = message.get(\"subprotocol\")\n                extra_headers = self.default_headers + list(message.get(\"headers\", []))\n                extensions = []\n                if self.config.ws_per_message_deflate:\n                    extensions.append(PerMessageDeflate())\n                output = self.conn.send(\n                    wsproto.events.AcceptConnection(\n                        subprotocol=subprotocol,\n                        extensions=extensions,\n                        extra_headers=extra_headers,\n                    )\n                )\n                self.transport.write(output)\n\n            elif message_type == \"websocket.close\":\n                self.queue.put_nowait({\"type\": \"websocket.disconnect\", \"code\": None})\n                self.logger.info(\n                    '%s - \"WebSocket %s\" 403',\n                    self.scope[\"client\"],\n                    get_path_with_query_string(self.scope),\n                )\n                self.handshake_complete = True\n                self.close_sent = True\n                msg = events.RejectConnection(status_code=403, headers=[])\n                output = self.conn.send(msg)\n                self.transport.write(output)\n                self.transport.close()\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.close_sent:\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                output = self.conn.send(wsproto.events.Message(data=data))\n                if not self.transport.is_closing():\n                    self.transport.write(output)\n\n            elif message_type == \"websocket.close\":\n                self.close_sent = True\n                code = message.get(\"code\", 1000)\n                reason = message.get(\"reason\", \"\") or \"\"\n                self.queue.put_nowait({\"type\": \"websocket.disconnect\", \"code\": code})\n                output = self.conn.send(\n                    wsproto.events.CloseConnection(code=code, reason=reason)\n                )\n                if not self.transport.is_closing():\n                    self.transport.write(output)\n                    self.transport.close()\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 receive(self):\n        message = await self.queue.get()\n        if self.read_paused and self.queue.empty():\n            self.read_paused = False\n            self.transport.resume_reading()\n        return message\n", "path": "uvicorn/protocols/websockets/wsproto_impl.py"}], "after_files": [{"content": "import asyncio\nimport logging\nfrom urllib.parse import unquote\n\nimport h11\nimport wsproto\nfrom wsproto import ConnectionType, events\nfrom wsproto.connection import ConnectionState\nfrom wsproto.extensions import PerMessageDeflate\nfrom wsproto.utilities import RemoteProtocolError\n\nfrom uvicorn.logging import TRACE_LOG_LEVEL\nfrom uvicorn.protocols.utils import (\n    get_local_addr,\n    get_path_with_query_string,\n    get_remote_addr,\n    is_ssl,\n)\n\n\nclass WSProtocol(asyncio.Protocol):\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        self.default_headers = server_state.default_headers\n\n        # Connection state\n        self.transport = None\n        self.server = None\n        self.client = None\n        self.scheme = None\n\n        # WebSocket state\n        self.connect_event = None\n        self.queue = asyncio.Queue()\n        self.handshake_complete = False\n        self.close_sent = False\n\n        self.conn = wsproto.WSConnection(connection_type=ConnectionType.SERVER)\n\n        self.read_paused = False\n        self.writable = asyncio.Event()\n        self.writable.set()\n\n        # Buffers\n        self.bytes = b\"\"\n        self.text = \"\"\n\n    # Protocol interface\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\n        if self.logger.level <= TRACE_LOG_LEVEL:\n            prefix = \"%s:%d - \" % tuple(self.client) if self.client else \"\"\n            self.logger.log(TRACE_LOG_LEVEL, \"%sWebSocket connection made\", prefix)\n\n    def connection_lost(self, exc):\n        self.queue.put_nowait({\"type\": \"websocket.disconnect\"})\n        self.connections.remove(self)\n\n        if self.logger.level <= TRACE_LOG_LEVEL:\n            prefix = \"%s:%d - \" % tuple(self.client) if self.client else \"\"\n            self.logger.log(TRACE_LOG_LEVEL, \"%sWebSocket connection lost\", prefix)\n\n        if exc is None:\n            self.transport.close()\n\n    def eof_received(self):\n        pass\n\n    def data_received(self, data):\n        try:\n            self.conn.receive_data(data)\n        except RemoteProtocolError as err:\n            if err.event_hint is not None:\n                self.transport.write(self.conn.send(err.event_hint))\n                self.transport.close()\n            else:\n                self.handle_no_connect(events.CloseConnection())\n        else:\n            self.handle_events()\n\n    def handle_events(self):\n        for event in self.conn.events():\n            if isinstance(event, events.Request):\n                self.handle_connect(event)\n            elif isinstance(event, events.TextMessage):\n                self.handle_text(event)\n            elif isinstance(event, events.BytesMessage):\n                self.handle_bytes(event)\n            elif isinstance(event, events.RejectConnection):\n                self.handle_no_connect(event)\n            elif isinstance(event, events.RejectData):\n                self.handle_no_connect(event)\n            elif isinstance(event, events.CloseConnection):\n                self.handle_close(event)\n            elif isinstance(event, events.Ping):\n                self.handle_ping(event)\n\n    def pause_writing(self):\n        \"\"\"\n        Called by the transport when the write buffer exceeds the high water mark.\n        \"\"\"\n        self.writable.clear()\n\n    def resume_writing(self):\n        \"\"\"\n        Called by the transport when the write buffer drops below the low water mark.\n        \"\"\"\n        self.writable.set()\n\n    def shutdown(self):\n        self.queue.put_nowait({\"type\": \"websocket.disconnect\", \"code\": 1012})\n        output = self.conn.send(wsproto.events.CloseConnection(code=1012))\n        self.transport.write(output)\n        self.transport.close()\n\n    def on_task_complete(self, task):\n        self.tasks.discard(task)\n\n    # Event handlers\n\n    def handle_connect(self, event):\n        self.connect_event = event\n        headers = [(b\"host\", event.host.encode())]\n        headers += [(key.lower(), value) for key, value in event.extra_headers]\n        raw_path, _, query_string = event.target.partition(\"?\")\n        self.scope = {\n            \"type\": \"websocket\",\n            \"asgi\": {\"version\": self.config.asgi_version, \"spec_version\": \"2.3\"},\n            \"http_version\": \"1.1\",\n            \"scheme\": self.scheme,\n            \"server\": self.server,\n            \"client\": self.client,\n            \"root_path\": self.root_path,\n            \"path\": unquote(raw_path),\n            \"raw_path\": raw_path.encode(\"ascii\"),\n            \"query_string\": query_string.encode(\"ascii\"),\n            \"headers\": headers,\n            \"subprotocols\": event.subprotocols,\n        }\n        self.queue.put_nowait({\"type\": \"websocket.connect\"})\n        task = self.loop.create_task(self.run_asgi())\n        task.add_done_callback(self.on_task_complete)\n        self.tasks.add(task)\n\n    def handle_no_connect(self, event):\n        headers = [\n            (b\"content-type\", b\"text/plain; charset=utf-8\"),\n            (b\"connection\", b\"close\"),\n        ]\n        msg = h11.Response(status_code=400, headers=headers, reason=\"Bad Request\")\n        output = self.conn.send(msg)\n        msg = h11.Data(data=event.reason.encode(\"utf-8\"))\n        output += self.conn.send(msg)\n        msg = h11.EndOfMessage()\n        output += self.conn.send(msg)\n        self.transport.write(output)\n        self.transport.close()\n\n    def handle_text(self, event):\n        self.text += event.data\n        if event.message_finished:\n            self.queue.put_nowait({\"type\": \"websocket.receive\", \"text\": self.text})\n            self.text = \"\"\n            if not self.read_paused:\n                self.read_paused = True\n                self.transport.pause_reading()\n\n    def handle_bytes(self, event):\n        self.bytes += event.data\n        # todo: we may want to guard the size of self.bytes and self.text\n        if event.message_finished:\n            self.queue.put_nowait({\"type\": \"websocket.receive\", \"bytes\": self.bytes})\n            self.bytes = b\"\"\n            if not self.read_paused:\n                self.read_paused = True\n                self.transport.pause_reading()\n\n    def handle_close(self, event):\n        if self.conn.state == ConnectionState.REMOTE_CLOSING:\n            self.transport.write(self.conn.send(event.response()))\n        self.queue.put_nowait({\"type\": \"websocket.disconnect\", \"code\": event.code})\n        self.transport.close()\n\n    def handle_ping(self, event):\n        self.transport.write(self.conn.send(event.response()))\n\n    def send_500_response(self):\n        headers = [\n            (b\"content-type\", b\"text/plain; charset=utf-8\"),\n            (b\"connection\", b\"close\"),\n        ]\n        if self.conn.connection is None:\n            output = self.conn.send(wsproto.events.RejectConnection(status_code=500))\n        else:\n            msg = h11.Response(\n                status_code=500, headers=headers, reason=\"Internal Server Error\"\n            )\n            output = self.conn.send(msg)\n            msg = h11.Data(data=b\"Internal Server Error\")\n            output += self.conn.send(msg)\n            msg = h11.EndOfMessage()\n            output += self.conn.send(msg)\n        self.transport.write(output)\n\n    async def run_asgi(self):\n        try:\n            result = await self.app(self.scope, self.receive, self.send)\n        except BaseException as exc:\n            msg = \"Exception in ASGI application\\n\"\n            self.logger.error(msg, exc_info=exc)\n            if not self.handshake_complete:\n                self.send_500_response()\n            self.transport.close()\n        else:\n            if not self.handshake_complete:\n                msg = \"ASGI callable returned without completing 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                self.transport.close()\n\n    async def send(self, message):\n        await self.writable.wait()\n\n        message_type = message[\"type\"]\n\n        if not self.handshake_complete:\n            if message_type == \"websocket.accept\":\n                self.logger.info(\n                    '%s - \"WebSocket %s\" [accepted]',\n                    self.scope[\"client\"],\n                    get_path_with_query_string(self.scope),\n                )\n                self.handshake_complete = True\n                subprotocol = message.get(\"subprotocol\")\n                extra_headers = self.default_headers + list(message.get(\"headers\", []))\n                extensions = []\n                if self.config.ws_per_message_deflate:\n                    extensions.append(PerMessageDeflate())\n                output = self.conn.send(\n                    wsproto.events.AcceptConnection(\n                        subprotocol=subprotocol,\n                        extensions=extensions,\n                        extra_headers=extra_headers,\n                    )\n                )\n                self.transport.write(output)\n\n            elif message_type == \"websocket.close\":\n                self.queue.put_nowait({\"type\": \"websocket.disconnect\", \"code\": None})\n                self.logger.info(\n                    '%s - \"WebSocket %s\" 403',\n                    self.scope[\"client\"],\n                    get_path_with_query_string(self.scope),\n                )\n                self.handshake_complete = True\n                self.close_sent = True\n                msg = events.RejectConnection(status_code=403, headers=[])\n                output = self.conn.send(msg)\n                self.transport.write(output)\n                self.transport.close()\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.close_sent:\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                output = self.conn.send(wsproto.events.Message(data=data))\n                if not self.transport.is_closing():\n                    self.transport.write(output)\n\n            elif message_type == \"websocket.close\":\n                self.close_sent = True\n                code = message.get(\"code\", 1000)\n                reason = message.get(\"reason\", \"\") or \"\"\n                self.queue.put_nowait({\"type\": \"websocket.disconnect\", \"code\": code})\n                output = self.conn.send(\n                    wsproto.events.CloseConnection(code=code, reason=reason)\n                )\n                if not self.transport.is_closing():\n                    self.transport.write(output)\n                    self.transport.close()\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 receive(self):\n        message = await self.queue.get()\n        if self.read_paused and self.queue.empty():\n            self.read_paused = False\n            self.transport.resume_reading()\n        return message\n", "path": "uvicorn/protocols/websockets/wsproto_impl.py"}]}