1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
|
import asyncio
from uvicorn._types import (
ASGIReceiveCallable,
ASGISendCallable,
HTTPResponseBodyEvent,
HTTPResponseStartEvent,
Scope,
)
CLOSE_HEADER = (b"connection", b"close")
HIGH_WATER_LIMIT = 65536
class FlowControl:
def __init__(self, transport: asyncio.Transport) -> None:
self._transport = transport
self.read_paused = False
self.write_paused = False
self._is_writable_event = asyncio.Event()
self._is_writable_event.set()
async def drain(self) -> None:
await self._is_writable_event.wait()
def pause_reading(self) -> None:
if not self.read_paused:
self.read_paused = True
self._transport.pause_reading()
def resume_reading(self) -> None:
if self.read_paused:
self.read_paused = False
self._transport.resume_reading()
def pause_writing(self) -> None:
if not self.write_paused:
self.write_paused = True
self._is_writable_event.clear()
def resume_writing(self) -> None:
if self.write_paused:
self.write_paused = False
self._is_writable_event.set()
async def service_unavailable(
scope: "Scope", receive: "ASGIReceiveCallable", send: "ASGISendCallable"
) -> None:
response_start: "HTTPResponseStartEvent" = {
"type": "http.response.start",
"status": 503,
"headers": [
(b"content-type", b"text/plain; charset=utf-8"),
(b"connection", b"close"),
],
}
await send(response_start)
response_body: "HTTPResponseBodyEvent" = {
"type": "http.response.body",
"body": b"Service Unavailable",
"more_body": False,
}
await send(response_body)
|