From: Florimond Manca Date: Sun, 15 Sep 2019 21:14:48 +0000 (+0200) Subject: Stream -> TCPStream (#339) X-Git-Tag: 0.7.3~7 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=315a18b4cf81f95683e6afdc10dba3c75eb512b7;p=thirdparty%2Fhttpx.git Stream -> TCPStream (#339) --- diff --git a/httpx/__init__.py b/httpx/__init__.py index a070d685..b464daa9 100644 --- a/httpx/__init__.py +++ b/httpx/__init__.py @@ -5,7 +5,7 @@ from .concurrency.asyncio import AsyncioBackend from .concurrency.base import ( BaseBackgroundManager, BasePoolSemaphore, - BaseStream, + BaseTCPStream, ConcurrencyBackend, ) from .config import ( @@ -112,7 +112,7 @@ __all__ = [ "TooManyRedirects", "WriteTimeout", "AsyncDispatcher", - "BaseStream", + "BaseTCPStream", "ConcurrencyBackend", "Dispatcher", "URL", diff --git a/httpx/concurrency/asyncio.py b/httpx/concurrency/asyncio.py index 9e50445a..1a145bed 100644 --- a/httpx/concurrency/asyncio.py +++ b/httpx/concurrency/asyncio.py @@ -1,13 +1,3 @@ -""" -The `Stream` class here provides a lightweight layer over -`asyncio.StreamReader` and `asyncio.StreamWriter`. - -Similarly `PoolSemaphore` is a lightweight layer over `BoundedSemaphore`. - -These classes help encapsulate the timeout logic, make it easier to unit-test -protocols, and help keep the rest of the package more `async`/`await` -based, and less strictly `asyncio`-specific. -""" import asyncio import functools import ssl @@ -21,7 +11,7 @@ from .base import ( BaseEvent, BasePoolSemaphore, BaseQueue, - BaseStream, + BaseTCPStream, ConcurrencyBackend, TimeoutFlag, ) @@ -50,7 +40,7 @@ def ssl_monkey_patch() -> None: MonkeyPatch.write = _fixed_write -class Stream(BaseStream): +class TCPStream(BaseTCPStream): def __init__( self, stream_reader: asyncio.StreamReader, @@ -176,13 +166,13 @@ class AsyncioBackend(ConcurrencyBackend): self._loop = asyncio.new_event_loop() return self._loop - async def connect( + async def open_tcp_stream( self, hostname: str, port: int, ssl_context: typing.Optional[ssl.SSLContext], timeout: TimeoutConfig, - ) -> BaseStream: + ) -> BaseTCPStream: try: stream_reader, stream_writer = await asyncio.wait_for( # type: ignore asyncio.open_connection(hostname, port, ssl=ssl_context), @@ -191,17 +181,17 @@ class AsyncioBackend(ConcurrencyBackend): except asyncio.TimeoutError: raise ConnectTimeout() - return Stream( + return TCPStream( stream_reader=stream_reader, stream_writer=stream_writer, timeout=timeout ) async def start_tls( self, - stream: BaseStream, + stream: BaseTCPStream, hostname: str, ssl_context: ssl.SSLContext, timeout: TimeoutConfig, - ) -> BaseStream: + ) -> BaseTCPStream: loop = self.loop if not hasattr(loop, "start_tls"): # pragma: no cover @@ -209,7 +199,7 @@ class AsyncioBackend(ConcurrencyBackend): "asyncio.AbstractEventLoop.start_tls() is only available in Python 3.7+" ) - assert isinstance(stream, Stream) + assert isinstance(stream, TCPStream) stream_reader = asyncio.StreamReader() protocol = asyncio.StreamReaderProtocol(stream_reader) diff --git a/httpx/concurrency/base.py b/httpx/concurrency/base.py index e62253dd..63fb1432 100644 --- a/httpx/concurrency/base.py +++ b/httpx/concurrency/base.py @@ -37,9 +37,9 @@ class TimeoutFlag: self.raise_on_write_timeout = True -class BaseStream: +class BaseTCPStream: """ - A stream with read/write operations. Abstracts away any asyncio-specific + A TCP stream 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. """ @@ -110,22 +110,22 @@ class BasePoolSemaphore: class ConcurrencyBackend: - async def connect( + async def open_tcp_stream( self, hostname: str, port: int, ssl_context: typing.Optional[ssl.SSLContext], timeout: TimeoutConfig, - ) -> BaseStream: + ) -> BaseTCPStream: raise NotImplementedError() # pragma: no cover async def start_tls( self, - stream: BaseStream, + stream: BaseTCPStream, hostname: str, ssl_context: ssl.SSLContext, timeout: TimeoutConfig, - ) -> BaseStream: + ) -> BaseTCPStream: raise NotImplementedError() # pragma: no cover def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore: diff --git a/httpx/dispatch/connection.py b/httpx/dispatch/connection.py index 529d373a..6e8cf0d0 100644 --- a/httpx/dispatch/connection.py +++ b/httpx/dispatch/connection.py @@ -85,7 +85,7 @@ class HTTPConnection(AsyncDispatcher): on_release = functools.partial(self.release_func, self) logger.debug(f"start_connect host={host!r} port={port!r} timeout={timeout!r}") - stream = await self.backend.connect(host, port, ssl_context, timeout) + stream = await self.backend.open_tcp_stream(host, port, ssl_context, timeout) http_version = stream.get_http_version() logger.debug(f"connected http_version={http_version!r}") diff --git a/httpx/dispatch/http11.py b/httpx/dispatch/http11.py index cfd227f0..64d49469 100644 --- a/httpx/dispatch/http11.py +++ b/httpx/dispatch/http11.py @@ -2,7 +2,7 @@ import typing import h11 -from ..concurrency.base import BaseStream, ConcurrencyBackend, TimeoutFlag +from ..concurrency.base import BaseTCPStream, ConcurrencyBackend, TimeoutFlag from ..config import TimeoutConfig, TimeoutTypes from ..models import AsyncRequest, AsyncResponse from ..utils import get_logger @@ -31,7 +31,7 @@ class HTTP11Connection: def __init__( self, - stream: BaseStream, + stream: BaseTCPStream, backend: ConcurrencyBackend, on_release: typing.Optional[OnReleaseCallback] = None, ): diff --git a/httpx/dispatch/http2.py b/httpx/dispatch/http2.py index cf518f1b..4ddd29f6 100644 --- a/httpx/dispatch/http2.py +++ b/httpx/dispatch/http2.py @@ -4,7 +4,7 @@ import typing import h2.connection import h2.events -from ..concurrency.base import BaseEvent, BaseStream, ConcurrencyBackend, TimeoutFlag +from ..concurrency.base import BaseEvent, BaseTCPStream, ConcurrencyBackend, TimeoutFlag from ..config import TimeoutConfig, TimeoutTypes from ..models import AsyncRequest, AsyncResponse from ..utils import get_logger @@ -17,7 +17,7 @@ class HTTP2Connection: def __init__( self, - stream: BaseStream, + stream: BaseTCPStream, backend: ConcurrencyBackend, on_release: typing.Callable = None, ): diff --git a/tests/dispatch/utils.py b/tests/dispatch/utils.py index c7cbedb5..aee4c18a 100644 --- a/tests/dispatch/utils.py +++ b/tests/dispatch/utils.py @@ -5,7 +5,7 @@ import h2.config import h2.connection import h2.events -from httpx import AsyncioBackend, BaseStream, Request, TimeoutConfig +from httpx import AsyncioBackend, BaseTCPStream, Request, TimeoutConfig from tests.concurrency import sleep @@ -15,13 +15,13 @@ class MockHTTP2Backend: self.backend = AsyncioBackend() if backend is None else backend self.server = None - async def connect( + async def open_tcp_stream( self, hostname: str, port: int, ssl_context: typing.Optional[ssl.SSLContext], timeout: TimeoutConfig, - ) -> BaseStream: + ) -> BaseTCPStream: self.server = MockHTTP2Server(self.app, backend=self.backend) return self.server @@ -30,7 +30,7 @@ class MockHTTP2Backend: return getattr(self.backend, name) -class MockHTTP2Server(BaseStream): +class MockHTTP2Server(BaseTCPStream): def __init__(self, app, backend): config = h2.config.H2Configuration(client_side=False) self.conn = h2.connection.H2Connection(config=config) @@ -42,7 +42,7 @@ class MockHTTP2Server(BaseStream): self.return_data = {} self.returning = {} - # Stream interface + # TCP stream interface def get_http_version(self) -> str: return "HTTP/2" diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index b9312eaa..ab93b302 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -19,7 +19,7 @@ async def test_start_tls_on_socket_stream(https_server): ctx = SSLConfig().load_ssl_context_no_verify(HTTPVersionConfig()) timeout = TimeoutConfig(5) - stream = await backend.connect( + stream = await backend.open_tcp_stream( https_server.url.host, https_server.url.port, None, timeout )