diff --git a/CHANGELOG.md b/CHANGELOG.md index d530eb204..be23933ca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,33 @@ ## Unreleased +### WolverineFx (core) + +- **New `IMessageBus.StreamAsync` primitive for streaming requests.** + The mirror image of `StreamAsync`: a caller hands one handler invocation an + `IAsyncEnumerable` stream of messages and awaits a single `TResponse`. The handler + declares `IAsyncEnumerable` as its message type + (`Task Handle(IAsyncEnumerable messages, CancellationToken token)`) and + consumes the stream incrementally — nothing is materialized by the framework. Local invocation + only; a missing handler fails fast with a `NotSupportedException` naming the expected signature. + Cascading messages and `DeliveryOptions` work as with any invoked handler. See the + [message bus guide](https://wolverinefx.net/guide/messaging/message-bus.html#streaming-requests). + Note: this adds two members to `ICommandBus`, which is source-breaking for custom + `IMessageBus`/`ICommandBus` implementors (same precedent as the original `StreamAsync` addition). + +### WolverineFx.Grpc + +- **Proto-first client-streaming RPCs (`stream TRequest → TResponse`) are now code-generated.** + A `[WolverineGrpcService]` stub declaring the fourth canonical gRPC shape no longer fails fast at + startup — Wolverine generates a wrapper that adapts the inbound `IAsyncStreamReader` to + `IAsyncEnumerable` and forwards it to the new `IMessageBus.StreamAsync` for a + single response. Tenant-id detection applies to client-streaming methods; before/after middleware + and the `Validate` convention are not woven (same constraint as bidirectional streaming). The + server-side exception interceptor now also translates exceptions from client-streaming handlers + per AIP-193, and `IGrpcEndpointManifest` surfaces the new `GrpcRpcStreamKind.ClientStreaming` + descriptors. The code-first (protobuf-net.Grpc) path is unchanged and still skips this shape. See + the [gRPC streaming guide](https://wolverinefx.net/guide/grpc/streaming.html). + ### WolverineFx.Http - **New `openapi` command for build-time OpenAPI generation without starting the host.** diff --git a/docs/guide/grpc/contracts.md b/docs/guide/grpc/contracts.md index f076d5f79..7929762a5 100644 --- a/docs/guide/grpc/contracts.md +++ b/docs/guide/grpc/contracts.md @@ -101,11 +101,13 @@ await foreach (var item in greeter.StreamGreetings(new StreamGreetingsRequest { The [GreeterCodeFirstGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/GreeterCodeFirstGrpc) sample demonstrates this end-to-end. See [Samples](./samples#greetercodefirstgrpc) for a walkthrough. -::: warning Bidirectional streaming is not supported on the generated-implementation path +::: warning Bidirectional and client streaming are not supported on the generated-implementation path The generated implementation recognises **unary** (`Task`) and **server streaming** (`IAsyncEnumerable`) method shapes. An interface method with an `IAsyncEnumerable` -*parameter* (bidirectional streaming) is silently skipped — no startup error, but the method will -not be mapped. Use a hand-written service class for bidi RPCs on code-first contracts. +*parameter* (bidirectional or client streaming) is silently skipped — no startup error, but the +method will not be mapped. Use a hand-written service class for bidi or client-streaming RPCs on +code-first contracts. Proto-first stubs code-generate all four shapes, including +[client streaming](./streaming#client-streaming-proto-first). ::: ::: warning No conflict allowed @@ -147,8 +149,9 @@ public static class PingHandler Any class whose name ends in `GrpcService` is picked up by `MapWolverineGrpcServices()`. If the suffix convention doesn't fit, apply `[WolverineGrpcService]` instead. -Wolverine generates a thin **delegation wrapper** around the class at startup (named -`{ClassName}GrpcHandler`). The wrapper implements the same `[ServiceContract]` interface, weaves +Wolverine generates a thin **delegation wrapper** around the class at startup, named by stripping +any `GrpcService` suffix and appending `GrpcHandler` (`PingGrpcService` → `PingGrpcHandler`). +The wrapper implements the same `[ServiceContract]` interface, weaves any `Validate` / `[WolverineBefore]` middleware defined on the service class, then calls into the inner class — which Wolverine resolves from the DI container or constructs via `ActivatorUtilities` if no explicit registration exists. This gives hand-written service classes @@ -266,6 +269,36 @@ declaration changes. Cancellation from the client propagates into the handler's in both styles, so mid-stream cancellation cleanly unwinds. For the broader streaming story (bidirectional, limitations, timing) see [Streaming](./streaming). +## Client streaming (proto-first) + +A `stream TRequest → TResponse` RPC is code-generated on the proto-first path only. The handler +receives the whole inbound stream as `IAsyncEnumerable` — the message type Wolverine's +[`IMessageBus.StreamAsync`](/guide/messaging/message-bus.html#streaming-requests) dispatches +on — and returns the single response: + +```proto +service Greeter { + rpc CollectGreetings (stream HelloRequest) returns (GreetingSummary); +} +``` + +```csharp +public static class GreeterHandler +{ + public static async Task Handle( + IAsyncEnumerable requests, + CancellationToken cancellationToken) + { + var count = 0; + await foreach (var request in requests.WithCancellation(cancellationToken)) count++; + return new GreetingSummary { Count = count }; + } +} +``` + +See [Streaming — Client streaming](./streaming#client-streaming-proto-first) for the generated +wrapper shape and middleware caveats. + ## Mixing both in one host Nothing stops you from running both styles together. The registration order doesn't matter — call diff --git a/docs/guide/grpc/errors.md b/docs/guide/grpc/errors.md index eb07cb8ef..ee2dcddc3 100644 --- a/docs/guide/grpc/errors.md +++ b/docs/guide/grpc/errors.md @@ -15,7 +15,10 @@ through to the default table. ## Default mapping (AIP-193) `WolverineGrpcExceptionInterceptor` is registered automatically by `AddWolverineGrpc` and applies to -both code-first and proto-first services. It translates ordinary .NET exceptions thrown by handlers +both code-first and proto-first services. It intercepts **unary**, **server-streaming**, and +**client-streaming** RPCs; bidirectional streaming is deliberately not intercepted today, since a +bidi wrapper streams responses incrementally and a trailing translation would arrive after items +had already been written. It translates ordinary .NET exceptions thrown by handlers into `RpcException` with the canonical status code from the table below. See [Overriding the default table](#overriding-the-default-table) if the defaults don't match your domain model. diff --git a/docs/guide/grpc/handlers.md b/docs/guide/grpc/handlers.md index c3ee28001..3c8e496cf 100644 --- a/docs/guide/grpc/handlers.md +++ b/docs/guide/grpc/handlers.md @@ -77,7 +77,8 @@ couples the handler to the gRPC transport and prevents it from being reused over 1. **Code-first (hand-written with wrapper)**: any concrete class whose name ends in `GrpcService` (or that carries `[WolverineGrpcService]`) and implements a `[ServiceContract]` interface gets a - generated **delegation wrapper**. Wolverine emits `{ClassName}GrpcHandler` that implements the same + generated **delegation wrapper**. Wolverine emits a wrapper named by stripping any `GrpcService` + suffix and appending `GrpcHandler` (`PingGrpcService` → `PingGrpcHandler`) that implements the same contract interface, weaves any `Validate` / `[WolverineBefore]` middleware, then delegates each call to the inner class via `ActivatorUtilities`. The inner class does not need an explicit DI registration. 2. **Code-first (generated implementation)**: any **interface** carrying both `[WolverineGrpcService]` @@ -104,7 +105,7 @@ dotnet run -- wolverine-diagnostics codegen-preview --grpc Greeter If you're debugging discovery, `describe` proves Wolverine found the stub; `codegen-preview --grpc` shows the exact generated override and the handler method each RPC forwards to. See [`codegen-preview`](/guide/command-line#codegen-preview) for the full set of accepted identifiers -(bare proto service name, stub class name, or short `-g` alias). +(bare proto service name, stub class name, or generated wrapper name, plus the short `-g` flag alias). ## Validate convention @@ -156,8 +157,8 @@ public Task PlaceOrder(PlaceOrderRequest request, CallContext contex - `ValidateAsync` returning `Task` is also supported when the check is asynchronous. - Validate is matched **per request type**: a `Validate(PlaceOrderRequest)` does not fire for RPC methods whose first parameter is a different request type on the same service class. -- Validate is not woven for **bidirectional streaming** methods — there is no single request - instance in scope before the streaming loop begins. +- Validate is not woven for **bidirectional or client streaming** methods — there is no single + request instance in scope before the stream is consumed. - Validation runs **before** any `[WolverineBefore]` middleware that is not itself a validate hook. ::: tip diff --git a/docs/guide/grpc/index.md b/docs/guide/grpc/index.md index 233c7fccb..7c96a72ce 100644 --- a/docs/guide/grpc/index.md +++ b/docs/guide/grpc/index.md @@ -14,7 +14,9 @@ gRPC gives you another edge protocol for the same handlers. Benefits: - **Strongly-typed contracts** shared across .NET and non-.NET services via `.proto` files, or code-first contracts that never leave C#. -- **Streaming** first-class — plays naturally with Wolverine's [`IMessageBus.StreamAsync`](/guide/messaging/message-bus.html#streaming-responses). +- **Streaming** first-class — server and bidirectional streaming play naturally with Wolverine's + [`IMessageBus.StreamAsync`](/guide/messaging/message-bus.html#streaming-responses), and client + streaming with [`IMessageBus.StreamAsync`](/guide/messaging/message-bus.html#streaming-requests). - **Wolverine handler reuse** — the same handler can back a REST endpoint, an async message, and a gRPC call without duplication. - **Canonical error semantics** — ordinary .NET exceptions thrown by a handler are mapped to the @@ -32,8 +34,8 @@ building: can pick (or mix) them. - [Error Handling](./errors) — the default AIP-193 exception → `StatusCode` table plus the opt-in `google.rpc.Status` pipeline for rich, structured details. -- [Streaming](./streaming) — server streaming today, bidirectional via a manual bridge, and the - shape of the cancellation contract. +- [Streaming](./streaming) — server, client, and bidirectional streaming, and the shape of the + cancellation contract. - [Typed gRPC Clients](./client) — `AddWolverineGrpcClient()`, the Wolverine-flavored wrapper over `Grpc.Net.ClientFactory` that adds envelope-header propagation and `RpcException` → typed-exception translation on the consuming side. @@ -89,7 +91,7 @@ and comparisons to the official `grpc-dotnet` examples. | [PingPongWithGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/PingPongWithGrpc) | Code-first **unary** | `[ServiceContract]` + `WolverineGrpcServiceBase` forwarding to a plain handler | | [PingPongWithGrpcStreaming](https://github.com/JasperFx/wolverine/tree/main/src/Samples/PingPongWithGrpcStreaming) | Code-first **server streaming** | Handler returning `IAsyncEnumerable`, forwarded via `Bus.StreamAsync` | | [GreeterCodeFirstGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/GreeterCodeFirstGrpc) | Code-first **generated implementation** | `[WolverineGrpcService]` on an interface — Wolverine generates the service class; no concrete class written | -| [GreeterProtoFirstGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/GreeterProtoFirstGrpc) | **Proto-first** unary + server streaming + exception mapping | Abstract `[WolverineGrpcService]` stub subclassing a generated `*Base` + handlers | +| [GreeterProtoFirstGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/GreeterProtoFirstGrpc) | **Proto-first** unary + server streaming + client streaming + exception mapping | Abstract `[WolverineGrpcService]` stub subclassing a generated `*Base` + handlers | | [RacerWithGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/RacerWithGrpc) | Code-first **bidirectional streaming** | Per-update bridge: client `IAsyncEnumerable` → `Bus.StreamAsync` for each item | | [GreeterWithGrpcErrors](https://github.com/JasperFx/wolverine/tree/main/src/Samples/GreeterWithGrpcErrors) | Code-first **rich error details** | FluentValidation → `BadRequest` plus inline `MapException` → `PreconditionFailure`, with a client that unpacks both | | [ProgressTrackerWithGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/ProgressTrackerWithGrpc) | Code-first **server streaming + cancellation** | Realistic job-progress stream: handler yields `JobProgress` updates; client cancels mid-stream | @@ -123,9 +125,10 @@ and comparisons to the official `grpc-dotnet` examples. ## Current Limitations -- **Pure client streaming** (`stream TRequest → TResponse`) has no out-of-the-box adapter path yet. - Proto-first stubs that declare this shape fail fast at startup with a clear error rather than - silently skipping. Bidirectional streaming is fully supported — see [Streaming](./streaming). +- **Client streaming is proto-first only.** The code-first (protobuf-net.Grpc) + generated-implementation path does not recognize the `IAsyncEnumerable → Task` + shape — implement those methods by hand against `IMessageBus.StreamAsync`. All four RPC + shapes are code-generated for proto-first stubs — see [Streaming](./streaming). ## Roadmap diff --git a/docs/guide/grpc/multi-tenancy.md b/docs/guide/grpc/multi-tenancy.md index 907117247..c5c537350 100644 --- a/docs/guide/grpc/multi-tenancy.md +++ b/docs/guide/grpc/multi-tenancy.md @@ -153,5 +153,6 @@ Core authorization, which will surface as `Unauthenticated`/`PermissionDenied`. `CallContext` parameter — that's the only route to the underlying `ServerCallContext` and its request metadata. Methods without one fall back to the runtime interceptor. - Code-first **server-streaming** methods (returning `IAsyncEnumerable` directly) can't host - the async detection step; proto-first server-streaming and bidirectional methods (returning - `Task`) are fully covered. + the async detection step; proto-first server-streaming, client-streaming, and bidirectional + methods are fully covered — every proto-first RPC shape ends with a `ServerCallContext` + parameter, which is all detection needs. diff --git a/docs/guide/grpc/samples.md b/docs/guide/grpc/samples.md index 087a5e24d..ff7fd6245 100644 --- a/docs/guide/grpc/samples.md +++ b/docs/guide/grpc/samples.md @@ -24,7 +24,7 @@ changes is whether your business code knows about gRPC. | [PingPongWithGrpc](#pingpongwithgrpc) | Unary | Code-first (hand-written) | [Greeter](https://github.com/grpc/grpc-dotnet/tree/master/examples#greeter) | | [PingPongWithGrpcStreaming](#pingpongwithgrpcstreaming) | Server streaming | Code-first (hand-written) | [Counter](https://github.com/grpc/grpc-dotnet/tree/master/examples#counter) | | [GreeterCodeFirstGrpc](#greetercodefirstgrpc) | Unary + server streaming | Code-first (generated) | [Coder](https://github.com/grpc/grpc-dotnet/tree/master/examples#coder) | -| [GreeterProtoFirstGrpc](#greeterprotofirstgrpc) | Unary + server streaming + exception mapping | Proto-first | [Greeter](https://github.com/grpc/grpc-dotnet/tree/master/examples#greeter) | +| [GreeterProtoFirstGrpc](#greeterprotofirstgrpc) | Unary + server streaming + client streaming + exception mapping | Proto-first | [Greeter](https://github.com/grpc/grpc-dotnet/tree/master/examples#greeter) | | [RacerWithGrpc](#racerwithgrpc) | Bidirectional streaming | Code-first (hand-written) | [Racer](https://github.com/grpc/grpc-dotnet/tree/master/examples#racer) | | [GreeterWithGrpcErrors](#greeterwithgrpcerrors) | Unary + rich error details | Code-first (hand-written) | (no direct equivalent — closest is Greeter + a custom interceptor) | | [ProgressTrackerWithGrpc](#progresstrackerwithgrpc) | Server streaming + cancellation | Code-first (generated) | [Progressor](https://github.com/grpc/grpc-dotnet/tree/master/examples#progressor) | @@ -195,10 +195,10 @@ cancellation story are identical on the wire; only the authoring model differs. Layout: [`src/Samples/GreeterProtoFirstGrpc/`](https://github.com/JasperFx/wolverine/tree/main/src/Samples/GreeterProtoFirstGrpc) with `Messages`, `Server`, `Client`. -A proto-first sample that exercises three things at once: unary RPC, server streaming, and the -default exception → `StatusCode` mapping. The `.proto` declares `SayHello` (unary) and -`StreamGreetings` (server streaming); Grpc.Tools generates `Greeter.GreeterBase`; a single line -hands the rest to Wolverine: +A proto-first sample that exercises four things at once: unary RPC, server streaming, client +streaming, and the default exception → `StatusCode` mapping. The `.proto` declares `SayHello` +(unary), `StreamGreetings` (server streaming), and `CollectGreetings` (client streaming); +Grpc.Tools generates `Greeter.GreeterBase`; a single line hands the rest to Wolverine: ```csharp [WolverineGrpcService] @@ -206,11 +206,15 @@ public abstract class GreeterGrpcService : Greeter.GreeterBase; ``` That's the whole service. Wolverine generates the concrete `GreeterGrpcHandler` wrapper at -startup, and the `Handle` methods on `GreeterHandler` supply the behaviour. +startup, and the `Handle` methods on `GreeterHandler` supply the behaviour — including the +client-streaming handler, which receives the whole inbound stream as +`IAsyncEnumerable` and folds it into one `GreetingSummary` (see +[Streaming — Client streaming](./streaming#client-streaming-proto-first)). -**What to copy**: the abstract-stub pattern for proto-first, plus how throwing -`ArgumentException` / `KeyNotFoundException` / `InvalidOperationException` from a handler yields -the matching gRPC `StatusCode` on the client (see [Error Handling](./errors)). +**What to copy**: the abstract-stub pattern for proto-first, the +`IAsyncEnumerable → Task` handler shape for client streaming, plus how +throwing `ArgumentException` / `KeyNotFoundException` / `InvalidOperationException` from a handler +yields the matching gRPC `StatusCode` on the client (see [Error Handling](./errors)). ### Compared to grpc-dotnet's Greeter @@ -246,8 +250,9 @@ public async IAsyncEnumerable Race( } ``` -**What to copy**: the "one Wolverine stream per incoming command" shape. It's the path to bidi -today until `IMessageBus.StreamAsync` lands. +**What to copy**: the "one Wolverine stream per incoming command" shape. It's still the path to +bidi — `IMessageBus.StreamAsync` exists, but it's the *client-streaming* +primitive (a stream of requests folded into a **single** response), not a bidi loop. ### Compared to grpc-dotnet's Racer diff --git a/docs/guide/grpc/streaming.md b/docs/guide/grpc/streaming.md index 7166771e3..4e8dd64fb 100644 --- a/docs/guide/grpc/streaming.md +++ b/docs/guide/grpc/streaming.md @@ -1,10 +1,10 @@ # Streaming -Wolverine covers **unary**, **server streaming**, and **bidirectional streaming** natively. -**Pure client streaming** (`stream TRequest → TResponse`) has no adapter yet and fails fast at -startup in proto-first mode with a clear diagnostic error. +Wolverine covers all four gRPC RPC shapes natively: **unary**, **server streaming**, +**client streaming** (proto-first), and **bidirectional streaming**. -This page covers server and bidirectional streaming in depth, plus cancellation and current gaps. +This page covers server, client, and bidirectional streaming in depth, plus cancellation and +current gaps. ## Server streaming (first-class) @@ -61,6 +61,76 @@ forces a genuine async yield point and prevents that. You can drop it when your awaits real I/O (a database call, an HTTP request, etc.). ::: +## Client streaming (proto-first) + +Client streaming inverts the server-streaming shape: the client sends a stream of N request +messages and the server answers with a **single** response once the stream completes — think +telemetry ingest (`stream LocationPing → LocationIngestAck`), batched uploads, or metering. + +Declare the RPC in your `.proto` file and mark your stub with `[WolverineGrpcService]` as usual: + +```proto +service TelemetryService { + rpc ReportLocations (stream LocationPing) returns (LocationIngestAck); +} +``` + +```csharp +[WolverineGrpcService] +public abstract class TelemetryServiceStub : TelemetryService.TelemetryServiceBase; +``` + +Wolverine generates the bridge at startup — the inbound `IAsyncStreamReader` is adapted to +`IAsyncEnumerable` and handed to +[`IMessageBus.StreamAsync`](/guide/messaging/message-bus.html#streaming-requests) as a whole: + +```csharp +// Generated by Wolverine +public override async Task ReportLocations( + IAsyncStreamReader requestStream, ServerCallContext context) +{ + return await _bus.StreamAsync( + WolverineGrpcStreamAdapters.ReadAllAsync(requestStream, context.CancellationToken), + context.CancellationToken); +} +``` + +The handler receives the entire inbound stream as its message and folds it into one response: + +```csharp +public static async Task Handle( + IAsyncEnumerable pings, + CancellationToken cancellationToken) +{ + var count = 0; + await foreach (var ping in pings.WithCancellation(cancellationToken)) + { + // process each ping incrementally — nothing is buffered by the framework + count++; + } + + return new LocationIngestAck { Received = count }; +} +``` + +The handler processes items **incrementally** as the client sends them — Wolverine never +materializes the stream into a collection, so memory stays constant regardless of stream length. +An empty stream (client completes without sending anything) still reaches the handler, which +returns its response from a zero-item drain. + +::: info Before-middleware and Validate hooks +Like bidirectional streaming, before-frames (including the `Validate → Status?` short-circuit) +are **not** woven into client-streaming methods in the generated wrapper — they need a single +`TRequest` in scope when the method begins, which the streaming signature doesn't provide. +::: + +::: warning Code-first client streaming is not generated +The client-streaming adapter path is **proto-first only**. On the code-first +(protobuf-net.Grpc) generated-implementation path, an interface method taking +`IAsyncEnumerable` and returning `Task` is not code-generated — implement +such methods by hand, calling `IMessageBus.StreamAsync` yourself. +::: + ## Bidirectional streaming Wolverine supports bidirectional streaming for both **proto-first** (generated wrapper) and @@ -178,14 +248,12 @@ but your detached tasks keep running. Always thread the token through. ## Current limitations -- **Pure client streaming** (`stream TRequest → TResponse`) has no adapter path yet. In proto-first - mode, a service whose `.proto` declares this shape fails fast at startup with a diagnostic - error — it's not silently skipped. If you need this today, implement the service method by hand - without the Wolverine shim, or reshape the contract to server streaming + a final summary - response. -- **Before-middleware and Validate hooks are not woven into bidi methods** in the proto-first - generated wrapper. Use code-first with a manual shim for per-stream authentication or - request-level validation before the loop begins. +- **Client streaming is proto-first only.** The code-first (protobuf-net.Grpc) + generated-implementation path does not recognize the `IAsyncEnumerable → Task` + shape — implement those methods by hand against `IMessageBus.StreamAsync`. +- **Before-middleware and Validate hooks are not woven into bidi or client-streaming methods** in + the proto-first generated wrapper. Use code-first with a manual shim for per-stream + authentication or request-level validation before the loop begins. - **Back-pressure is cooperative, not flow-controlled by default.** HTTP/2 provides windowing, but if your handler produces faster than your client consumes and your DTOs are large, memory usage can spike before backpressure propagates. For large payloads, consider chunking at the contract @@ -206,3 +274,4 @@ but your detached tasks keep running. Always thread the token through. - [Samples](./samples) — `PingPongWithGrpcStreaming`, `ProgressTrackerWithGrpc`, and `RacerWithGrpc` are the canonical streaming walkthroughs, covering server streaming (hand-written), server streaming (generated + cancellation), and bidirectional streaming respectively. + `GreeterProtoFirstGrpc` includes the client-streaming `CollectGreetings` RPC. diff --git a/docs/guide/messaging/message-bus.md b/docs/guide/messaging/message-bus.md index 553e29fca..6eb9bc9ce 100644 --- a/docs/guide/messaging/message-bus.md +++ b/docs/guide/messaging/message-bus.md @@ -287,6 +287,58 @@ Wolverine iterates the sequence and cascades each item as a new message. `Stream when the caller wants to consume the items directly. ::: +## Streaming Requests + +`StreamAsync` also has an inverse overload: `StreamAsync` sends a **stream of +request messages** to one handler invocation and awaits a **single** `Task` — the arity +tells the two apart (one type argument streams responses out; two stream requests in). The handler +declares `IAsyncEnumerable` as its message type and folds the stream however it likes: + +```cs +public static class LocationIngestHandler +{ + public static async Task Handle( + IAsyncEnumerable pings, + CancellationToken cancellationToken) + { + var count = 0; + await foreach (var ping in pings.WithCancellation(cancellationToken)) + { + count++; // process incrementally — nothing is buffered by the framework + } + + return new LocationIngestAck(count); + } +} + +public static async Task ingest(IMessageBus bus, IAsyncEnumerable pings, CancellationToken ct) +{ + var ack = await bus.StreamAsync(pings, ct); + Console.WriteLine($"Ingested {ack.Count} pings"); +} +``` + +A few things worth knowing about `StreamAsync`: + +- **The handler's message type is `IAsyncEnumerable` itself.** Discovery and dispatch key off + that closed generic type, so exactly one handler per element type receives the whole stream. +- **Locally-handled messages only.** A stream can't be serialized to a remote endpoint. If no local handler + accepts `IAsyncEnumerable`, the call fails fast with a `NotSupportedException` naming the + expected handler signature. +- **Consumption is incremental.** Wolverine hands the live stream to the handler without materializing + it — memory stays constant no matter how many items the caller streams. An empty stream still invokes + the handler, which returns its response from a zero-item drain. +- **Cancellation propagates into the handler** through the handler's `CancellationToken` parameter, and + from there into the stream via `WithCancellation`. +- **Cascading messages work as usual.** Return a tuple to both answer the caller and publish follow-on + messages, exactly like any other invoked handler. +- **`DeliveryOptions` is supported** for headers, tenant id, and correlation metadata via the overload + `StreamAsync(IAsyncEnumerable messages, DeliveryOptions options, ...)`. + +This is the primitive behind gRPC client streaming — see +[gRPC Services / Streaming](/guide/grpc/streaming#client-streaming-proto-first) for exposing a +stream-folding handler over the wire. + ## Sending or Publishing Messages [Publish/Subscribe](https://docs.microsoft.com/en-us/azure/architecture/patterns/publisher-subscriber) is a messaging pattern where the senders of messages do not need to specifically know what the specific subscribers are for a given message. In this case, some kind of middleware or infrastructure is responsible for either allowing subscribers to express interest in what messages they need to receive or apply routing rules to send the published messages to the right places. Wolverine's messaging support was largely built to support the publish/subscribe messaging pattern. diff --git a/docs/guide/samples.md b/docs/guide/samples.md index 5c6c43083..61360cba0 100644 --- a/docs/guide/samples.md +++ b/docs/guide/samples.md @@ -25,7 +25,9 @@ There are several sample projects in the Wolverine codebase showing off bits and | [WolverineChat](https://github.com/JasperFx/wolverine/tree/main/src/Samples/WolverineChat) | Small SignalR chat application used by the SignalR transport documentation — runs without any Docker dependencies | | [PingPongWithGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/PingPongWithGrpc) | Code-first **unary** gRPC — walkthrough: [gRPC Samples / PingPongWithGrpc](/guide/grpc/samples#pingpongwithgrpc) | | [PingPongWithGrpcStreaming](https://github.com/JasperFx/wolverine/tree/main/src/Samples/PingPongWithGrpcStreaming) | Code-first **server streaming** gRPC — walkthrough: [gRPC Samples / PingPongWithGrpcStreaming](/guide/grpc/samples#pingpongwithgrpcstreaming) | -| [GreeterProtoFirstGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/GreeterProtoFirstGrpc) | **Proto-first** gRPC (unary + streaming + exception mapping) — walkthrough: [gRPC Samples / GreeterProtoFirstGrpc](/guide/grpc/samples#greeterprotofirstgrpc) | +| [GreeterCodeFirstGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/GreeterCodeFirstGrpc) | Code-first **generated implementation** gRPC — `[WolverineGrpcService]` on the interface, no service class written — walkthrough: [gRPC Samples / GreeterCodeFirstGrpc](/guide/grpc/samples#greetercodefirstgrpc) | +| [GreeterProtoFirstGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/GreeterProtoFirstGrpc) | **Proto-first** gRPC (unary + server/client streaming + exception mapping) — walkthrough: [gRPC Samples / GreeterProtoFirstGrpc](/guide/grpc/samples#greeterprotofirstgrpc) | | [RacerWithGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/RacerWithGrpc) | Code-first **bidirectional streaming** gRPC — walkthrough: [gRPC Samples / RacerWithGrpc](/guide/grpc/samples#racerwithgrpc) | | [GreeterWithGrpcErrors](https://github.com/JasperFx/wolverine/tree/main/src/Samples/GreeterWithGrpcErrors) | Code-first gRPC with **rich error details** (FluentValidation → `BadRequest`, domain exceptions → `PreconditionFailure`) — walkthrough: [gRPC Samples / GreeterWithGrpcErrors](/guide/grpc/samples#greeterwithgrpcerrors) | +| [ProgressTrackerWithGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/ProgressTrackerWithGrpc) | Code-first **server streaming + mid-stream cancellation** gRPC (generated implementation) — walkthrough: [gRPC Samples / ProgressTrackerWithGrpc](/guide/grpc/samples#progresstrackerwithgrpc) | | [OrderChainWithGrpc](https://github.com/JasperFx/wolverine/tree/main/src/Samples/OrderChainWithGrpc) | Two Wolverine gRPC services chained via `AddWolverineGrpcClient()` — proves envelope-header propagation and typed-exception round-trip across a hop with no user plumbing. Walkthrough: [gRPC Samples / OrderChainWithGrpc](/guide/grpc/samples#orderchainwithgrpc) | diff --git a/src/Samples/GreeterCodeFirstGrpc/Server/Server.csproj b/src/Samples/GreeterCodeFirstGrpc/Server/Server.csproj index 41b706497..d41916677 100644 --- a/src/Samples/GreeterCodeFirstGrpc/Server/Server.csproj +++ b/src/Samples/GreeterCodeFirstGrpc/Server/Server.csproj @@ -9,6 +9,9 @@ + + diff --git a/src/Samples/GreeterProtoFirstGrpc/Client/Program.cs b/src/Samples/GreeterProtoFirstGrpc/Client/Program.cs index d456534c1..a795d1cc1 100644 --- a/src/Samples/GreeterProtoFirstGrpc/Client/Program.cs +++ b/src/Samples/GreeterProtoFirstGrpc/Client/Program.cs @@ -26,6 +26,17 @@ Console.WriteLine($" {reply.Message}"); } +// Client streaming: stream several HelloRequests, get back one folded GreetingSummary. +using var collect = client.CollectGreetings(); +foreach (var name in new[] { "Erik", "Ripley", "Newt" }) +{ + await collect.RequestStream.WriteAsync(new HelloRequest { Name = name }); +} + +await collect.RequestStream.CompleteAsync(); +var summary = await collect; +Console.WriteLine($"CollectGreetings -> {summary.Message} ({summary.Count} greetings)"); + // Exception mapping (AIP-193): handler throws KeyNotFoundException → client sees NotFound. try { diff --git a/src/Samples/GreeterProtoFirstGrpc/Messages/Protos/greeter.proto b/src/Samples/GreeterProtoFirstGrpc/Messages/Protos/greeter.proto index eac8cdd3a..0c2816e1f 100644 --- a/src/Samples/GreeterProtoFirstGrpc/Messages/Protos/greeter.proto +++ b/src/Samples/GreeterProtoFirstGrpc/Messages/Protos/greeter.proto @@ -5,11 +5,12 @@ option csharp_namespace = "GreeterProtoFirstGrpc.Messages"; package greet; // Proto-first service exercising Wolverine's GrpcServiceChain code generation. -// Covers unary and server-streaming shapes. +// Covers unary, server-streaming, and client-streaming shapes. service Greeter { rpc SayHello (HelloRequest) returns (HelloReply); rpc SayGoodbye (GoodbyeRequest) returns (GoodbyeReply); rpc StreamGreetings (StreamGreetingsRequest) returns (stream HelloReply); + rpc CollectGreetings (stream HelloRequest) returns (GreetingSummary); rpc Fault (FaultRequest) returns (FaultReply); } @@ -34,6 +35,13 @@ message StreamGreetingsRequest { int32 count = 2; } +// Single reply folding a client-streamed batch of HelloRequests — the handler +// receives the whole inbound stream as IAsyncEnumerable. +message GreetingSummary { + int32 count = 1; + string message = 2; +} + // Exercises the gRPC exception interceptor + mapper by asking the handler to // throw a specific exception kind. AIP-193 mapping applies automatically. message FaultRequest { diff --git a/src/Samples/GreeterProtoFirstGrpc/README.md b/src/Samples/GreeterProtoFirstGrpc/README.md index 48d4e7911..ae4fc3498 100644 --- a/src/Samples/GreeterProtoFirstGrpc/README.md +++ b/src/Samples/GreeterProtoFirstGrpc/README.md @@ -6,8 +6,8 @@ abstract `[WolverineGrpcService] abstract class GreeterGrpcService : Greeter.Gre — Wolverine generates the concrete subclass at startup that forwards each RPC to the bus. -Covers unary RPCs, a server-streaming RPC, and the AIP-193 -exception-to-StatusCode mapping. +Covers unary RPCs, a server-streaming RPC, a client-streaming RPC, and the +AIP-193 exception-to-StatusCode mapping. - `Messages` — `greeter.proto` (`GrpcServices="Both"` generates both sides). - `Server` — ASP.NET Core + Wolverine host. Handlers are plain Wolverine @@ -39,9 +39,16 @@ StreamGreetings -> Hello, Erik [2] Hello, Erik [3] Hello, Erik [4] +CollectGreetings -> Hello, Erik & Ripley & Newt (3 greetings) Fault('key') -> RpcException: NotFound (missing key) ``` +The `CollectGreetings` line demonstrates client streaming: the client streams +several `HelloRequest` messages, and the generated wrapper hands the whole +inbound stream to a Wolverine handler as `IAsyncEnumerable` via +`IMessageBus.StreamAsync`, which folds it into a single +`GreetingSummary` reply. + The `Fault('key')` line comes from a handler that throws `KeyNotFoundException`. The built-in interceptor maps it to `StatusCode.NotFound` per AIP-193 — no user code translates between the diff --git a/src/Samples/GreeterProtoFirstGrpc/Server/GreeterGrpcService.cs b/src/Samples/GreeterProtoFirstGrpc/Server/GreeterGrpcService.cs index 785d83cac..d1f40a205 100644 --- a/src/Samples/GreeterProtoFirstGrpc/Server/GreeterGrpcService.cs +++ b/src/Samples/GreeterProtoFirstGrpc/Server/GreeterGrpcService.cs @@ -6,8 +6,9 @@ namespace GreeterProtoFirstGrpc.Server; /// /// Proto-first Wolverine gRPC stub. The proto-generated Greeter.GreeterBase /// supplies the gRPC contract; at startup, Wolverine code-generates a concrete -/// subclass of this stub that overrides every unary RPC and forwards it to -/// — no bridging code is written by hand. +/// subclass of this stub that overrides every RPC — unary, server-streaming, and +/// client-streaming alike — and forwards it to the matching +/// operation. No bridging code is written by hand. /// [WolverineGrpcService] public abstract class GreeterGrpcService : Greeter.GreeterBase; diff --git a/src/Samples/GreeterProtoFirstGrpc/Server/GreeterHandler.cs b/src/Samples/GreeterProtoFirstGrpc/Server/GreeterHandler.cs index a7c5bcf49..d78182287 100644 --- a/src/Samples/GreeterProtoFirstGrpc/Server/GreeterHandler.cs +++ b/src/Samples/GreeterProtoFirstGrpc/Server/GreeterHandler.cs @@ -28,6 +28,26 @@ public static async IAsyncEnumerable Handle( } } + // Client streaming: the generated wrapper adapts the RPC's inbound stream to + // IAsyncEnumerable and forwards it via IMessageBus.StreamAsync, + // so the handler folds N requests into one reply. + public static async Task Handle( + IAsyncEnumerable requests, + CancellationToken cancellationToken) + { + var names = new List(); + await foreach (var request in requests.WithCancellation(cancellationToken)) + { + names.Add(request.Name); + } + + return new GreetingSummary + { + Count = names.Count, + Message = names.Count == 0 ? "Hello, nobody" : $"Hello, {string.Join(" & ", names)}" + }; + } + public static FaultReply Handle(FaultRequest request) => throw FaultExceptions.Throw(request.Kind); } diff --git a/src/Samples/GreeterProtoFirstGrpc/Server/Server.csproj b/src/Samples/GreeterProtoFirstGrpc/Server/Server.csproj index b5f85ec1d..7c6804f90 100644 --- a/src/Samples/GreeterProtoFirstGrpc/Server/Server.csproj +++ b/src/Samples/GreeterProtoFirstGrpc/Server/Server.csproj @@ -9,6 +9,9 @@ + + diff --git a/src/Samples/GreeterWithGrpcErrors/Server/Server.csproj b/src/Samples/GreeterWithGrpcErrors/Server/Server.csproj index 61e1bbf28..728536e25 100644 --- a/src/Samples/GreeterWithGrpcErrors/Server/Server.csproj +++ b/src/Samples/GreeterWithGrpcErrors/Server/Server.csproj @@ -9,6 +9,9 @@ + + diff --git a/src/Samples/OrderChainWithGrpc/InventoryServer/InventoryServer.csproj b/src/Samples/OrderChainWithGrpc/InventoryServer/InventoryServer.csproj index d4f05e71c..4cde1438d 100644 --- a/src/Samples/OrderChainWithGrpc/InventoryServer/InventoryServer.csproj +++ b/src/Samples/OrderChainWithGrpc/InventoryServer/InventoryServer.csproj @@ -9,6 +9,9 @@ + + diff --git a/src/Samples/OrderChainWithGrpc/OrderServer/OrderServer.csproj b/src/Samples/OrderChainWithGrpc/OrderServer/OrderServer.csproj index ffa3b55f5..6d00c4a84 100644 --- a/src/Samples/OrderChainWithGrpc/OrderServer/OrderServer.csproj +++ b/src/Samples/OrderChainWithGrpc/OrderServer/OrderServer.csproj @@ -9,6 +9,9 @@ + + diff --git a/src/Samples/PingPongWithGrpc/Ponger/Ponger.csproj b/src/Samples/PingPongWithGrpc/Ponger/Ponger.csproj index a1a70c7ef..956cd1ea8 100644 --- a/src/Samples/PingPongWithGrpc/Ponger/Ponger.csproj +++ b/src/Samples/PingPongWithGrpc/Ponger/Ponger.csproj @@ -9,6 +9,9 @@ + + diff --git a/src/Samples/PingPongWithGrpcStreaming/Ponger/Ponger.csproj b/src/Samples/PingPongWithGrpcStreaming/Ponger/Ponger.csproj index b17230191..05949c67a 100644 --- a/src/Samples/PingPongWithGrpcStreaming/Ponger/Ponger.csproj +++ b/src/Samples/PingPongWithGrpcStreaming/Ponger/Ponger.csproj @@ -9,6 +9,9 @@ + + diff --git a/src/Samples/ProgressTrackerWithGrpc/Server/Server.csproj b/src/Samples/ProgressTrackerWithGrpc/Server/Server.csproj index d34881252..4e580d033 100644 --- a/src/Samples/ProgressTrackerWithGrpc/Server/Server.csproj +++ b/src/Samples/ProgressTrackerWithGrpc/Server/Server.csproj @@ -9,6 +9,9 @@ + + diff --git a/src/Samples/RacerWithGrpc/README.md b/src/Samples/RacerWithGrpc/README.md index 267fd6c87..b73adacfc 100644 --- a/src/Samples/RacerWithGrpc/README.md +++ b/src/Samples/RacerWithGrpc/README.md @@ -7,8 +7,10 @@ computes the current standings on every update and streams back a Architecturally: the gRPC service consumes the client stream itself and calls `IMessageBus.StreamAsync(update)` per inbound item, re-yielding -each `RacePosition` the handler produces. Bidi "just works" without -Wolverine accepting an inbound `IAsyncEnumerable` on the bus. +each `RacePosition` the handler produces. Bidi "just works" without a +bidi-shaped primitive on the bus — `IMessageBus.StreamAsync` +accepts an inbound stream, but folds it into a single response (client +streaming), so the per-item bridge is still the bidi pattern. - `RacerContracts` — `[ServiceContract] IRacingService.RaceAsync(IAsyncEnumerable)`. - `RacerServer` — host with a singleton `RaceState` and `RaceStreamHandler`. diff --git a/src/Samples/RacerWithGrpc/RacerServer/RacerServer.csproj b/src/Samples/RacerWithGrpc/RacerServer/RacerServer.csproj index d179e3fde..13fd83242 100644 --- a/src/Samples/RacerWithGrpc/RacerServer/RacerServer.csproj +++ b/src/Samples/RacerWithGrpc/RacerServer/RacerServer.csproj @@ -9,6 +9,9 @@ + + diff --git a/src/Testing/CoreTests/Acceptance/streaming_request_support.cs b/src/Testing/CoreTests/Acceptance/streaming_request_support.cs new file mode 100644 index 000000000..9304d34a8 --- /dev/null +++ b/src/Testing/CoreTests/Acceptance/streaming_request_support.cs @@ -0,0 +1,314 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using System.Runtime.CompilerServices; +using Wolverine.Tracking; +using Xunit; + +namespace CoreTests.Acceptance; + +// --------------------------------------------------------------------------- +// Message types +// --------------------------------------------------------------------------- + +public record NumberToSum(int Value); + +public record NumberSum(int Total, int Count); + +// Separate element type routed to a handler without a CancellationToken parameter. +public record PlainNumber(int Value); + +// Separate element type so the cascading test routes to a dedicated handler that +// returns a (response, cascading message) tuple. +public record CascadingNumber(int Value); + +public record StreamIngestionCompleted(int Count); + +// Separate element type routed to a handler that throws mid-drain. +public record FaultingNumber(int Value); + +// Deliberately has NO stream handler — used to assert the clear error message. +public record UnhandledNumber(int Value); + +// --------------------------------------------------------------------------- +// Handlers +// --------------------------------------------------------------------------- + +public static class NumberStreamHandler +{ + public static async Task Handle(IAsyncEnumerable numbers, + CancellationToken cancellationToken) + { + var total = 0; + var count = 0; + await foreach (var number in numbers.WithCancellation(cancellationToken)) + { + total += number.Value; + count++; + } + + return new NumberSum(total, count); + } +} + +public static class PlainNumberStreamHandler +{ + public static async Task Handle(IAsyncEnumerable numbers) + { + var total = 0; + var count = 0; + await foreach (var number in numbers) + { + total += number.Value; + count++; + } + + return new NumberSum(total, count); + } +} + +public static class CascadingNumberStreamHandler +{ + public static async Task<(NumberSum, StreamIngestionCompleted)> Handle( + IAsyncEnumerable numbers) + { + var total = 0; + var count = 0; + await foreach (var number in numbers) + { + total += number.Value; + count++; + } + + return (new NumberSum(total, count), new StreamIngestionCompleted(count)); + } +} + +public static class StreamIngestionCompletedHandler +{ + public static void Handle(StreamIngestionCompleted completed, StreamCompletionTracker tracker) + { + tracker.Add(completed); + } +} + +public class StreamCompletionTracker +{ + private readonly List _completions = new(); + public IReadOnlyList Completions => _completions; + public void Add(StreamIngestionCompleted completed) => _completions.Add(completed); +} + +public static class FaultingNumberStreamHandler +{ + public static async Task Handle(IAsyncEnumerable numbers) + { + var count = 0; + await foreach (var _ in numbers) + { + count++; + if (count >= 2) + { + throw new InvalidOperationException("stream handler faulted mid-drain"); + } + } + + return new NumberSum(0, count); + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +public class streaming_request_support +{ + private static async IAsyncEnumerable toStream(IEnumerable items) + { + foreach (var item in items) + { + yield return item; + await Task.Yield(); + } + } + + [Fact] + public async Task stream_request_returns_aggregated_response() + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine() + .StartAsync(); + + var bus = host.MessageBus(); + + var numbers = toStream(Enumerable.Range(1, 4).Select(i => new NumberToSum(i))); + var sum = await bus.StreamAsync(numbers); + + sum.Total.ShouldBe(10); + sum.Count.ShouldBe(4); + } + + [Fact] + public async Task empty_stream_still_returns_response() + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine() + .StartAsync(); + + var bus = host.MessageBus(); + + var sum = await bus.StreamAsync(toStream(Array.Empty())); + + sum.Total.ShouldBe(0); + sum.Count.ShouldBe(0); + } + + [Fact] + public async Task handler_without_cancellation_token_parameter_works() + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine() + .StartAsync(); + + var bus = host.MessageBus(); + + var numbers = toStream([new PlainNumber(5), new PlainNumber(7)]); + var sum = await bus.StreamAsync(numbers); + + sum.Total.ShouldBe(12); + sum.Count.ShouldBe(2); + } + + [Fact] + public async Task no_stream_handler_throws_clear_not_supported() + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine() + .StartAsync(); + + var bus = host.MessageBus(); + + var ex = await Should.ThrowAsync(async () => + { + await bus.StreamAsync(toStream([new UnhandledNumber(1)])); + }); + + ex.Message.ShouldContain(nameof(UnhandledNumber)); + ex.Message.ShouldContain("IAsyncEnumerable"); + } + + [Fact] + public async Task handler_exception_mid_drain_surfaces_to_caller() + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine() + .StartAsync(); + + var bus = host.MessageBus(); + + var numbers = toStream(Enumerable.Range(0, 10).Select(i => new FaultingNumber(i))); + var ex = await Should.ThrowAsync(async () => + { + await bus.StreamAsync(numbers); + }); + + ex.Message.ShouldBe("stream handler faulted mid-drain"); + } + + [Fact] + public async Task cancellation_propagates_into_the_handler() + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine() + .StartAsync(); + + var bus = host.MessageBus(); + + using var cts = new CancellationTokenSource(); + + async IAsyncEnumerable infinite( + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + var i = 0; + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return new NumberToSum(i++); + if (i >= 2) + { + await cts.CancelAsync(); + } + + await Task.Yield(); + } + // ReSharper disable once IteratorNeverReturns + } + + await Should.ThrowAsync(async () => + { + await bus.StreamAsync(infinite(), cts.Token); + }); + } + + [Fact] + public async Task cascading_messages_from_stream_handler_are_published() + { + var tracker = new StreamCompletionTracker(); + + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Services.AddSingleton(tracker); + }) + .StartAsync(); + + NumberSum? sum = null; + await host.ExecuteAndWaitAsync(async context => + { + var numbers = toStream([new CascadingNumber(2), new CascadingNumber(3)]); + sum = await context.StreamAsync(numbers); + }); + + sum.ShouldNotBeNull(); + sum.Total.ShouldBe(5); + sum.Count.ShouldBe(2); + + tracker.Completions.Count.ShouldBe(1); + tracker.Completions[0].Count.ShouldBe(2); + } + + [Fact] + public async Task stream_request_with_delivery_options() + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine() + .StartAsync(); + + var bus = host.MessageBus(); + var options = new DeliveryOptions(); + + var numbers = toStream([new NumberToSum(1), new NumberToSum(2)]); + var sum = await bus.StreamAsync(numbers, options); + + sum.Total.ShouldBe(3); + } + + [Fact] + public async Task ordinary_single_message_invoke_is_unaffected_by_stream_chains() + { + // The IAsyncEnumerable chain registered in this assembly must not + // interfere with normal single-message request/reply on unrelated types. + using var host = await Host.CreateDefaultBuilder() + .UseWolverine() + .StartAsync(); + + var bus = host.MessageBus(); + + var items = new List(); + await foreach (var item in bus.StreamAsync(new StreamRequest(2))) + { + items.Add(item); + } + + items.Count.ShouldBe(2); + } +} diff --git a/src/Testing/CoreTests/TestMessageContextTests.cs b/src/Testing/CoreTests/TestMessageContextTests.cs index c9854cbbd..c03a37b5d 100644 --- a/src/Testing/CoreTests/TestMessageContextTests.cs +++ b/src/Testing/CoreTests/TestMessageContextTests.cs @@ -256,6 +256,36 @@ public async Task invoke_with_expected_response_and_filter_miss() ex.Message.ShouldStartWith("There is no matching expectation for the request message"); } + [Fact] + public async Task stream_request_records_invocation_and_returns_configured_response() + { + var response = new NumberResponse(21); + theSpy.WhenInvokedMessageOf>().RespondWith(response); + + var stream = numberRequests(); + (await theContext.StreamAsync(stream)) + .ShouldBeSameAs(response); + + theSpy.Invoked.Single().ShouldBeSameAs(stream); + } + + [Fact] + public async Task stream_request_with_expected_response_miss() + { + var ex = await Should.ThrowAsync(async () => + { + await theContext.StreamAsync(numberRequests()); + }); + + ex.Message.ShouldStartWith("There is no matching expectation for the request message"); + } + + private static async IAsyncEnumerable numberRequests() + { + yield return new NumberRequest(1, 2); + await Task.Yield(); + } + [Fact] public async Task invoke_with_expected_response_no_filter_hit_to_endpoint_by_uri() { diff --git a/src/Wolverine.Grpc.Tests/GrpcBidiStreaming/Protos/grpc_bidi_test.proto b/src/Wolverine.Grpc.Tests/GrpcBidiStreaming/Protos/grpc_bidi_test.proto index ce537f506..44f1feda1 100644 --- a/src/Wolverine.Grpc.Tests/GrpcBidiStreaming/Protos/grpc_bidi_test.proto +++ b/src/Wolverine.Grpc.Tests/GrpcBidiStreaming/Protos/grpc_bidi_test.proto @@ -11,8 +11,9 @@ service BidiEchoTest { rpc Echo (stream EchoRequest) returns (stream EchoReply); } -// Client-streaming service — used only by unit tests that verify the fail-fast error message -// when a proto-first stub declares a client-streaming RPC. Never registered with a host. +// Client-streaming service — used only by classification/chain-construction unit tests. +// Never registered with a host; the end-to-end client-streaming coverage lives in +// GrpcClientStreaming/Protos/grpc_client_stream_test.proto. service ClientStreamTest { rpc Collect (stream EchoRequest) returns (EchoReply); } diff --git a/src/Wolverine.Grpc.Tests/GrpcBidiStreaming/grpc_bidi_streaming_tests.cs b/src/Wolverine.Grpc.Tests/GrpcBidiStreaming/grpc_bidi_streaming_tests.cs index 78b2a4b69..213065893 100644 --- a/src/Wolverine.Grpc.Tests/GrpcBidiStreaming/grpc_bidi_streaming_tests.cs +++ b/src/Wolverine.Grpc.Tests/GrpcBidiStreaming/grpc_bidi_streaming_tests.cs @@ -117,17 +117,17 @@ public void classifies_client_streaming_method_correctly() } /// -/// Verifies the fail-fast contract: constructing a from a -/// proto-first stub that declares a client-streaming RPC must throw -/// immediately, before any code generation runs. -/// This prevents silent no-ops at runtime (the generated wrapper would have no method to -/// delegate client-streaming requests to). +/// Client-streaming stubs were rejected at chain construction with +/// until the IMessageBus.StreamAsync adapter path existed. These tests pin the +/// inverted contract: construction succeeds and the RPC is classified onto +/// . The end-to-end wire behavior lives in +/// GrpcClientStreaming/grpc_client_streaming_tests. /// [Collection("GrpcSerialTests")] -public class grpc_client_streaming_fail_fast_tests +public class grpc_client_streaming_chain_construction_tests { [Fact] - public async Task stub_with_client_streaming_method_throws_not_supported_at_chain_construction() + public async Task stub_with_client_streaming_method_builds_a_chain_with_the_method_classified() { DynamicCodeBuilder.WithinCodegenCommand = true; try @@ -139,13 +139,13 @@ public async Task stub_with_client_streaming_method_throws_not_supported_at_chai var graph = host.Services.GetRequiredService(); - var ex = Should.Throw( - () => new GrpcServiceChain(typeof(ClientStreamingOnlyStub), graph)); + var chain = new GrpcServiceChain(typeof(ClientStreamingOnlyStub), graph); - // Message must name the unsupported shape and the offending method so - // the user can immediately identify what to fix. - ex.Message.ShouldContain("Client-streaming"); - ex.Message.ShouldContain("Collect"); + chain.ClientStreamingMethods.Count.ShouldBe(1); + chain.ClientStreamingMethods[0].Name.ShouldBe("Collect"); + chain.UnaryMethods.ShouldBeEmpty(); + chain.ServerStreamingMethods.ShouldBeEmpty(); + chain.BidirectionalStreamingMethods.ShouldBeEmpty(); } finally { @@ -155,5 +155,5 @@ public async Task stub_with_client_streaming_method_throws_not_supported_at_chai } // Internal so GetExportedTypes() skips it — it must never land in the proto-first -// discovery scan and break the BidiStreamingFixture that shares this assembly. +// discovery scan and change what the BidiStreamingFixture that shares this assembly maps. internal abstract class ClientStreamingOnlyStub : ClientStreamTest.ClientStreamTestBase; diff --git a/src/Wolverine.Grpc.Tests/GrpcClientStreaming/ClientStreamingFixture.cs b/src/Wolverine.Grpc.Tests/GrpcClientStreaming/ClientStreamingFixture.cs new file mode 100644 index 000000000..8ee711845 --- /dev/null +++ b/src/Wolverine.Grpc.Tests/GrpcClientStreaming/ClientStreamingFixture.cs @@ -0,0 +1,60 @@ +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.DependencyInjection; +using Wolverine.Grpc.Tests.GrpcClientStreaming.Generated; +using Xunit; + +namespace Wolverine.Grpc.Tests.GrpcClientStreaming; + +/// +/// Boots an in-process gRPC host to exercise proto-first client-streaming code-gen +/// end-to-end. Isolated from other fixtures so client-streaming assertions don't drift +/// when unary/server-streaming/bidi tests evolve. +/// +public class ClientStreamingFixture : IAsyncLifetime +{ + private WebApplication? _app; + public GrpcChannel? Channel { get; private set; } + public IServiceProvider Services => _app?.Services + ?? throw new InvalidOperationException("Fixture has not been initialized yet."); + + public async Task InitializeAsync() + { + var builder = WebApplication.CreateBuilder([]); + builder.WebHost.UseTestServer(); + + builder.Host.UseWolverine(opts => + { + opts.ApplicationAssembly = typeof(ClientStreamingFixture).Assembly; + }); + + builder.Services.AddGrpc(); + builder.Services.AddWolverineGrpc(); + + _app = builder.Build(); + _app.UseRouting(); + _app.MapWolverineGrpcServices(); + + await _app.StartAsync(); + + var handler = _app.GetTestServer().CreateHandler(); + Channel = GrpcChannel.ForAddress("http://localhost", new GrpcChannelOptions + { + HttpHandler = handler + }); + } + + public async Task DisposeAsync() + { + Channel?.Dispose(); + if (_app != null) + { + await _app.StopAsync(); + await _app.DisposeAsync(); + } + } + + public CollectTest.CollectTestClient CreateClient() + => new(Channel!); +} diff --git a/src/Wolverine.Grpc.Tests/GrpcClientStreaming/CollectHandler.cs b/src/Wolverine.Grpc.Tests/GrpcClientStreaming/CollectHandler.cs new file mode 100644 index 000000000..323371cdd --- /dev/null +++ b/src/Wolverine.Grpc.Tests/GrpcClientStreaming/CollectHandler.cs @@ -0,0 +1,24 @@ +using Wolverine.Grpc.Tests.GrpcClientStreaming.Generated; + +namespace Wolverine.Grpc.Tests.GrpcClientStreaming; + +/// +/// Wolverine handler for the client-streaming shape: receives the whole inbound RPC +/// stream as and folds it into a single reply. +/// +public static class CollectHandler +{ + public static async Task Handle(IAsyncEnumerable numbers, + CancellationToken cancellationToken) + { + var total = 0; + var count = 0; + await foreach (var number in numbers.WithCancellation(cancellationToken)) + { + total += number.Value; + count++; + } + + return new SumReply { Total = total, Count = count }; + } +} diff --git a/src/Wolverine.Grpc.Tests/GrpcClientStreaming/CollectStub.cs b/src/Wolverine.Grpc.Tests/GrpcClientStreaming/CollectStub.cs new file mode 100644 index 000000000..5bb715489 --- /dev/null +++ b/src/Wolverine.Grpc.Tests/GrpcClientStreaming/CollectStub.cs @@ -0,0 +1,10 @@ +using Wolverine.Grpc.Tests.GrpcClientStreaming.Generated; + +namespace Wolverine.Grpc.Tests.GrpcClientStreaming; + +/// +/// Proto-first stub for the client-streaming tests. Carries no extra methods — +/// the generated wrapper's Collect override is produced entirely by Wolverine codegen. +/// +[WolverineGrpcService] +public abstract class CollectStub : CollectTest.CollectTestBase; diff --git a/src/Wolverine.Grpc.Tests/GrpcClientStreaming/Protos/grpc_client_stream_test.proto b/src/Wolverine.Grpc.Tests/GrpcClientStreaming/Protos/grpc_client_stream_test.proto new file mode 100644 index 000000000..4665f8cbd --- /dev/null +++ b/src/Wolverine.Grpc.Tests/GrpcClientStreaming/Protos/grpc_client_stream_test.proto @@ -0,0 +1,22 @@ +syntax = "proto3"; + +option csharp_namespace = "Wolverine.Grpc.Tests.GrpcClientStreaming.Generated"; + +package wolverine.grpc.tests.client_streaming; + +// Test-only service exercising the client-streaming generated wrapper for proto-first +// Wolverine gRPC stubs. The client streams NumberRequest messages; the generated wrapper +// forwards the whole stream to IMessageBus.StreamAsync and the handler folds it into +// a single SumReply so tests can assert the many-in/one-out contract. +service CollectTest { + rpc Collect (stream NumberRequest) returns (SumReply); +} + +message NumberRequest { + int32 value = 1; +} + +message SumReply { + int32 total = 1; + int32 count = 2; +} diff --git a/src/Wolverine.Grpc.Tests/GrpcClientStreaming/grpc_client_streaming_tests.cs b/src/Wolverine.Grpc.Tests/GrpcClientStreaming/grpc_client_streaming_tests.cs new file mode 100644 index 000000000..f24285676 --- /dev/null +++ b/src/Wolverine.Grpc.Tests/GrpcClientStreaming/grpc_client_streaming_tests.cs @@ -0,0 +1,77 @@ +using Grpc.Core; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; +using Wolverine.Grpc.Tests.GrpcClientStreaming.Generated; +using Xunit; + +namespace Wolverine.Grpc.Tests.GrpcClientStreaming; + +/// +/// End-to-end tests for the proto-first client-streaming generated wrapper. Verifies that +/// the generated code adapts the inbound stream to +/// IAsyncEnumerable<NumberRequest>, forwards it to +/// IMessageBus.StreamAsync, and returns the handler's single . +/// +public class grpc_client_streaming_tests : IClassFixture +{ + private readonly ClientStreamingFixture _fixture; + + public grpc_client_streaming_tests(ClientStreamingFixture fixture) + { + _fixture = fixture; + } + + [Fact] + public async Task streamed_requests_fold_into_a_single_reply() + { + using var call = _fixture.CreateClient().Collect(); + + await call.RequestStream.WriteAsync(new NumberRequest { Value = 1 }); + await call.RequestStream.WriteAsync(new NumberRequest { Value = 2 }); + await call.RequestStream.WriteAsync(new NumberRequest { Value = 3 }); + await call.RequestStream.CompleteAsync(); + + var reply = await call; + + reply.Total.ShouldBe(6); + reply.Count.ShouldBe(3); + } + + [Fact] + public async Task zero_requests_still_produce_a_reply() + { + using var call = _fixture.CreateClient().Collect(); + await call.RequestStream.CompleteAsync(); + + var reply = await call; + + reply.Total.ShouldBe(0); + reply.Count.ShouldBe(0); + } + + [Fact] + public async Task cancelling_the_call_aborts_with_cancelled_status() + { + using var cts = new CancellationTokenSource(); + using var call = _fixture.CreateClient().Collect(cancellationToken: cts.Token); + + await call.RequestStream.WriteAsync(new NumberRequest { Value = 1 }); + await cts.CancelAsync(); + + var ex = await Should.ThrowAsync(async () => await call); + ex.StatusCode.ShouldBe(StatusCode.Cancelled); + } + + [Fact] + public void client_streaming_method_is_classified_on_the_chain() + { + var graph = _fixture.Services.GetRequiredService(); + var chain = graph.Chains.Single(c => c.StubType == typeof(CollectStub)); + + chain.ClientStreamingMethods.Count.ShouldBe(1); + chain.ClientStreamingMethods[0].Name.ShouldBe("Collect"); + chain.UnaryMethods.ShouldBeEmpty(); + chain.ServerStreamingMethods.ShouldBeEmpty(); + chain.BidirectionalStreamingMethods.ShouldBeEmpty(); + } +} diff --git a/src/Wolverine.Grpc.Tests/ProtoFirst/proto_first_grpc_tests.cs b/src/Wolverine.Grpc.Tests/ProtoFirst/proto_first_grpc_tests.cs index ef28935db..146aed3aa 100644 --- a/src/Wolverine.Grpc.Tests/ProtoFirst/proto_first_grpc_tests.cs +++ b/src/Wolverine.Grpc.Tests/ProtoFirst/proto_first_grpc_tests.cs @@ -71,6 +71,22 @@ public async Task round_trip_server_streaming_call_through_generated_wrapper() received.ShouldBe(["Hello, Erik [0]", "Hello, Erik [1]", "Hello, Erik [2]"]); } + [Fact] + public async Task round_trip_client_streaming_call_through_generated_wrapper() + { + var client = new Greeter.GreeterClient(_fixture.Channel); + + using var call = client.CollectGreetings(); + await call.RequestStream.WriteAsync(new HelloRequest { Name = "Erik" }); + await call.RequestStream.WriteAsync(new HelloRequest { Name = "Ripley" }); + await call.RequestStream.CompleteAsync(); + + var summary = await call; + + summary.Count.ShouldBe(2); + summary.Message.ShouldBe("Hello, Erik & Ripley"); + } + [Fact] public async Task mid_stream_cancellation_stops_enumeration_early() { @@ -172,7 +188,7 @@ public void discovers_every_virtual_unary_method_on_the_proto_base() } [Fact] - public void classifies_unary_and_server_streaming_methods_distinctly() + public void classifies_unary_and_streaming_methods_distinctly() { var classified = GrpcServiceChain.DiscoverSupportedMethods(typeof(Greeter.GreeterBase)) .ToDictionary(m => m.Method.Name, m => m.Kind); @@ -180,6 +196,7 @@ public void classifies_unary_and_server_streaming_methods_distinctly() classified["SayHello"].ShouldBe(GrpcMethodKind.Unary); classified["SayGoodbye"].ShouldBe(GrpcMethodKind.Unary); classified["StreamGreetings"].ShouldBe(GrpcMethodKind.ServerStreaming); + classified["CollectGreetings"].ShouldBe(GrpcMethodKind.ClientStreaming); } [Fact] @@ -191,7 +208,7 @@ public void discovered_methods_are_sorted_alphabetically_for_byte_stable_codegen .Select(m => m.Method.Name) .ToList(); - names.ShouldBe(["Fault", "SayGoodbye", "SayHello", "StreamGreetings"]); + names.ShouldBe(["CollectGreetings", "Fault", "SayGoodbye", "SayHello", "StreamGreetings"]); } [Fact] diff --git a/src/Wolverine.Grpc.Tests/Wolverine.Grpc.Tests.csproj b/src/Wolverine.Grpc.Tests/Wolverine.Grpc.Tests.csproj index ae5a9931d..cea3bdb5c 100644 --- a/src/Wolverine.Grpc.Tests/Wolverine.Grpc.Tests.csproj +++ b/src/Wolverine.Grpc.Tests/Wolverine.Grpc.Tests.csproj @@ -64,6 +64,7 @@ avoids shadowing the Wolverine.Attributes.MiddlewareScoping enum from sibling tests. --> + diff --git a/src/Wolverine.Grpc.Tests/grpc_capabilities_descriptor_source_3267.cs b/src/Wolverine.Grpc.Tests/grpc_capabilities_descriptor_source_3267.cs index 288dea4c0..5f55d0911 100644 --- a/src/Wolverine.Grpc.Tests/grpc_capabilities_descriptor_source_3267.cs +++ b/src/Wolverine.Grpc.Tests/grpc_capabilities_descriptor_source_3267.cs @@ -191,6 +191,24 @@ public async Task capabilities_include_the_bidirectional_streaming_origin() echo.Tags.ShouldContain("grpc"); } + [Fact] + public async Task capabilities_include_the_client_streaming_origin() + { + var runtime = _fixture.Services.GetRequiredService(); + var capabilities = await ServiceCapabilities.ReadFrom(runtime, null, CancellationToken.None); + + var collect = capabilities.GrpcEndpoints.Single(e => + e.ServiceName == "CollectTest" && e.MethodName == "Collect"); + + collect.StreamKind.ShouldBe(GrpcRpcStreamKind.ClientStreaming); + collect.Mode.ShouldBe(GrpcServiceDiscoveryMode.ProtoFirst); + // The surfaced message is the per-item element type of the inbound request stream; the actual bus + // message is IAsyncEnumerable. The response is unwrapped from Task. + collect.RequestType!.FullName.ShouldBe(typeof(GrpcClientStreaming.Generated.NumberRequest).FullName); + collect.ResponseType!.FullName.ShouldBe(typeof(GrpcClientStreaming.Generated.SumReply).FullName); + collect.Tags.ShouldContain("grpc"); + } + [Fact] public async Task capabilities_only_surface_bus_forwarding_modes() { diff --git a/src/Wolverine.Grpc.Tests/grpc_endpoint_manifest_3265.cs b/src/Wolverine.Grpc.Tests/grpc_endpoint_manifest_3265.cs index 63f845ae4..399597a90 100644 --- a/src/Wolverine.Grpc.Tests/grpc_endpoint_manifest_3265.cs +++ b/src/Wolverine.Grpc.Tests/grpc_endpoint_manifest_3265.cs @@ -134,6 +134,22 @@ public void proto_first_server_streaming_stream_greetings() e.HandlerType.ShouldBe(typeof(GreeterGrpcService)); } + // --- proto-first: client-streaming ----------------------------------------------------------------------------- + + [Fact] + public void proto_first_client_streaming_collect_greetings() + { + var e = protoFirst("CollectGreetings"); + e.StreamKind.ShouldBe(GrpcRpcStreamKind.ClientStreaming); + e.ServiceName.ShouldBe("Greeter"); + // The surfaced message is the per-item element type of the inbound request stream; the actual bus + // message is IAsyncEnumerable. + e.RequestType.ShouldBe(typeof(HelloRequest)); + // The response is unwrapped from Task. + e.ResponseType.ShouldBe(typeof(GreetingSummary)); + e.HandlerType.ShouldBe(typeof(GreeterGrpcService)); + } + // --- code-first: unary ----------------------------------------------------------------------------------------- [Fact] @@ -261,6 +277,23 @@ public void manifest_never_surfaces_hand_written_or_direct_mapped_modes() e.Mode == GrpcServiceDiscoveryMode.ProtoFirst || e.Mode == GrpcServiceDiscoveryMode.CodeFirst); } + [Fact] + public void proto_first_client_streaming_collect_is_surfaced() + { + var collect = Endpoints.Single(e => + e.Mode == GrpcServiceDiscoveryMode.ProtoFirst + && e.ServiceName == "CollectTest" + && e.MethodName == "Collect"); + + collect.StreamKind.ShouldBe(GrpcRpcStreamKind.ClientStreaming); + // The surfaced message is the per-item element type of the inbound request stream — NumberRequest — NOT + // the IAsyncStreamReader wrapper (the actual bus message is IAsyncEnumerable). + collect.RequestType.ShouldBe(typeof(GrpcClientStreaming.Generated.NumberRequest)); + // The response is unwrapped from Task. + collect.ResponseType.ShouldBe(typeof(GrpcClientStreaming.Generated.SumReply)); + collect.HandlerType.ShouldBe(typeof(GrpcClientStreaming.CollectStub)); + } + [Fact] public void every_surfaced_endpoint_carries_a_request_message_and_known_stream_kind() { @@ -268,6 +301,7 @@ public void every_surfaced_endpoint_carries_a_request_message_and_known_stream_k e.RequestType != null && (e.StreamKind == GrpcRpcStreamKind.Unary || e.StreamKind == GrpcRpcStreamKind.ServerStreaming - || e.StreamKind == GrpcRpcStreamKind.BidirectionalStreaming)); + || e.StreamKind == GrpcRpcStreamKind.BidirectionalStreaming + || e.StreamKind == GrpcRpcStreamKind.ClientStreaming)); } } diff --git a/src/Wolverine.Grpc/GrpcEndpointManifest.cs b/src/Wolverine.Grpc/GrpcEndpointManifest.cs index fd83b52ed..37f8d73fc 100644 --- a/src/Wolverine.Grpc/GrpcEndpointManifest.cs +++ b/src/Wolverine.Grpc/GrpcEndpointManifest.cs @@ -60,8 +60,7 @@ private static IReadOnlyList build(GrpcGraph graph) var descriptors = new List(); // Proto-first: every RPC kind Wolverine forwards to the bus. The chain pre-classifies its methods into - // unary / server-streaming / bidirectional-streaming lists (client-streaming is rejected at construction, - // so it never reaches here). + // unary / server-streaming / client-streaming / bidirectional-streaming lists. foreach (var chain in graph.Chains) { // Unary: Task Name(TRequest, ServerCallContext) → InvokeAsync(request). @@ -110,6 +109,25 @@ private static IReadOnlyList build(GrpcGraph graph) GrpcServiceDiscoveryMode.ProtoFirst, GrpcRpcStreamKind.BidirectionalStreaming)); } + + // Client-streaming: Task Name(IAsyncStreamReader, ServerCallContext). + // The whole inbound stream is forwarded via StreamAsync, so the surfaced request type is the + // per-item element type of the request stream (the actual bus message is IAsyncEnumerable). + foreach (var method in chain.ClientStreamingMethods) + { + var p = method.GetParameters(); + var requestType = genericArgument(p[0].ParameterType); + if (requestType == null) continue; // defensive: a client-streaming reader always has an element type + + descriptors.Add(new GrpcEndpointDescriptor( + chain.ProtoServiceName, + method.Name, + requestType, + genericArgument(method.ReturnType), + chain.StubType, + GrpcServiceDiscoveryMode.ProtoFirst, + GrpcRpcStreamKind.ClientStreaming)); + } } // Code-first: unary and server-streaming are bus-forwarded (no bidi shape in the code-first model). diff --git a/src/Wolverine.Grpc/GrpcServiceChain.cs b/src/Wolverine.Grpc/GrpcServiceChain.cs index daa57972a..85bb09216 100644 --- a/src/Wolverine.Grpc/GrpcServiceChain.cs +++ b/src/Wolverine.Grpc/GrpcServiceChain.cs @@ -15,9 +15,11 @@ namespace Wolverine.Grpc; /// /// Represents a proto-first gRPC service that Wolverine will wrap with a generated -/// concrete subclass. Each method on the proto-generated *Base class that -/// matches the unary-RPC signature Task<TResponse>(TRequest, ServerCallContext) -/// is overridden to forward to . +/// concrete subclass. Each virtual RPC method on the proto-generated *Base class is +/// overridden to forward to the matching operation for its shape: +/// unary → , server/bidirectional streaming → +/// , client streaming → +/// . /// public class GrpcServiceChain : Chain, ICodeFile { @@ -63,7 +65,6 @@ public GrpcServiceChain(Type stubType, GrpcGraph parent) + "The stub must derive from a class carrying [BindServiceMethod] (generated by Grpc.Tools from a .proto file)."); SupportedMethods = DiscoverSupportedMethods(ProtoServiceBase).ToArray(); - AssertNoUnsupportedStreamingKinds(stubType, SupportedMethods); UnaryMethods = SupportedMethods .Where(m => m.Kind == GrpcMethodKind.Unary) @@ -73,6 +74,10 @@ public GrpcServiceChain(Type stubType, GrpcGraph parent) .Where(m => m.Kind == GrpcMethodKind.ServerStreaming) .Select(m => m.Method) .ToArray(); + ClientStreamingMethods = SupportedMethods + .Where(m => m.Kind == GrpcMethodKind.ClientStreaming) + .Select(m => m.Method) + .ToArray(); BidirectionalStreamingMethods = SupportedMethods .Where(m => m.Kind == GrpcMethodKind.BidirectionalStreaming) .Select(m => m.Method) @@ -94,6 +99,13 @@ public GrpcServiceChain(Type stubType, GrpcGraph parent) /// public IReadOnlyList ServerStreamingMethods { get; } + /// + /// Client-streaming RPC methods (Task<TResponse> Name(IAsyncStreamReader<TRequest>, ServerCallContext)) + /// that adapt the inbound stream to and forward it to + /// for a single response. + /// + public IReadOnlyList ClientStreamingMethods { get; } + /// /// Bidirectional-streaming RPC methods /// (Task Name(IAsyncStreamReader<TRequest>, IServerStreamWriter<TResponse>, ServerCallContext)) @@ -216,6 +228,9 @@ void ICodeFile.AssembleTypes(GeneratedAssembly assembly) assembly.ReferenceAssembly(ProtoServiceBase.Assembly); assembly.ReferenceAssembly(typeof(IMessageBus).Assembly); assembly.ReferenceAssembly(typeof(ServerCallContext).Assembly); + // Wolverine.Grpc — home of WolverineGrpcStreamAdapters.ReadAllAsync, used by the + // client-streaming forward frame to adapt IAsyncStreamReader to IAsyncEnumerable + assembly.ReferenceAssembly(typeof(WolverineGrpcStreamAdapters).Assembly); _generatedType = assembly.AddType(TypeName, StubType); @@ -252,9 +267,9 @@ void ICodeFile.AssembleTypes(GeneratedAssembly assembly) } // Before-frames (including Validate short-circuit) require a concrete TRequest - // in scope when the method begins. Bidi methods start with an IAsyncStreamReader - // rather than a single T — per-call middleware is not woven for bidi methods. - if (rpc.Kind != GrpcMethodKind.BidirectionalStreaming) + // in scope when the method begins. Bidi and client-streaming methods start with an + // IAsyncStreamReader rather than a single T — per-call middleware is not woven for them. + if (rpc.Kind is not (GrpcMethodKind.BidirectionalStreaming or GrpcMethodKind.ClientStreaming)) { // Registered middleware befores (from grpc.AddMiddleware()) — cloned per method. foreach (var frame in CodeFirstGrpcServiceChain.CloneFrames(Middleware)) @@ -290,13 +305,18 @@ void ICodeFile.AssembleTypes(GeneratedAssembly assembly) generatedMethod.Frames.Add(new ForwardServerStreamToMessageBusFrame(rpc.Method, busField)); break; + case GrpcMethodKind.ClientStreaming: + generatedMethod.AsyncMode = AsyncMode.AsyncTask; + generatedMethod.Frames.Add(new ForwardClientStreamToMessageBusFrame(rpc.Method, busField)); + break; + case GrpcMethodKind.BidirectionalStreaming: generatedMethod.AsyncMode = AsyncMode.AsyncTask; generatedMethod.Frames.Add(new ForwardBidiStreamToMessageBusFrame(rpc.Method, busField)); break; } - if (rpc.Kind != GrpcMethodKind.BidirectionalStreaming) + if (rpc.Kind is not (GrpcMethodKind.BidirectionalStreaming or GrpcMethodKind.ClientStreaming)) { // Inline after-hooks declared directly on the stub class. foreach (var after in afters) @@ -379,8 +399,8 @@ public static IEnumerable DiscoverUnaryMethods(Type protoServiceBase /// /// Walks the proto-generated service base, classifying each virtual RPC method by its signature. - /// Methods with shapes Wolverine can't yet generate (client-streaming, bidirectional-streaming) - /// are still returned — the chain constructor fails fast rather than silently skipping them. + /// All four canonical gRPC shapes (unary, server-streaming, client-streaming, bidirectional) are + /// code-generated; methods matching none of them (e.g., user-added non-RPC virtuals) are skipped. /// Results are sorted by method name so generated source is byte-stable across runs, which /// keeps diagnostic diffs and code-gen caches deterministic even when reflection reorders methods. /// @@ -458,31 +478,6 @@ public static IEnumerable DiscoverSupportedMethods(Type protoServ private static bool IsGenericOf(Type t, Type openGeneric) => t.IsGenericType && t.GetGenericTypeDefinition() == openGeneric; - - private static void AssertNoUnsupportedStreamingKinds(Type stubType, IReadOnlyList methods) - { - var unsupported = methods - .Where(m => m.Kind is GrpcMethodKind.ClientStreaming) - .ToList(); - - if (unsupported.Count == 0) return; - - var detail = unsupported - .Select(m => $" - {m.Method.Name} ({m.Kind})") - .Aggregate((a, b) => a + Environment.NewLine + b); - - throw new NotSupportedException( - $"Proto-first gRPC stub {stubType.FullNameInCode()} declares RPC method(s) whose shape " - + "Wolverine cannot yet code-generate. Supported today: unary, server-streaming, and bidirectional-streaming. " - + "Client-streaming (stream TRequest → TResponse) has no adapter path yet." - + Environment.NewLine - + "Unsupported method(s):" - + Environment.NewLine - + detail - + Environment.NewLine - + "Workaround: move the affected RPC(s) into a separate service whose stub is NOT marked " - + "[WolverineGrpcService], and implement those methods by hand (calling IMessageBus directly)."); - } } /// @@ -506,7 +501,9 @@ public enum GrpcMethodKind /// /// Client-streaming RPC: Task<TResponse> Name(IAsyncStreamReader<TRequest>, ServerCallContext). - /// Not yet code-generated by Wolverine — detection fails fast at startup rather than silently skipping. + /// The inbound stream is adapted to and forwarded to + /// + /// for a single response. /// ClientStreaming, @@ -669,6 +666,62 @@ public override void GenerateCode(GeneratedMethod method, ISourceWriter writer) } } +/// +/// Emits a call bridging a gRPC client-streaming RPC to Wolverine's +/// : +/// the is adapted to via +/// and the single response is returned. +/// +/// +/// Generated code shape: +/// +/// return await _bus.StreamAsync<TRequest, TResponse>( +/// WolverineGrpcStreamAdapters.ReadAllAsync(requestStream, context.CancellationToken), +/// context.CancellationToken); +/// +/// Before-frames (including the Validate short-circuit) are not woven into client-streaming +/// methods for the same reason as bidirectional ones: they require a concrete TRequest +/// in scope before the stream is consumed, which the signature does not provide. +/// +internal sealed class ForwardClientStreamToMessageBusFrame : AsyncFrame +{ + private readonly MethodInfo _rpc; + private readonly InjectedField _busField; + + public ForwardClientStreamToMessageBusFrame(MethodInfo rpc, InjectedField busField) + { + _rpc = rpc; + _busField = busField; + } + + public override void GenerateCode(GeneratedMethod method, ISourceWriter writer) + { + var parameters = _rpc.GetParameters(); + var readerName = ForwardUnaryToMessageBusFrame.ParameterName(parameters, 0); + var contextName = ForwardUnaryToMessageBusFrame.ParameterName(parameters, 1); + + // IAsyncStreamReader → TRequest; Task → TResponse + var requestType = parameters[0].ParameterType.GetGenericArguments()[0]; + var responseType = _rpc.ReturnType.GetGenericArguments()[0]; + var cancellation = $"{contextName}.{nameof(ServerCallContext.CancellationToken)}"; + + var busInvoke = + $"{_busField.Usage}.{nameof(IMessageBus.StreamAsync)}<{requestType.FullNameInCode()}, {responseType.FullNameInCode()}>" + + $"({typeof(WolverineGrpcStreamAdapters).FullNameInCode()}.{nameof(WolverineGrpcStreamAdapters.ReadAllAsync)}({readerName}, {cancellation}), {cancellation})"; + + if (Next == null) + { + writer.Write($"return await {busInvoke};"); + } + else + { + writer.Write($"var result = await {busInvoke};"); + Next.GenerateCode(method, writer); + writer.Write("return result;"); + } + } +} + /// /// Emits a Status? null-check that short-circuits RPC execution when a /// Validate / ValidateAsync method on the proto-first stub returns diff --git a/src/Wolverine.Grpc/WolverineGrpcExceptionInterceptor.cs b/src/Wolverine.Grpc/WolverineGrpcExceptionInterceptor.cs index 5fd063689..b5e0f00a4 100644 --- a/src/Wolverine.Grpc/WolverineGrpcExceptionInterceptor.cs +++ b/src/Wolverine.Grpc/WolverineGrpcExceptionInterceptor.cs @@ -33,9 +33,10 @@ namespace Wolverine.Grpc; /// /// /// Applies to both code-first (protobuf-net.Grpc) and proto-first (Grpc.Tools) services, -/// since both route through the same ASP.NET Core gRPC pipeline. Unary and server-streaming -/// RPCs are intercepted; client-streaming and bidirectional are deferred pending the -/// matching IMessageBus overloads. +/// since both route through the same ASP.NET Core gRPC pipeline. Unary, server-streaming, +/// and client-streaming RPCs are intercepted; bidirectional remains deferred — its generated +/// wrapper streams responses incrementally, so a trailing exception translation would arrive +/// after items have already been written. /// /// public sealed class WolverineGrpcExceptionInterceptor : Interceptor @@ -66,6 +67,21 @@ public override async Task UnaryServerHandler( } } + public override async Task ClientStreamingServerHandler( + IAsyncStreamReader requestStream, + ServerCallContext context, + ClientStreamingServerMethod continuation) + { + try + { + return await continuation(requestStream, context); + } + catch (Exception ex) when (ex is not RpcException) + { + throw Translate(ex, context); + } + } + public override async Task ServerStreamingServerHandler( TRequest request, IServerStreamWriter responseStream, diff --git a/src/Wolverine.Grpc/WolverineGrpcStreamAdapters.cs b/src/Wolverine.Grpc/WolverineGrpcStreamAdapters.cs new file mode 100644 index 000000000..22d6c3215 --- /dev/null +++ b/src/Wolverine.Grpc/WolverineGrpcStreamAdapters.cs @@ -0,0 +1,30 @@ +using System.Runtime.CompilerServices; +using Grpc.Core; + +namespace Wolverine.Grpc; + +/// +/// Adapters used by Wolverine's generated gRPC wrappers to bridge gRPC streaming primitives +/// to the shapes works with. +/// +/// +/// Grpc.Core.Api and Grpc.Net.Common both declare a Grpc.Core.AsyncStreamReaderExtensions +/// type, which makes any generated call to their ReadAllAsync ambiguous (CS0433) in an +/// application referencing both. Wolverine's own adapter sidesteps the collision. +/// +public static class WolverineGrpcStreamAdapters +{ + /// + /// Exposes an as an so a + /// client-streaming RPC's inbound stream can be handed to + /// . + /// + public static async IAsyncEnumerable ReadAllAsync(IAsyncStreamReader reader, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + while (await reader.MoveNext(cancellationToken).ConfigureAwait(false)) + { + yield return reader.Current; + } + } +} diff --git a/src/Wolverine/Configuration/GrpcEndpointManifest.cs b/src/Wolverine/Configuration/GrpcEndpointManifest.cs index 3d4beec87..d08fff49e 100644 --- a/src/Wolverine/Configuration/GrpcEndpointManifest.cs +++ b/src/Wolverine/Configuration/GrpcEndpointManifest.cs @@ -22,8 +22,8 @@ public enum GrpcServiceDiscoveryMode /// /// The RPC cardinality of a discovered gRPC endpoint — the dimension CritterWatch needs to render the right -/// chain-detail affordance for a gRPC origin. Only the kinds whose generated wrapper forwards the request to the -/// Wolverine message bus are represented; client-streaming has no Wolverine forwarding path and never appears. +/// chain-detail affordance for a gRPC origin. Every kind's generated wrapper forwards the request(s) to the +/// Wolverine message bus. /// public enum GrpcRpcStreamKind { @@ -37,14 +37,21 @@ public enum GrpcRpcStreamKind /// Bidirectional-streaming RPC — each inbound request-stream item is forwarded via IMessageBus.StreamAsync. /// Only proto-first services reach this shape today. /// - BidirectionalStreaming + BidirectionalStreaming, + + /// + /// Client-streaming RPC — the inbound request stream is forwarded as a whole via + /// IMessageBus.StreamAsync for a single response. Only proto-first services reach this shape today. + /// + ClientStreaming } /// /// A discovered gRPC endpoint and the Wolverine message type it forwards to the message bus. For proto-first and /// code-first services the generated wrapper forwards the request to the bus, so is the /// published Wolverine message. Unary RPCs forward via IMessageBus.InvokeAsync; server- and -/// bidirectional-streaming RPCs forward via IMessageBus.StreamAsync. Hand-written and direct-mapped services +/// bidirectional-streaming RPCs forward via IMessageBus.StreamAsync; client-streaming RPCs forward the whole +/// inbound stream via IMessageBus.StreamAsync. Hand-written and direct-mapped services /// are excluded — Wolverine does not own their dispatch, so there is no reliable message-publishing origin to surface. /// /// The service name — the proto service's simple name for proto-first, or the contract's @@ -52,12 +59,14 @@ public enum GrpcRpcStreamKind /// The RPC method name. /// The request type — the Wolverine message forwarded to the bus. For unary and /// server-streaming RPCs this is the request parameter; for bidirectional-streaming it is the per-item element type of -/// the inbound request stream (each item is forwarded individually). -/// The response type — unwrapped from Task<T> for unary RPCs, or the element -/// type of the outbound response stream for streaming RPCs; null for a method with no typed response. +/// the inbound request stream (each item is forwarded individually); for client-streaming it is the element type of +/// the inbound stream (the actual bus message is IAsyncEnumerable<RequestType>). +/// The response type — unwrapped from Task<T> for unary and client-streaming +/// RPCs, or the element type of the outbound response stream for server-/bidirectional-streaming RPCs; null +/// for a method with no typed response. /// The identity of the discovered service (stub or contract type). /// How the service was discovered. -/// The RPC cardinality (unary, server-streaming, or bidirectional-streaming). +/// The RPC cardinality (unary, server-streaming, client-streaming, or bidirectional-streaming). public sealed record GrpcEndpointDescriptor( string ServiceName, string MethodName, @@ -76,8 +85,8 @@ public interface IGrpcEndpointManifest { /// /// The discovered gRPC endpoints whose generated wrapper forwards the request to the message bus — unary, - /// server-streaming, and bidirectional-streaming RPCs across proto-first and code-first services. Empty when - /// gRPC is enabled but no such services were discovered. + /// server-streaming, client-streaming, and bidirectional-streaming RPCs across proto-first and code-first + /// services. Empty when gRPC is enabled but no such services were discovered. /// IReadOnlyList Endpoints { get; } } diff --git a/src/Wolverine/IMessageBus.cs b/src/Wolverine/IMessageBus.cs index 4b5dac573..ba5b30c66 100644 --- a/src/Wolverine/IMessageBus.cs +++ b/src/Wolverine/IMessageBus.cs @@ -135,6 +135,37 @@ public interface ICommandBus /// /// IAsyncEnumerable StreamAsync(object message, DeliveryOptions options, CancellationToken cancellation = default); + + /// + /// Execute the message handling for an inbound stream of messages right now and wait for the + /// single response. The handler must accept as its + /// message type, e.g. Task<TResponse> Handle(IAsyncEnumerable<TRequest> messages, CancellationToken token). + /// Only supported for locally-handled messages. + /// + /// + /// + /// Optional timeout + /// + /// + /// + Task StreamAsync(IAsyncEnumerable messages, + CancellationToken cancellation = default, TimeSpan? timeout = default); + + /// + /// Execute the message handling for an inbound stream of messages right now and wait for the + /// single response. The handler must accept as its + /// message type, e.g. Task<TResponse> Handle(IAsyncEnumerable<TRequest> messages, CancellationToken token). + /// Only supported for locally-handled messages. + /// + /// + /// Use to pass in extra metadata like headers or group id or correlation information to the command execution + /// + /// Optional timeout + /// + /// + /// + Task StreamAsync(IAsyncEnumerable messages, + DeliveryOptions options, CancellationToken cancellation = default, TimeSpan? timeout = default); } /// diff --git a/src/Wolverine/Runtime/MessageBus.cs b/src/Wolverine/Runtime/MessageBus.cs index c76ae4242..bbc35ee6b 100644 --- a/src/Wolverine/Runtime/MessageBus.cs +++ b/src/Wolverine/Runtime/MessageBus.cs @@ -1,5 +1,6 @@ using System.Diagnostics; using JasperFx.Core; +using JasperFx.Core.Reflection; using JasperFx.MultiTenancy; using Wolverine.Persistence.Durability; using Wolverine.Runtime.Routing; @@ -214,6 +215,47 @@ public IAsyncEnumerable StreamAsync(object message, Delive return Runtime.FindInvoker(message.GetType()).StreamAsync(message, this, cancellation, options); } + public Task StreamAsync(IAsyncEnumerable messages, + CancellationToken cancellation = default, TimeSpan? timeout = default) + { + if (messages == null) + { + throw new ArgumentNullException(nameof(messages)); + } + + Runtime.AssertHasStarted(); + + return findStreamInvoker().InvokeAsync(messages, this, cancellation, timeout); + } + + public Task StreamAsync(IAsyncEnumerable messages, + DeliveryOptions options, CancellationToken cancellation = default, TimeSpan? timeout = default) + { + if (messages == null) + { + throw new ArgumentNullException(nameof(messages)); + } + + Runtime.AssertHasStarted(); + + return findStreamInvoker().InvokeAsync(messages, this, cancellation, timeout, options); + } + + private IMessageInvoker findStreamInvoker() + { + // The concrete runtime type of an IAsyncEnumerable instance is a compiler-generated + // iterator, so dispatch must key off the declared stream type rather than message.GetType() + var messageType = typeof(IAsyncEnumerable); + if (!Runtime.Options.HandlerGraph.CanHandle(messageType)) + { + throw new NotSupportedException( + $"StreamAsync is only supported for locally-handled message streams, and no handler accepts {messageType.FullNameInCode()} as its message type. " + + $"Define a handler like 'Task Handle(IAsyncEnumerable<{typeof(TRequest).FullNameInCode()}> messages, CancellationToken token)'."); + } + + return Runtime.FindInvoker(messageType); + } + public IReadOnlyList PreviewSubscriptions(object message) { return Runtime.RoutingFor(message.GetType()).RouteForPublish(message, null); diff --git a/src/Wolverine/TestMessageContext.cs b/src/Wolverine/TestMessageContext.cs index 157833a61..077a5458e 100644 --- a/src/Wolverine/TestMessageContext.cs +++ b/src/Wolverine/TestMessageContext.cs @@ -237,6 +237,27 @@ IAsyncEnumerable ICommandBus.StreamAsync(object message, D return EmptyAsyncEnumerable(cancellation); } + Task ICommandBus.StreamAsync(IAsyncEnumerable messages, + CancellationToken cancellation, TimeSpan? timeout) + { + _invoked.Add(messages); + + var response = findResponse(messages); + return Task.FromResult(response); + } + + Task ICommandBus.StreamAsync(IAsyncEnumerable messages, + DeliveryOptions options, CancellationToken cancellation, TimeSpan? timeout) + { + var envelope = new Envelope(messages); + options.Override(envelope); + + _invoked.Add(envelope); + + var response = findResponse(messages); + return Task.FromResult(response); + } + private static async IAsyncEnumerable EmptyAsyncEnumerable( [EnumeratorCancellation] CancellationToken cancellation = default) { diff --git a/src/Wolverine/Transports/Sending/SendingEnvelopeLifecycle.cs b/src/Wolverine/Transports/Sending/SendingEnvelopeLifecycle.cs index b89f3535f..478df8288 100644 --- a/src/Wolverine/Transports/Sending/SendingEnvelopeLifecycle.cs +++ b/src/Wolverine/Transports/Sending/SendingEnvelopeLifecycle.cs @@ -155,4 +155,12 @@ public IAsyncEnumerable StreamAsync(object message, Cancel public IAsyncEnumerable StreamAsync(object message, DeliveryOptions options, CancellationToken cancellation = default) => _bus.StreamAsync(message, options, cancellation); + + public Task StreamAsync(IAsyncEnumerable messages, + CancellationToken cancellation = default, TimeSpan? timeout = default) + => _bus.StreamAsync(messages, cancellation, timeout); + + public Task StreamAsync(IAsyncEnumerable messages, + DeliveryOptions options, CancellationToken cancellation = default, TimeSpan? timeout = default) + => _bus.StreamAsync(messages, options, cancellation, timeout); }