From 4ab181c5ab47e7a874fdd5fafa064018bd141f98 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jeremy=20Lain=C3=A9?= Date: Mon, 15 Mar 2021 13:34:06 +0100 Subject: [PATCH 1/2] Add a DatagramSocket abstraction --- httpcore/_backends/asyncio.py | 80 +++++++++++++++++++++++++++++++++-- httpcore/_backends/auto.py | 20 ++++++++- httpcore/_backends/base.py | 36 +++++++++++++++- httpcore/_backends/curio.py | 46 ++++++++++++++++++-- httpcore/_backends/trio.py | 45 ++++++++++++++++++-- httpcore/_types.py | 1 + httpcore/_utils.py | 27 +++++++++++- 7 files changed, 241 insertions(+), 14 deletions(-) diff --git a/httpcore/_backends/asyncio.py b/httpcore/_backends/asyncio.py index dd0d4217b..4ab469233 100644 --- a/httpcore/_backends/asyncio.py +++ b/httpcore/_backends/asyncio.py @@ -1,7 +1,8 @@ import asyncio import socket +from collections import deque from ssl import SSLContext -from typing import Optional +from typing import Deque, Optional, Tuple, cast from .. import _utils from .._exceptions import ( @@ -14,8 +15,14 @@ WriteTimeout, map_exceptions, ) -from .._types import TimeoutDict -from .base import AsyncBackend, AsyncLock, AsyncSemaphore, AsyncSocketStream +from .._types import SocketAddr, TimeoutDict +from .base import ( + AsyncBackend, + AsyncDatagramSocket, + AsyncLock, + AsyncSemaphore, + AsyncSocketStream, +) SSL_MONKEY_PATCH_APPLIED = False @@ -78,6 +85,50 @@ async def backport_start_tls( return ssl_protocol._app_transport +class DatagramProtocol(asyncio.DatagramProtocol): + def connection_lost(self, exc: Optional[Exception]) -> None: + self.close_event.set() + + def connection_made(self, transport: asyncio.BaseTransport) -> None: + self.close_event = asyncio.Event() + self.read_queue: Deque[Tuple[bytes, SocketAddr]] = deque() + self.read_event = asyncio.Event() + + def datagram_received(self, data: bytes, addr: SocketAddr) -> None: + self.read_queue.append((data, addr)) + self.read_event.set() + + +class DatagramSocket(AsyncDatagramSocket): + def __init__( + self, + transport: asyncio.DatagramTransport, + protocol: DatagramProtocol, + remote_addr: SocketAddr, + ): + self._protocol = protocol + self._transport = transport + self.remote_addr = remote_addr + + async def aclose(self) -> None: + self._transport.close() + await self._protocol.close_event.wait() + + async def receive(self) -> Tuple[bytes, SocketAddr]: + if not self._protocol.read_queue and not self._transport.is_closing(): + await self._protocol.read_event.wait() + + data, addr = self._protocol.read_queue.popleft() + + if not self._protocol.read_queue: + self._protocol.read_event.clear() + + return data, addr + + async def send(self, data: bytes, addr: SocketAddr) -> None: + self._transport.sendto(data, addr) + + class SocketStream(AsyncSocketStream): def __init__( self, stream_reader: asyncio.StreamReader, stream_writer: asyncio.StreamWriter @@ -246,6 +297,29 @@ def __init__(self) -> None: ssl_monkey_patch() SSL_MONKEY_PATCH_APPLIED = True + async def open_udp_socket( + self, + hostname: bytes, + port: int, + timeout: TimeoutDict, + *, + local_address: Optional[str], + ) -> AsyncDatagramSocket: + loop = asyncio.get_event_loop() + + infos = await loop.getaddrinfo(hostname, port, type=socket.SOCK_DGRAM) + _, local_addr, remote_addr = _utils.udp_socket_parameters(infos[0][4]) + + transport, protocol = await loop.create_datagram_endpoint( + lambda: DatagramProtocol(), + local_addr=local_addr, + ) + return DatagramSocket( + cast(asyncio.DatagramTransport, transport), + cast(DatagramProtocol, protocol), + remote_addr, + ) + async def open_tcp_stream( self, hostname: bytes, diff --git a/httpcore/_backends/auto.py b/httpcore/_backends/auto.py index 73d2b4a59..170440096 100644 --- a/httpcore/_backends/auto.py +++ b/httpcore/_backends/auto.py @@ -4,7 +4,13 @@ import sniffio from .._types import TimeoutDict -from .base import AsyncBackend, AsyncLock, AsyncSemaphore, AsyncSocketStream +from .base import ( + AsyncBackend, + AsyncDatagramSocket, + AsyncLock, + AsyncSemaphore, + AsyncSocketStream, +) # The following line is imported from the _sync modules from .sync import SyncBackend, SyncLock, SyncSemaphore, SyncSocketStream # noqa @@ -32,6 +38,18 @@ def backend(self) -> AsyncBackend: raise RuntimeError(f"Unsupported concurrency backend {backend!r}") return self._backend_implementation + async def open_udp_socket( + self, + hostname: bytes, + port: int, + timeout: TimeoutDict, + *, + local_address: Optional[str], + ) -> AsyncDatagramSocket: + return await self.backend.open_udp_socket( + hostname, port, timeout, local_address=local_address + ) + async def open_tcp_stream( self, hostname: bytes, diff --git a/httpcore/_backends/base.py b/httpcore/_backends/base.py index 1ca6e31b5..e2c66d37d 100644 --- a/httpcore/_backends/base.py +++ b/httpcore/_backends/base.py @@ -1,8 +1,8 @@ from ssl import SSLContext from types import TracebackType -from typing import TYPE_CHECKING, Optional, Type +from typing import TYPE_CHECKING, Optional, Tuple, Type -from .._types import TimeoutDict +from .._types import SocketAddr, TimeoutDict if TYPE_CHECKING: # pragma: no cover from .sync import SyncBackend @@ -39,6 +39,28 @@ def lookup_sync_backend(name: str) -> "SyncBackend": return SyncBackend() +class AsyncDatagramSocket: + """ + A datagram socket with read/write operations. Abstracts away any asyncio-specific + interfaces into a more generic base class, that we can use with alternate + backends, or for stand-alone test cases. + """ + + remote_addr: SocketAddr + + def get_http_version(self) -> str: + return "HTTP/3" + + async def aclose(self) -> None: + raise NotImplementedError() # pragma: no cover + + async def receive(self) -> Tuple[bytes, SocketAddr]: + raise NotImplementedError() # pragma: no cover + + async def send(self, data: bytes, addr: SocketAddr) -> None: + raise NotImplementedError() # pragma: no cover + + class AsyncSocketStream: """ A socket stream with read/write operations. Abstracts away any asyncio-specific @@ -104,6 +126,16 @@ async def release(self) -> None: class AsyncBackend: + async def open_udp_socket( + self, + hostname: bytes, + port: int, + timeout: TimeoutDict, + *, + local_address: Optional[str], + ) -> AsyncDatagramSocket: + raise NotImplementedError() # pragma: no cover + async def open_tcp_stream( self, hostname: bytes, diff --git a/httpcore/_backends/curio.py b/httpcore/_backends/curio.py index 2c1ae57ea..63d37299b 100644 --- a/httpcore/_backends/curio.py +++ b/httpcore/_backends/curio.py @@ -1,5 +1,5 @@ from ssl import SSLContext, SSLSocket -from typing import Optional +from typing import Optional, Tuple import curio import curio.io @@ -13,9 +13,15 @@ WriteTimeout, map_exceptions, ) -from .._types import TimeoutDict -from .._utils import get_logger, is_socket_readable -from .base import AsyncBackend, AsyncLock, AsyncSemaphore, AsyncSocketStream +from .._types import SocketAddr, TimeoutDict +from .._utils import get_logger, is_socket_readable, udp_socket_parameters +from .base import ( + AsyncBackend, + AsyncDatagramSocket, + AsyncLock, + AsyncSemaphore, + AsyncSocketStream, +) logger = get_logger(__name__) @@ -60,6 +66,21 @@ async def release(self) -> None: await self.semaphore.release() +class DatagramSocket(AsyncDatagramSocket): + def __init__(self, socket: curio.io.Socket, remote_addr: SocketAddr) -> None: + self.socket = socket + self.remote_addr = remote_addr + + async def aclose(self) -> None: + await self.socket.close() + + async def receive(self) -> Tuple[bytes, SocketAddr]: + return await self.socket.recvfrom(10000) + + async def send(self, data: bytes, addr: SocketAddr) -> None: + await self.socket.sendto(data, addr) + + class SocketStream(AsyncSocketStream): def __init__(self, socket: curio.io.Socket) -> None: self.read_lock = curio.Lock() @@ -136,6 +157,23 @@ def is_readable(self) -> bool: class CurioBackend(AsyncBackend): + async def open_udp_socket( + self, + hostname: bytes, + port: int, + timeout: TimeoutDict, + *, + local_address: Optional[str], + ) -> AsyncDatagramSocket: + infos = await curio.socket.getaddrinfo( + hostname, port, type=curio.socket.SOCK_DGRAM + ) + family, local_addr, remote_addr = udp_socket_parameters(infos[0][4]) + + socket = curio.socket.socket(family, curio.socket.SOCK_DGRAM) + socket.bind(local_addr) + return DatagramSocket(socket, remote_addr=remote_addr) + async def open_tcp_stream( self, hostname: bytes, diff --git a/httpcore/_backends/trio.py b/httpcore/_backends/trio.py index 244f49e25..c80693675 100644 --- a/httpcore/_backends/trio.py +++ b/httpcore/_backends/trio.py @@ -1,5 +1,5 @@ from ssl import SSLContext -from typing import Optional +from typing import Optional, Tuple import trio @@ -13,14 +13,36 @@ WriteTimeout, map_exceptions, ) -from .._types import TimeoutDict -from .base import AsyncBackend, AsyncLock, AsyncSemaphore, AsyncSocketStream +from .._types import SocketAddr, TimeoutDict +from .._utils import udp_socket_parameters +from .base import ( + AsyncBackend, + AsyncDatagramSocket, + AsyncLock, + AsyncSemaphore, + AsyncSocketStream, +) def none_as_inf(value: Optional[float]) -> float: return value if value is not None else float("inf") +class DatagramSocket(AsyncDatagramSocket): + def __init__(self, socket: trio.socket, remote_addr: SocketAddr) -> None: + self.socket = socket + self.remote_addr = remote_addr + + async def aclose(self) -> None: + self.socket.close() + + async def receive(self) -> Tuple[bytes, SocketAddr]: + return await self.socket.recvfrom(10000) + + async def send(self, data: bytes, addr: SocketAddr) -> None: + await self.socket.sendto(data, addr) + + class SocketStream(AsyncSocketStream): def __init__(self, stream: trio.abc.Stream) -> None: self.stream = stream @@ -130,6 +152,23 @@ async def release(self) -> None: class TrioBackend(AsyncBackend): + async def open_udp_socket( + self, + hostname: bytes, + port: int, + timeout: TimeoutDict, + *, + local_address: Optional[str], + ) -> AsyncDatagramSocket: + infos = await trio.socket.getaddrinfo( + hostname, port, type=trio.socket.SOCK_DGRAM + ) + family, local_addr, remote_addr = udp_socket_parameters(infos[0][4]) + + socket = trio.socket.socket(family, trio.socket.SOCK_DGRAM) + await socket.bind(local_addr) + return DatagramSocket(socket, remote_addr=remote_addr) + async def open_tcp_stream( self, hostname: bytes, diff --git a/httpcore/_types.py b/httpcore/_types.py index 2f9eeba7f..34e263292 100644 --- a/httpcore/_types.py +++ b/httpcore/_types.py @@ -9,4 +9,5 @@ Origin = Tuple[bytes, bytes, int] URL = Tuple[bytes, bytes, Optional[int], bytes] Headers = List[Tuple[bytes, bytes]] +SocketAddr = Tuple TimeoutDict = Mapping[str, Optional[float]] diff --git a/httpcore/_utils.py b/httpcore/_utils.py index 2520549e5..0de908b79 100644 --- a/httpcore/_utils.py +++ b/httpcore/_utils.py @@ -2,10 +2,11 @@ import logging import os import select +import socket import sys import typing -from ._types import URL, Origin +from ._types import URL, Origin, SocketAddr _LOGGER_INITIALIZED = False TRACE_LOG_LEVEL = 5 @@ -94,3 +95,27 @@ def is_socket_readable(sock_fd: int) -> bool: p = select.poll() p.register(sock_fd, select.POLLIN) return bool(p.poll(0)) + + +def udp_socket_parameters( + remote_addr: SocketAddr, +) -> typing.Tuple[int, SocketAddr, SocketAddr]: + """ + Determine the family, local and remote address to use for + the given remote address. + """ + family = socket.AF_INET6 + local_host = "::" + + if len(remote_addr) == 2: + # determine behaviour for IPv4 + if sys.platform == "win32": + # on Windows, we must use an IPv4 socket to reach an IPv4 host + family = socket.AF_INET + local_host = "0.0.0.0" + else: + # other platforms support dual-stack sockets + remote_addr = ("::ffff:" + remote_addr[0], remote_addr[1], 0, 0) + + local_addr = (local_host, 0) + return (family, local_addr, remote_addr) From 3e1a307386d123df75c049b63e5beed65aed64a2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jeremy=20Lain=C3=A9?= Date: Mon, 15 Mar 2021 15:43:09 +0100 Subject: [PATCH 2/2] WIP: add HTTP/3 backend --- httpcore/_async/connection.py | 53 ++++- httpcore/_async/connection_pool.py | 13 ++ httpcore/_async/http3.py | 354 +++++++++++++++++++++++++++++ setup.py | 1 + 4 files changed, 411 insertions(+), 10 deletions(-) create mode 100644 httpcore/_async/http3.py diff --git a/httpcore/_async/connection.py b/httpcore/_async/connection.py index 530663d83..f77a73f39 100644 --- a/httpcore/_async/connection.py +++ b/httpcore/_async/connection.py @@ -1,7 +1,13 @@ from ssl import SSLContext -from typing import Optional, Tuple, cast - -from .._backends.auto import AsyncBackend, AsyncLock, AsyncSocketStream, AutoBackend +from typing import Optional, Tuple, Union, cast + +from .._backends.auto import ( + AsyncBackend, + AsyncDatagramSocket, + AsyncLock, + AsyncSocketStream, + AutoBackend, +) from .._exceptions import ConnectError, ConnectTimeout from .._types import URL, Headers, Origin, TimeoutDict from .._utils import exponential_backoff, get_logger, url_to_origin @@ -24,15 +30,17 @@ def __init__( self, origin: Origin, http2: bool = False, + http3: bool = False, uds: str = None, ssl_context: SSLContext = None, - socket: AsyncSocketStream = None, + socket: Union[AsyncDatagramSocket, AsyncSocketStream] = None, local_address: str = None, retries: int = 0, backend: AsyncBackend = None, ): self.origin = origin self.http2 = http2 + self.http3 = http3 self.uds = uds self.ssl_context = SSLContext() if ssl_context is None else ssl_context self.socket = socket @@ -45,6 +53,7 @@ def __init__( self.connection: Optional[AsyncBaseHTTPConnection] = None self.is_http11 = False self.is_http2 = False + self.is_http3 = False self.connect_failed = False self.expires_at: Optional[float] = None self.backend = AutoBackend() if backend is None else backend @@ -55,6 +64,8 @@ def __repr__(self) -> str: http_version = "HTTP/1.1" elif self.is_http2: http_version = "HTTP/2" + elif self.is_http3: + http_version = "HTTP/3" return f"" def info(self) -> str: @@ -105,7 +116,9 @@ async def arequest( ) return await self.connection.arequest(method, url, headers, stream, ext) - async def _open_socket(self, timeout: TimeoutDict = None) -> AsyncSocketStream: + async def _open_socket( + self, timeout: TimeoutDict = None + ) -> Union[AsyncDatagramSocket, AsyncSocketStream]: scheme, hostname, port = self.origin timeout = {} if timeout is None else timeout ssl_context = self.ssl_context if scheme == b"https" else None @@ -115,7 +128,14 @@ async def _open_socket(self, timeout: TimeoutDict = None) -> AsyncSocketStream: while True: try: - if self.uds is None: + if self.http3: + return await self.backend.open_udp_socket( + hostname, + port, + timeout, + local_address=self.local_address, + ) + elif self.uds is None: return await self.backend.open_tcp_stream( hostname, port, @@ -138,22 +158,35 @@ async def _open_socket(self, timeout: TimeoutDict = None) -> AsyncSocketStream: self.connect_failed = True raise - def _create_connection(self, socket: AsyncSocketStream) -> None: + def _create_connection( + self, socket: Union[AsyncDatagramSocket, AsyncSocketStream] + ) -> None: http_version = socket.get_http_version() logger.trace( "create_connection socket=%r http_version=%r", socket, http_version ) - if http_version == "HTTP/2": + if http_version == "HTTP/3": + from .http3 import AsyncHTTP3Connection + + self.is_http3 = True + self.connection = AsyncHTTP3Connection( + socket=cast(AsyncDatagramSocket, socket), + backend=self.backend, + ssl_context=self.ssl_context, + ) + elif http_version == "HTTP/2": from .http2 import AsyncHTTP2Connection self.is_http2 = True self.connection = AsyncHTTP2Connection( - socket=socket, backend=self.backend, ssl_context=self.ssl_context + socket=cast(AsyncSocketStream, socket), + backend=self.backend, + ssl_context=self.ssl_context, ) else: self.is_http11 = True self.connection = AsyncHTTP11Connection( - socket=socket, ssl_context=self.ssl_context + socket=cast(AsyncSocketStream, socket), ssl_context=self.ssl_context ) @property diff --git a/httpcore/_async/connection_pool.py b/httpcore/_async/connection_pool.py index 46ede0ba4..8a85a2629 100644 --- a/httpcore/_async/connection_pool.py +++ b/httpcore/_async/connection_pool.py @@ -88,6 +88,7 @@ class AsyncConnectionPool(AsyncHTTPTransport): * **keepalive_expiry** - `Optional[float]` - The maximum time to allow before closing a keep-alive connection. * **http2** - `bool` - Enable HTTP/2 support. + * **http3** - `bool` - Enable HTTP/3 support. * **uds** - `str` - Path to a Unix Domain Socket to use instead of TCP sockets. * **local_address** - `Optional[str]` - Local address to connect from. Can also be used to connect using a particular address family. Using @@ -106,6 +107,7 @@ def __init__( max_keepalive_connections: int = None, keepalive_expiry: float = None, http2: bool = False, + http3: bool = False, uds: str = None, local_address: str = None, retries: int = 0, @@ -127,6 +129,7 @@ def __init__( self._max_keepalive_connections = max_keepalive_connections self._keepalive_expiry = keepalive_expiry self._http2 = http2 + self._http3 = http3 self._uds = uds self._local_address = local_address self._retries = retries @@ -144,6 +147,15 @@ def __init__( "package is not installed. Use 'pip install httpcore[http2]'." ) + if http3: + try: + import aioquic # noqa: F401 + except ImportError: + raise ImportError( + "Attempted to use http3=True, but the 'aioquic' " + "package is not installed. Use 'pip install httpcore[http3]'." + ) + @property def _connection_semaphore(self) -> AsyncSemaphore: # We do this lazily, to make sure backend autodetection always @@ -171,6 +183,7 @@ def _create_connection( return AsyncHTTPConnection( origin=origin, http2=self._http2, + http3=self._http3, uds=self._uds, ssl_context=self._ssl_context, local_address=self._local_address, diff --git a/httpcore/_async/http3.py b/httpcore/_async/http3.py new file mode 100644 index 000000000..484759500 --- /dev/null +++ b/httpcore/_async/http3.py @@ -0,0 +1,354 @@ +from ssl import SSLContext +from typing import AsyncIterator, Dict, List, Tuple, cast + +from aioquic.h3.connection import H3_ALPN, H3Connection +from aioquic.h3.events import DataReceived, H3Event, HeadersReceived +from aioquic.quic.configuration import QuicConfiguration +from aioquic.quic.connection import QuicConnection + +from .._backends.auto import ( + AsyncBackend, + AsyncDatagramSocket, + AsyncLock, + AsyncSocketStream, +) +from .._bytestreams import AsyncIteratorByteStream, PlainByteStream +from .._exceptions import RemoteProtocolError +from .._types import URL, Headers, TimeoutDict +from .._utils import get_logger +from .base import AsyncByteStream, ConnectionState +from .http import AsyncBaseHTTPConnection + +logger = get_logger(__name__) + + +class AsyncHTTP3Connection(AsyncBaseHTTPConnection): + def __init__( + self, + socket: AsyncDatagramSocket, + backend: AsyncBackend, + ssl_context: SSLContext = None, + ): + self.socket = socket + self.ssl_context = SSLContext() if ssl_context is None else ssl_context + + self.backend = backend + self._quic = QuicConnection( + configuration=QuicConfiguration(is_client=True, alpn_protocols=H3_ALPN) + ) + self._http = H3Connection(self._quic) + + self.sent_connection_init = False + self.streams = {} # type: Dict[int, AsyncHTTP3Stream] + self.events = {} # type: Dict[int, List[H3Event]] + + self.state = ConnectionState.ACTIVE + + def __repr__(self) -> str: + return f"" + + def info(self) -> str: + return f"HTTP/3, {self.state.name}, {len(self.streams)} streams" + + @property + def init_lock(self) -> AsyncLock: + # We do this lazily, to make sure backend autodetection always + # runs within an async context. + if not hasattr(self, "_initialization_lock"): + self._initialization_lock = self.backend.create_lock() + return self._initialization_lock + + @property + def read_lock(self) -> AsyncLock: + # We do this lazily, to make sure backend autodetection always + # runs within an async context. + if not hasattr(self, "_read_lock"): + self._read_lock = self.backend.create_lock() + return self._read_lock + + async def start_tls( + self, hostname: bytes, timeout: TimeoutDict = None + ) -> AsyncSocketStream: + raise NotImplementedError("TLS upgrade not supported on HTTP/3 connections.") + + def get_state(self) -> ConnectionState: + return self.state + + def mark_as_ready(self) -> None: + if self.state == ConnectionState.IDLE: + self.state = ConnectionState.READY + + async def arequest( + self, + method: bytes, + url: URL, + headers: Headers = None, + stream: AsyncByteStream = None, + ext: dict = None, + ) -> Tuple[int, Headers, AsyncByteStream, dict]: + ext = {} if ext is None else ext + timeout = cast(TimeoutDict, ext.get("timeout", {})) + + async with self.init_lock: + if not self.sent_connection_init: + # The very first stream is responsible for initiating the connection. + self.state = ConnectionState.ACTIVE + await self.send_connection_init(timeout) + self.sent_connection_init = True + + # await self.max_streams_semaphore.acquire() + try: + stream_id = self._quic.get_next_available_stream_id() + + h3_stream = AsyncHTTP3Stream(stream_id=stream_id, connection=self) + self.streams[stream_id] = h3_stream + self.events[stream_id] = [] + return await h3_stream.arequest(method, url, headers, stream, ext) + except Exception: # noqa: PIE786 + # await self.max_streams_semaphore.release() + raise + + async def send_connection_init(self, timeout: TimeoutDict) -> None: + """ + The HTTP/3 connection requires some initial setup before we can start + using individual request/response streams on it. + """ + + logger.trace("initiate_connection=%r", self) + + now = await self.backend.time() + self._quic.connect(self.socket.remote_addr, now=now) + await self._transmit() + + @property + def is_closed(self) -> bool: + return False + + def is_socket_readable(self) -> bool: + return True + + async def aclose(self) -> None: + logger.trace("close_connection=%r", self) + if self.state != ConnectionState.CLOSED: + self.state = ConnectionState.CLOSED + + await self.socket.aclose() + + async def wait_for_event(self, stream_id: int, timeout: TimeoutDict) -> H3Event: + """ + Returns the next event for a given stream. + If no events are available yet, then waits on the network until + an event is available. + """ + async with self.read_lock: + while not self.events[stream_id]: + await self.receive_events(timeout) + return self.events[stream_id].pop(0) + + async def receive_events(self, timeout: TimeoutDict) -> None: + """ + Read some data from the network, and update the H2 state. + """ + data, addr = await self.socket.receive() + if data == b"": + raise RemoteProtocolError("Server disconnected") + + now = await self.backend.time() + self._quic.receive_datagram(data, addr, now=now) + + quic_event = self._quic.next_event() + while quic_event is not None: + for http_event in self._http.handle_event(quic_event): + event_stream_id = getattr(http_event, "stream_id", 0) + logger.trace( + "receive_event stream_id=%r event=%s", event_stream_id, http_event + ) + + if event_stream_id in self.events: + self.events[event_stream_id].append(http_event) + quic_event = self._quic.next_event() + await self._transmit() + + async def send_headers( + self, stream_id: int, headers: Headers, end_stream: bool, timeout: TimeoutDict + ) -> None: + logger.trace("send_headers stream_id=%r headers=%r", stream_id, headers) + self._http.send_headers( + stream_id=stream_id, headers=headers, end_stream=end_stream + ) + await self._transmit() + + async def send_data( + self, stream_id: int, chunk: bytes, timeout: TimeoutDict + ) -> None: + logger.trace("send_data stream_id=%r chunk=%r", stream_id, chunk) + self._http.send_data(stream_id, chunk, end_stream=False) + await self._transmit() + + async def end_stream(self, stream_id: int, timeout: TimeoutDict) -> None: + logger.trace("end_stream stream_id=%r", stream_id) + self._http.send_data(stream_id, b"", end_stream=True) + await self._transmit() + + async def close_stream(self, stream_id: int) -> None: + try: + logger.trace("close_stream stream_id=%r", stream_id) + del self.streams[stream_id] + del self.events[stream_id] + + if not self.streams: + if self.state == ConnectionState.ACTIVE: + self.state = ConnectionState.IDLE + elif self.state == ConnectionState.FULL: + await self.aclose() + finally: + # await self.max_streams_semaphore.release() + pass + + async def _transmit(self) -> None: + """ + Send pending datagrams to the peer and arm the timer if needed. + """ + self._transmit_task = None + + # send datagrams + now = await self.backend.time() + for data, addr in self._quic.datagrams_to_send(now=now): + await self.socket.send(data, addr) + + """ + # re-arm timer + timer_at = self._quic.get_timer() + if self._timer is not None and self._timer_at != timer_at: + self._timer.cancel() + self._timer = None + if self._timer is None and timer_at is not None: + self._timer = self._loop.call_at(timer_at, self._handle_timer) + self._timer_at = timer_at + """ + + +class AsyncHTTP3Stream: + def __init__(self, stream_id: int, connection: AsyncHTTP3Connection) -> None: + self.stream_ended = False + self.stream_id = stream_id + self.connection = connection + + async def arequest( + self, + method: bytes, + url: URL, + headers: Headers = None, + stream: AsyncByteStream = None, + ext: dict = None, + ) -> Tuple[int, Headers, AsyncByteStream, dict]: + headers = [] if headers is None else [(k.lower(), v) for (k, v) in headers] + stream = PlainByteStream(b"") if stream is None else stream + ext = {} if ext is None else ext + timeout = cast(TimeoutDict, ext.get("timeout", {})) + + # Send the request. + seen_headers = set(key for key, value in headers) + has_body = ( + b"content-length" in seen_headers or b"transfer-encoding" in seen_headers + ) + + await self.send_headers(method, url, headers, has_body, timeout) + if has_body: + await self.send_body(stream, timeout) + + # Receive the response. + status_code, headers = await self.receive_response(timeout) + response_stream = AsyncIteratorByteStream( + aiterator=self.body_iter(timeout), aclose_func=self._response_closed + ) + + ext = { + "http_version": "HTTP/3", + } + return (status_code, headers, response_stream, ext) + + async def send_headers( + self, + method: bytes, + url: URL, + headers: Headers, + has_body: bool, + timeout: TimeoutDict, + ) -> None: + scheme, hostname, port, path = url + + # In HTTP/3 the ':authority' pseudo-header is used instead of 'Host'. + # In order to gracefully handle HTTP/1.1 and HTTP/3 we always require + # HTTP/1.1 style headers, and map them appropriately if we end up on + # an HTTP/3 connection. + authority = None + for k, v in headers: + if k == b"host": + authority = v + break + + if authority is None: + default_port = {b"http": 80, b"https": 443}.get(scheme) + if port is not None and port != default_port: + authority = b"%s:%d" % (authority, port) + else: + authority = hostname + + headers = [ + (b":method", method), + (b":authority", authority), + (b":scheme", scheme), + (b":path", path), + ] + [ + (k, v) + for k, v in headers + if k + not in ( + b"host", + b"transfer-encoding", + ) + ] + end_stream = not has_body + + await self.connection.send_headers(self.stream_id, headers, end_stream, timeout) + + async def send_body(self, stream: AsyncByteStream, timeout: TimeoutDict) -> None: + async for data in stream: + await self.connection.send_data(self.stream_id, data, timeout) + + await self.connection.end_stream(self.stream_id, timeout) + + async def receive_response( + self, timeout: TimeoutDict + ) -> Tuple[int, List[Tuple[bytes, bytes]]]: + """ + Read the response status and headers from the network. + """ + while True: + event = await self.connection.wait_for_event(self.stream_id, timeout) + if isinstance(event, HeadersReceived): + self.stream_ended = event.stream_ended + break + + status_code = 200 + headers = [] + for k, v in event.headers: + if k == b":status": + status_code = int(v.decode("ascii", errors="ignore")) + elif not k.startswith(b":"): + headers.append((k, v)) + + return (status_code, headers) + + async def body_iter(self, timeout: TimeoutDict) -> AsyncIterator[bytes]: + while not self.stream_ended: + event = await self.connection.wait_for_event(self.stream_id, timeout) + if isinstance(event, DataReceived): + self.stream_ended = event.stream_ended + yield event.data + elif isinstance(event, HeadersReceived): + self.stream_ended = event.stream_ended + + async def _response_closed(self) -> None: + await self.connection.close_stream(self.stream_id) diff --git a/setup.py b/setup.py index 90dbd2d80..5cc68fcda 100644 --- a/setup.py +++ b/setup.py @@ -56,6 +56,7 @@ def get_packages(package): install_requires=["h11==0.*", "sniffio==1.*"], extras_require={ "http2": ["h2>=3,<5"], + "http3": ["aioquic>=0.9.11,<1"], }, classifiers=[ "Development Status :: 3 - Alpha",