Package restructure

src/hip/__init__.py

@@ -1,4 +1,4 @@
-from . import backends, parsers, transports
+from . import backends, parsers
 from .client import Client
 from .core.content import HTML, JSON, Binary, Content, Text
 from .core.forms import File, Files, Form, MultiPart
@@ -35,5 +35,4 @@
     "parsers",
     "sleep",
     "timeout",
-    "transports",
 ]

src/hip/client/__init__.py

@@ -0,0 +1,14 @@
+from .base import Client
+from .connection import Connection, GZipStream, open_connection
+from .pool import ConnectionPool
+from .transport import SendContext, Transport
+
+__all__ = [
+    "Client",
+    "Connection",
+    "ConnectionPool",
+    "GZipStream",
+    "SendContext",
+    "Transport",
+    "open_connection",
+]

src/hip/client.py → src/hip/client/base.py

@@ -1,12 +1,13 @@
 import types
 import typing
 
-from .core.content import HTML, JSON, Binary, Content, Text, select_media
-from .core.headers import Headers
-from .core.request import Method, Request
-from .core.response import Response
-from .core.urls import URL
-from .transports import ConnectionPool, SendContext, Transport
+from ..core.content import HTML, JSON, Binary, Content, Text, select_media
+from ..core.headers import Headers
+from ..core.request import Method, Request
+from ..core.response import Response
+from ..core.urls import URL
+from .pool import ConnectionPool
+from .transport import SendContext, Transport
 
 __all__ = ["Client"]
 

src/hip/transports/connection.py → src/hip/client/connection.py

@@ -1,22 +1,24 @@
+import io
 import types
 import typing
 
 if typing.TYPE_CHECKING:
     # Once we drop 3.10 support, we can use 'typing.Self'.
     from typing_extensions import Self
 
+import zlib
+
 from ..backends.sync import Lock, NetworkBackend
-from ..body import GZipStream, HTTPStream
 from ..core.content import Content
 from ..core.headers import Headers
 from ..core.request import Method, Request
 from ..core.response import Response, StatusCode
 from ..core.streams import Stream
 from ..core.urls import URL
 from ..parsers import HTTPParser
-from .base import SendContext, Transport
+from .transport import SendContext, Transport
 
-__all__ = ["Connection", "open_connection"]
+__all__ = ["Connection", "GZipStream", "open_connection"]
 
 
 class Connection(Transport):
@@ -94,7 +96,7 @@ def _recv_head(self) -> tuple[StatusCode, Headers]:
         return status_code, headers
 
     def _recv_body(self, headers) -> Stream:
-        stream: Stream = HTTPStream(self._parser, is_response=True)
+        stream: Stream = ResponseStream(self._parser)
         content_encoding = headers.get("Content-Encoding", "")
         if content_encoding.lower() == "gzip":
             stream = GZipStream(stream)
@@ -143,3 +145,86 @@ def open_connection(
         stream = backend.connect(host, port)
 
     return Connection(stream, url)
+
+
+class GZipStream(Stream):
+    def __init__(self, stream: Stream):
+        self._stream = stream
+        # Adding 16 enables gzip header and trailer handling.
+        self._decoder = zlib.decompressobj(wbits=16 + zlib.MAX_WBITS)
+        self._finished = False
+        self._buffer = bytearray()
+
+    def read(self, size=-1) -> bytes:
+        while (size < 0 or len(self._buffer) < size) and not self._finished:
+            compressed = self._stream.read(8192)
+
+            try:
+                if compressed:
+                    decoded = self._decoder.decompress(compressed)
+                else:
+                    decoded = self._decoder.flush()
+            except zlib.error as exc:
+                raise ValueError("Invalid gzip response content.") from exc
+
+            if not compressed:
+                if not self._decoder.eof:
+                    raise ValueError("Invalid gzip response content.")
+                self._finished = True
+
+            self._buffer.extend(decoded)
+        if size < 0:
+            output = bytes(self._buffer)
+            self._buffer.clear()
+        else:
+            output = bytes(self._buffer[:size])
+            del self._buffer[:size]
+
+        return output
+
+    def close(self) -> None:
+        self._stream.close()
+
+
+class ResponseStream(Stream):
+    def __init__(self, parser: HTTPParser):
+        self._parser = parser
+        self._buffer = io.BytesIO()
+        self._finished = False
+
+    def read(self, size=-1) -> bytes:
+        if self._finished:
+            return b""
+
+        sections = []
+        length = 0
+
+        # If we have any data in the buffer read that and clear the buffer.
+        buffered = self._buffer.read()
+        if buffered:
+            sections.append(buffered)
+            length += len(buffered)
+            self._buffer.seek(0)
+            self._buffer.truncate(0)
+
+        # Read each chunk in turn.
+        while (size < 0) or (length < size):
+            section = self._parser.recv_body()
+            sections.append(section)
+            length += len(section)
+            if section == b"":
+                self._finished = True
+                break
+
+        # If we've more data than requested, then push some back into the buffer.
+        output = b"".join(sections)
+        if size > -1 and len(output) > size:
+            output, remainder = output[:size], output[size:]
+            self._buffer.write(remainder)
+            self._buffer.seek(0)
+
+        return output
+
+    def close(self) -> None:
+        self._buffer.close()
+        self._parser.complete()

src/hip/transports/pool.py → src/hip/client/pool.py

@@ -9,11 +9,11 @@
 from ..backends.sync import NetworkBackend, Semaphore
 from ..core.request import Request
 from ..core.urls import URL
-from .base import SendContext, Transport
 from .connection import (
     Connection,
     open_connection,
 )
+from .transport import SendContext, Transport
 
 __all__ = [
     "ConnectionPool",

src/hip/server/__init__.py

@@ -0,0 +1,3 @@
+from .base import Server
+
+__all__ = ["Server"]

src/hip/server.py → src/hip/server/base.py

@@ -1,14 +1,15 @@
 import datetime
+import io
 import logging
 
-from .backends.sync import NetworkBackend
-from .body import HTTPStream
-from .core.content import Text
-from .core.headers import Headers
-from .core.request import Request
-from .core.response import Response
-from .core.timeouts import sleep
-from .parsers import HTTPParser
+from ..backends.sync import NetworkBackend
+from ..core.content import Text
+from ..core.headers import Headers
+from ..core.request import Request
+from ..core.response import Response
+from ..core.streams import Stream
+from ..core.timeouts import sleep
+from ..parsers import HTTPParser
 
 __all__ = ["Server"]
 
@@ -33,7 +34,7 @@ def handle_requests(self):
                 except EOFError:
                     # The client closed while waiting for a request line.
                     return
-                with HTTPStream(self._parser, is_response=False) as stream:
+                with RequestStream(self._parser) as stream:
                     body = stream.read()
                 request = Request(method, url, headers=headers, content=body)
                 # TODO: Handle endpoint exceptions
@@ -119,3 +120,46 @@ def until_interrupt(self):  # pragma: no cover
     def handle_stream(self, stream):
         connection = HTTPConnection(stream, self.app)
         connection.handle_requests()
+
+
+class RequestStream(Stream):
+    def __init__(self, parser: HTTPParser):
+        self._parser = parser
+        self._buffer = io.BytesIO()
+        self._finished = False
+
+    def read(self, size=-1) -> bytes:
+        if self._finished:
+            return b""
+
+        sections = []
+        length = 0
+
+        # If we have any data in the buffer read that and clear the buffer.
+        buffered = self._buffer.read()
+        if buffered:
+            sections.append(buffered)
+            length += len(buffered)
+            self._buffer.seek(0)
+            self._buffer.truncate(0)
+
+        # Read each chunk in turn.
+        while (size < 0) or (length < size):
+            section = self._parser.recv_body()
+            sections.append(section)
+            length += len(section)
+            if section == b"":
+                self._finished = True
+                break
+
+        # If we've more data than requested, then push some back into the buffer.
+        output = b"".join(sections)
+        if size > -1 and len(output) > size:
+            output, remainder = output[:size], output[size:]
+            self._buffer.write(remainder)
+            self._buffer.seek(0)
+
+        return output
+
+    def close(self) -> None:
+        self._buffer.close()