Asyncio Example ServerΒΆ
This example is a basic HTTP/2 server written using asyncio, using some functionality that was introduced in Python 3.5. This server represents basically just the same JSON-headers-returning server that was built in the Getting Started: Writing Your Own HTTP/2 Server document.
This example demonstrates some basic asyncio techniques.
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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 | # -*- coding: utf-8 -*- """ asyncio-server.py ~~~~~~~~~~~~~~~~~ A fully-functional HTTP/2 server using asyncio. Requires Python 3.5+. This example demonstrates handling requests with bodies, as well as handling those without. In particular, it demonstrates the fact that DataReceived may be called multiple times, and that applications must handle that possibility. """ import asyncio import io import json import ssl import collections from typing import List, Tuple from h2.config import H2Configuration from h2.connection import H2Connection from h2.events import ( ConnectionTerminated, DataReceived, RemoteSettingsChanged, RequestReceived, StreamEnded, StreamReset, WindowUpdated ) from h2.errors import ErrorCodes from h2.exceptions import ProtocolError, StreamClosedError from h2.settings import SettingCodes RequestData = collections.namedtuple('RequestData', ['headers', 'data']) class H2Protocol(asyncio.Protocol): def __init__(self): config = H2Configuration(client_side=False, header_encoding='utf-8') self.conn = H2Connection(config=config) self.transport = None self.stream_data = {} self.flow_control_futures = {} def connection_made(self, transport: asyncio.Transport): self.transport = transport self.conn.initiate_connection() self.transport.write(self.conn.data_to_send()) def connection_lost(self, exc): for future in self.flow_control_futures.values(): future.cancel() self.flow_control_futures = {} def data_received(self, data: bytes): try: events = self.conn.receive_data(data) except ProtocolError as e: self.transport.write(self.conn.data_to_send()) self.transport.close() else: self.transport.write(self.conn.data_to_send()) for event in events: if isinstance(event, RequestReceived): self.request_received(event.headers, event.stream_id) elif isinstance(event, DataReceived): self.receive_data(event.data, event.stream_id) elif isinstance(event, StreamEnded): self.stream_complete(event.stream_id) elif isinstance(event, ConnectionTerminated): self.transport.close() elif isinstance(event, StreamReset): self.stream_reset(event.stream_id) elif isinstance(event, WindowUpdated): self.window_updated(event.stream_id, event.delta) elif isinstance(event, RemoteSettingsChanged): if SettingCodes.INITIAL_WINDOW_SIZE in event.changed_settings: self.window_updated(None, 0) self.transport.write(self.conn.data_to_send()) def request_received(self, headers: List[Tuple[str, str]], stream_id: int): headers = collections.OrderedDict(headers) method = headers[':method'] # Store off the request data. request_data = RequestData(headers, io.BytesIO()) self.stream_data[stream_id] = request_data def stream_complete(self, stream_id: int): """ When a stream is complete, we can send our response. """ try: request_data = self.stream_data[stream_id] except KeyError: # Just return, we probably 405'd this already return headers = request_data.headers body = request_data.data.getvalue().decode('utf-8') data = json.dumps( {"headers": headers, "body": body}, indent=4 ).encode("utf8") response_headers = ( (':status', '200'), ('content-type', 'application/json'), ('content-length', str(len(data))), ('server', 'asyncio-h2'), ) self.conn.send_headers(stream_id, response_headers) asyncio.ensure_future(self.send_data(data, stream_id)) def receive_data(self, data: bytes, stream_id: int): """ We've received some data on a stream. If that stream is one we're expecting data on, save it off. Otherwise, reset the stream. """ try: stream_data = self.stream_data[stream_id] except KeyError: self.conn.reset_stream( stream_id, error_code=ErrorCodes.PROTOCOL_ERROR ) else: stream_data.data.write(data) def stream_reset(self, stream_id): """ A stream reset was sent. Stop sending data. """ if stream_id in self.flow_control_futures: future = self.flow_control_futures.pop(stream_id) future.cancel() async def send_data(self, data, stream_id): """ Send data according to the flow control rules. """ while data: while self.conn.local_flow_control_window(stream_id) < 1: try: await self.wait_for_flow_control(stream_id) except asyncio.CancelledError: return chunk_size = min( self.conn.local_flow_control_window(stream_id), len(data), self.conn.max_outbound_frame_size, ) try: self.conn.send_data( stream_id, data[:chunk_size], end_stream=(chunk_size == len(data)) ) except (StreamClosedError, ProtocolError): # The stream got closed and we didn't get told. We're done # here. break self.transport.write(self.conn.data_to_send()) data = data[chunk_size:] async def wait_for_flow_control(self, stream_id): """ Waits for a Future that fires when the flow control window is opened. """ f = asyncio.Future() self.flow_control_futures[stream_id] = f await f def window_updated(self, stream_id, delta): """ A window update frame was received. Unblock some number of flow control Futures. """ if stream_id and stream_id in self.flow_control_futures: f = self.flow_control_futures.pop(stream_id) f.set_result(delta) elif not stream_id: for f in self.flow_control_futures.values(): f.set_result(delta) self.flow_control_futures = {} ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) ssl_context.options |= ( ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1 | ssl.OP_NO_COMPRESSION ) ssl_context.load_cert_chain(certfile="cert.crt", keyfile="cert.key") ssl_context.set_alpn_protocols(["h2"]) loop = asyncio.get_event_loop() # Each client connection will create a new protocol instance coro = loop.create_server(H2Protocol, '127.0.0.1', 8443, ssl=ssl_context) server = loop.run_until_complete(coro) # Serve requests until Ctrl+C is pressed print('Serving on {}'.format(server.sockets[0].getsockname())) try: loop.run_forever() except KeyboardInterrupt: pass # Close the server server.close() loop.run_until_complete(server.wait_closed()) loop.close() |