more python style ood

Author: TingDaoK

awscrt/http_asyncio.py awscrt/aio/http_asyncio.pyawscrt/http_asyncio.py renamed to awscrt/aio/http_asyncio.py

@@ -9,11 +9,13 @@
 # SPDX-License-Identifier: Apache-2.0.
 
 import asyncio
+import io
+import _awscrt
 from concurrent.futures import Future
 import awscrt.exceptions
-from typing import List, Tuple, Optional, Union, Callable, Any
+from typing import List, Tuple, Optional, Union, Callable, Any, AsyncIterator
 from awscrt.http import (
-    HttpClientConnection, HttpRequest, HttpClientStream, HttpProxyOptions,
+    HttpClientConnectionBase, HttpRequest, HttClientStreamBase, HttpProxyOptions,
     Http2Setting, HttpVersion, Http2ClientStream
 )
 from awscrt.io import (
@@ -22,13 +24,12 @@
 from collections import deque
 
 
-class HttpClientConnectionAsync(HttpClientConnection):
+class HttpClientConnectionAsync(HttpClientConnectionBase):
     """
     An async HTTP client connection.
 
     Use `HttpClientConnectionAsync.new()` to establish a new connection.
     """
-    __slots__ = ('_host_name', '_port')
 
     @classmethod
     async def new(cls,
@@ -62,7 +63,7 @@ async def new(cls,
         Returns:
             HttpClientConnectionAsync: A new HTTP client connection.
         """
-        future = HttpClientConnection._generic_new(
+        future = cls._generic_new(
             host_name,
             port,
             bootstrap,
@@ -81,8 +82,8 @@ async def close(self) -> None:
         Returns:
             None: When shutdown is complete.
         """
-        close_future = super().close()
-        await asyncio.wrap_future(close_future)
+        _awscrt.http_connection_close(self._binding)
+        await asyncio.wrap_future(self.shutdown_future)
 
     def request(self,
                 request: 'HttpRequest',
@@ -100,7 +101,7 @@ def request(self,
         return HttpClientStreamAsync(self, request, loop)
 
 
-class Http2ClientConnectionAsync(HttpClientConnectionAsync):
+class Http2ClientConnectionAsync(HttpClientConnectionBase):
     """
     An async HTTP/2 client connection.
 
@@ -133,7 +134,7 @@ async def new(cls,
 
                     *   `settings` (List[Http2Setting]): List of settings that were changed.
         """
-        future = HttpClientConnection._generic_new(
+        future = cls._generic_new(
             host_name,
             port,
             bootstrap,
@@ -146,6 +147,18 @@ async def new(cls,
             asyncio_connection=True)
         return await asyncio.wrap_future(future)
 
+    async def close(self) -> None:
+        """Close the connection asynchronously.
+
+        Shutdown is asynchronous. This call has no effect if the connection is already
+        closing.
+
+        Returns:
+            None: When shutdown is complete.
+        """
+        _awscrt.http_connection_close(self._binding)
+        await asyncio.wrap_future(self.shutdown_future)
+
     def request(self,
                 request: 'HttpRequest',
                 manual_write: bool = False,
@@ -164,23 +177,7 @@ def request(self,
         return Http2ClientStreamAsync(self, request, manual_write, loop)
 
 
-class HttpClientStreamAsync(HttpClientStream):
-    """Async HTTP stream that sends a request and receives a response.
-
-    Create an HttpClientStreamAsync with `HttpClientConnectionAsync.request()`.
-
-    Attributes:
-        connection (HttpClientConnectionAsync): This stream's connection.
-
-        completion_future (asyncio.Future): Future that will contain
-            the response status code (int) when the request/response exchange
-            completes. If the exchange fails to complete, the Future will
-            contain an exception indicating why it failed.
-
-    Notes:
-        All async method on a stream (await stream.next(), etc.) must be performed in the
-        thread that owns the event loop used to create the stream
-    """
+class HttpClientStreamAsyncBase(HttClientStreamBase):
     __slots__ = (
         '_response_status_future',
         '_response_headers_future',
@@ -191,18 +188,6 @@ class HttpClientStreamAsync(HttpClientStream):
         '_status_code',
         '_loop')
 
-    def __init__(self, connection: HttpClientConnectionAsync, request: HttpRequest,
-                 loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
-        """Initialize an HTTP client stream.
-
-        Args:
-            connection (HttpClientConnectionAsync): The connection to send the request on.
-            request (HttpRequest): The HTTP request to send.
-            loop (Optional[asyncio.AbstractEventLoop]): Event loop to use for async operations.
-                If None, the current event loop is used.
-        """
-        self._init_common(connection, request, loop=loop)
-
     def _init_common(self, connection: HttpClientConnectionAsync,
                      request: HttpRequest,
                      http2_manual_write: bool = False,
@@ -231,7 +216,7 @@ def _init_common(self, connection: HttpClientConnectionAsync,
         self._status_code = None
 
         # Activate the stream immediately
-        self.activate()
+        _awscrt.http_client_stream_activate(self)
 
     def _on_response(self, status_code: int, name_value_pairs: List[Tuple[str, str]]) -> None:
         self._status_code = status_code
@@ -311,7 +296,38 @@ async def wait_for_completion(self) -> int:
         return await self._completion_future
 
 
-class Http2ClientStreamAsync(HttpClientStreamAsync, Http2ClientStream):
+class HttpClientStreamAsync(HttpClientStreamAsyncBase):
+    """Async HTTP stream that sends a request and receives a response.
+
+    Create an HttpClientStreamAsync with `HttpClientConnectionAsync.request()`.
+
+    Attributes:
+        connection (HttpClientConnectionAsync): This stream's connection.
+
+        completion_future (asyncio.Future): Future that will contain
+            the response status code (int) when the request/response exchange
+            completes. If the exchange fails to complete, the Future will
+            contain an exception indicating why it failed.
+
+    Notes:
+        All async method on a stream (await stream.next(), etc.) must be performed in the
+        thread that owns the event loop used to create the stream
+    """
+
+    def __init__(self, connection: HttpClientConnectionAsync, request: HttpRequest,
+                 loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
+        """Initialize an HTTP client stream.
+
+        Args:
+            connection (HttpClientConnectionAsync): The connection to send the request on.
+            request (HttpRequest): The HTTP request to send.
+            loop (Optional[asyncio.AbstractEventLoop]): Event loop to use for async operations.
+                If None, the current event loop is used.
+        """
+        super()._init_common(connection, request, loop=loop)
+
+
+class Http2ClientStreamAsync(HttpClientStreamAsyncBase):
     """HTTP/2 stream that sends a request and receives a response.
 
     Create an Http2ClientStreamAsync with `Http2ClientConnectionAsync.request()`.
@@ -321,17 +337,31 @@ def __init__(self, connection: HttpClientConnectionAsync, request: HttpRequest,
                  loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
         super()._init_common(connection, request, http2_manual_write=manual_write, loop=loop)
 
-    async def write_data_async(self,
-                               data_stream: Union[InputStream, Any],
-                               end_stream: bool = False) -> None:
+    async def write_data(self,
+                         data_stream: Union[InputStream, Any],
+                         end_stream: bool = False) -> None:
         """Write data to the stream asynchronously.
 
         Args:
-            data_stream (Union[InputStream, Any]): Data to write.
+            data_stream (AsyncIterator[bytes]): Async iterator that yields bytes to write.
+                Can be None to write an empty body, which is useful to finalize a request
+                with end_stream=True.
             end_stream (bool): Whether this is the last data to write.
 
         Returns:
             None: When the write completes.
         """
-        future = self.write_data(data_stream, end_stream)
+        future: Future = Future()
+        body_stream: InputStream = InputStream.wrap(data_stream, allow_none=True)
+
+        def on_write_complete(error_code: int) -> None:
+            if future.cancelled():
+                # the future was cancelled, so we don't need to set the result or exception
+                return
+            if error_code:
+                future.set_exception(awscrt.exceptions.from_code(error_code))
+            else:
+                future.set_result(None)
+
+        _awscrt.http2_client_stream_write_data(self, body_stream, end_stream, on_write_complete)
         await asyncio.wrap_future(future)