protocol_grpc: add grpc client

conformance/client_config.yaml

@@ -4,6 +4,7 @@ features:
     - HTTP_VERSION_2
   protocols:
     - PROTOCOL_CONNECT
+    - PROTOCOL_GRPC
   codecs:
     - CODEC_PROTO
   compressions:
@@ -22,4 +23,4 @@ features:
   supports_trailers: true
   supports_half_duplex_bidi_over_http1: true
   supports_connect_get: true
-  supports_message_receive_limit: true
+  supports_message_receive_limit: false

conformance/client_runner.py

@@ -218,6 +218,11 @@ async def handle_message(msg: client_compat_pb2.ClientCompatRequest) -> client_c
         payloads = []
         try:
             options = ClientOptions()
+            if msg.protocol == config_pb2.PROTOCOL_GRPC:
+                options.protocol = "grpc"
+            if msg.protocol == config_pb2.PROTOCOL_GRPC_WEB:
+                options.protocol = "grpc-web"
+
             if msg.compression == config_pb2.COMPRESSION_GZIP:
                 options.request_compression_name = "gzip"
 
@@ -424,8 +429,6 @@ async def read_requests() -> None:
         """Read requests from standard input and process them asynchronously."""
         loop = asyncio.get_event_loop()
         while req := await loop.run_in_executor(None, read_request):
-            await asyncio.sleep(0.01)
-
             loop.create_task(run_message(req))
 
     asyncio.run(read_requests())

conformance/run-testcase.txt

@@ -1 +1 @@
-gRPC Unexpected Requests/HTTPVersion:2/TLS:true/unary/multiple-requests
+Timeouts/HTTPVersion:2/Protocol:PROTOCOL_GRPC/Codec:CODEC_PROTO/Compression:COMPRESSION_IDENTITY/TLS:true/unary

pyproject.toml

@@ -14,14 +14,17 @@ dependencies = [
   "anyio>=4.7.0",
   "googleapis-common-protos>=1.70.0",
   "h2>=4.2.0",
-  "httpcore>=1.0.7",
+  "httpcore",
   "protobuf>=5.29.1",
   "pydantic>=2.10.4",
   "starlette>=0.46.0",
   "types-protobuf>=5.29.1.20241207",
   "yarl>=1.18.3",
 ]
 
+[tool.uv.sources]
+httpcore = { git = "https://github.com/tsubakiky/httpcore" }
+
 [tool.hatch.build.targets.wheel]
 packages = ["src/connect"]
 

src/connect/byte_stream.py

@@ -0,0 +1,56 @@
+"""Asynchronous byte stream utilities for HTTP core response handling."""
+
+from collections.abc import (
+    AsyncIterable,
+    AsyncIterator,
+)
+
+from connect.utils import (
+    AsyncByteStream,
+    get_acallable_attribute,
+    map_httpcore_exceptions,
+)
+
+
+class HTTPCoreResponseAsyncByteStream(AsyncByteStream):
+    """An asynchronous byte stream for reading and writing byte chunks."""
+
+    aiterator: AsyncIterable[bytes] | None
+    _closed: bool
+
+    def __init__(
+        self,
+        aiterator: AsyncIterable[bytes] | None = None,
+    ) -> None:
+        """Initialize the protocol connect instance.
+
+        Args:
+            aiterator (AsyncIterable[bytes] | None): An optional asynchronous iterable of bytes.
+
+        Returns:
+            None
+
+        """
+        self.aiterator = aiterator
+        self._closed = False
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        """Asynchronous iterator method to read byte chunks from the stream."""
+        if self.aiterator:
+            try:
+                with map_httpcore_exceptions():
+                    async for chunk in self.aiterator:
+                        yield chunk
+            except BaseException as exc:
+                await self.aclose()
+                raise exc
+
+    async def aclose(self) -> None:
+        """Asynchronously close the stream."""
+        if not self._closed and self.aiterator:
+            aclose = get_acallable_attribute(self.aiterator, "aclose")
+            if not aclose:
+                return
+
+            with map_httpcore_exceptions():
+                await aclose()

src/connect/client.py

@@ -27,9 +27,11 @@
 from connect.idempotency_level import IdempotencyLevel
 from connect.interceptor import apply_interceptors
 from connect.options import ClientOptions
-from connect.protocol import ProtocolClient, ProtocolClientParams
+from connect.protocol import Protocol, ProtocolClient, ProtocolClientParams
 from connect.protocol_connect import ProtocolConnect
+from connect.protocol_grpc import ProtocolGRPC
 from connect.session import AsyncClientSession
+from connect.utils import aiterate
 
 
 def parse_request_url(raw_url: str) -> URL:
@@ -75,7 +77,7 @@ class ClientConfig:
     """
 
     url: URL
-    protocol: ProtocolConnect
+    protocol: Protocol
     procedure: str
     codec: Codec
     request_compression_name: str | None
@@ -113,6 +115,10 @@ def __init__(self, raw_url: str, options: ClientOptions):
 
         self.url = url
         self.protocol = ProtocolConnect()
+        if options.protocol == "grpc":
+            self.protocol = ProtocolGRPC(web=False)
+        elif options.protocol == "grpc-web":
+            self.protocol = ProtocolGRPC(web=True)
         self.procedure = proto_path
         self.codec = ProtoBinaryCodec()
         self.request_compression_name = options.request_compression_name
@@ -228,9 +234,9 @@ def on_request_send(r: httpcore.Request) -> None:
 
             conn.on_request_send(on_request_send)
 
-            await conn.send(request.message, request.timeout, abort_event=request.abort_event)
+            await conn.send(aiterate([request.message]), request.timeout, abort_event=request.abort_event)
 
-            response = await recieve_unary_response(conn=conn, t=output)
+            response = await recieve_unary_response(conn=conn, t=output, abort_event=request.abort_event)
             return response
 
         unary_func = apply_interceptors(_unary_func, options.interceptors)
@@ -257,7 +263,7 @@ async def call_unary(request: UnaryRequest[T_Request]) -> UnaryResponse[T_Respon
             return response
 
         async def _stream_func(request: StreamRequest[T_Request]) -> StreamResponse[T_Response]:
-            conn = protocol_client.stream_conn(request.spec, request.headers)
+            conn = protocol_client.conn(request.spec, request.headers)
 
             def on_request_send(r: httpcore.Request) -> None:
                 method = r.method