Add gRPC streaming over the event bus by zZHorizonZz · Pull Request #281 · eclipse-vertx/vertx-grpc

zZHorizonZz · 2026-06-09T14:40:27Z

Motivation:

Add gRPC streaming over the event bus

Signed-off-by: Daniel Fiala <danfiala23@gmail.com>

…server streaming call initialization. Motivation: - Transition to `v1alpha` schema for EventBus transport messages for better compatibility and clarity. - Simplify and improve the initialization of server-side streaming calls with a more modular design. Changes: - Updated protobuf definitions and transport schema from `eventbus.transport` to `eventbus.transport.v1alpha`. - Refactored `EventBusGrpcServerStreamingCall` to use a dedicated `init` method for inbound message consumer setup. - Simplified server-side streaming call logic by removing `AtomicReference` and directly managing completion callbacks. - Updated documentation to specify the new schema path and updated associated examples.

…ams, update transport schema to `Initialize`/`Initialized` handshake. Motivation: - Transition from per-call addresses to multiplexed streams over shared private addresses, reducing registration churn and improving scalability. - Replace `Ack` with `Initialize`/`Initialized` handshake for clearer and more efficient stream initialization. - Enhance stream management with session-level abstractions. Changes: - Introduced `FrameHandler` interface to handle multiplexed streams. - Updated transport schema to use `Initialize`/`Initialized` messages. - Refactored `EventBusGrpcClientStreamingCall` and `EventBusGrpcServerStreamingCall` for improved stream lifecycle management. - Updated documentation and protocol definitions (`eventbus.transport.proto`) to reflect the new design. - Improved error handling and stream termination logic.

…ed async handling. Motivation: - Enhance the usability and consistency of the EventBus gRPC API by having client and server initialization methods return `Future` instead of direct instances. - Ensure proper integration with Vert.x's asynchronous model. Changes: - Updated `EventBusGrpcClient.client` and `EventBusGrpcServer.server` methods to return `Future` instances. - Refactored related tests and examples to use the updated async API. - Improved error handling and initialization completion by leveraging `Future` chaining. - Adjusted internal implementations to remove lazy registration in favor of explicit `bind()` calls.

…rame schema. Motivation: - Replace protobuf `Initialize`/`Initialized` handshake with explicit headers for reduced complexity and improved protocol clarity. - Simplify frame schema by removing unused message types and focusing on active bidirectional streaming fields. Changes: - Removed `Initialize`/`Initialized` protobuf messages from transport schema. - Refactored `EventBusGrpcServer` and `EventBusGrpcClient` to utilize headers for streaming handshake and session demultiplexing. - Introduced `StreamRegistration` abstraction for improved stream lifecycle management. - Updated tests to include new error handling scenarios for handshake validation. - Adjusted documentation to reflect the simplified method-level streaming protocol.

…pport max concurrent streams cap. Motivation: - Introduce configurable server options to improve flexibility and manageability of the gRPC server. - Add a cap on maximum concurrent streams to handle resource exhaustion effectively. Changes: - Added `EventBusGrpcServerOptions` to manage server configurations like `maxConcurrentStreams`. - Updated gRPC server methods to accept `EventBusGrpcServerOptions` for initialization. - Enhanced error handling to reject excessive stream openings with `RESOURCE_EXHAUSTED`. - Added tests for configurable options and error handling scenarios. - Refactored related implementation to handle the new options efficiently.

… server. Motivation: - Allow gRPC client and server to support multiple wire formats (e.g., PROTOBUF, JSON) for improved flexibility and interoperability. - Enable validation and rejection of unsupported wire formats at runtime. Changes: - Added `EventBusGrpcServerOptions` and `EventBusGrpcClientOptions` to configure supported wire formats. - Updated server to validate incoming wire formats against configured options and reject unsupported formats with `UNIMPLEMENTED` status. - Enhanced frame encoding/decoding logic to handle dynamic wire formats. - Introduced new helper methods for setting and validating wire formats in client and server implementations. - Added unit tests to verify correct handling of wire format configurations and error scenarios.

…tocol details. Motivation: - Improve documentation clarity by simplifying explanations and updating terminology to reflect current design. - Align descriptions with recent refactors and schema changes for better readability and understanding. Changes: - Simplified explanation of gRPC stream handling via the event bus, including handshake and streaming mechanisms. - Updated sections to match the latest transport schema and protocol design. - Reorganized content for better flow and removed redundant details. - Added examples and references for improved context.

… concurrent streams cap. Motivation: - Simplify the gRPC stream initialization protocol by removing unnecessary complexities in frame schema and handshake. - Eliminate the max concurrent streams cap for better scalability and to streamline server options. Changes: - Removed `maxConcurrentStreams` functionality and related validations from `EventBusGrpcServerOptions`. - Simplified stream initialization logic by replacing `MethodType` with explicit client/server streaming flags. - Refactored `ServiceMethod` to use boolean flags (`clientStreaming`, `serverStreaming`) instead of method type enums. - Updated test cases and documentation to reflect schema and initialization changes. - Consolidated stream management interfaces by removing `FrameHandler` and merging its functionality into `EventBusGrpcStreamBase`. - Updated generated code and templates to align with the simplified schema.

…ient/server streaming flags. Motivation: - Simplify gRPC method definitions by replacing the `MethodType` enum with boolean flags for `clientStreaming` and `serverStreaming`. - Align with recent changes in transport schema and improve readability. Changes: - Removed `MethodType` and replaced its usage with `Boolean clientStreaming` and `Boolean serverStreaming` in `ServiceMethod` and related interfaces. - Updated `TranscodingServiceMethod` and `TranscodingInvoker` implementations to reflect the new flags. - Modified template files, generated code, and test cases to align with updated method definitions. - Added a new example for server streaming with transcoding options.

…ect instantiation consistency. Motivation: - Ensure proper initialization of `pending` and `state` fields during object construction to avoid potential null pointer exceptions and improve code clarity. Changes: - Move `pending` and `state` field initializations from declarations to the constructor. - Removed default initializations from field declarations.

…re-aware message writes and improve async handling. Motivation: - Enhance backpressure handling by queuing writes and completing them asynchronously when the transport is ready. - Standardize the use of `Future` for transport frame writing across client and server implementations. - Improve stream initialization consistency by consolidating producer creation logic. Changes: - Added `MessageWrite` to encapsulate queued messages and associated promises. - Replaced synchronous write methods with `Future`-based methods for transport frame and message writes. - Updated server and client streaming calls to utilize a shared `MessageProducer` for sending transport frames. - Added test for queued writes completing upon stream readiness, validating backpressure handling. - Refactored dequeue logic for outbound messages, ensuring proper promise completion on transport frame sending.

The buffering added along with testConcurrentStreamsAllComplete does not belong in this branch: it changes a shared class for every transport in order to work around a problem that is not specific to the event bus. Both are reverted here. The problem it addressed is real. A response received before the application attaches its handlers is dropped, and the call then hangs rather than failing, which is what made testMultiplexAcrossClients time out intermittently on CI. It is reproducible on main over HTTP/2 with no event bus involved, so it belongs in vertx-grpc-common with its own reproducer rather than here. The buffering was also only a partial remedy: it keeps the messages but still releases the end on top of them, so a consumer that attaches its end handler before its message handler sees the end first.

- Rename MessageWrite#windowed to flowControlled, which says what the flag decides rather than how the window is implemented. - Express pingInterval, pingTimeout and maxPingInterval as Duration rather than long milliseconds. The impls convert at the boundary, since the endpoint works in milliseconds internally, and the interval advertised on the handshake stays milliseconds on the wire. - Drop the JSON plumbing from both options classes: the JsonGen annotation, the JsonObject constructor, toJson and toString, along with the two generated converters. GrpcClientOptions is a DataObject without any of it already.

…ke sequence Motivation: - Clarify gRPC event bus streaming behavior, including handshake rules, liveness probes, and message flow. - Provide detailed explanations for new mechanics and headers like `grpc-ping-interval` and `grpc-peer-address`. - Improve developer understanding of frame handling and liveness options. Changes: - Updated README with explanations about handshake rules and flow control mechanisms. - Added details about `grpc-ping-interval`, `grpc-peer-address`, and liveness probe management. - Revised `TransportFrame` comments to describe streaming behavior and wire formats. - Simplified redundant comments and improved clarity for `Ping` and codec-related behavior. - Enhanced liveness sections to include timeout conditions and server-side interval validation.

…iveness mechanics Motivation: - The term `pingInterval` was misleading as it described a stream's timeout duration instead of the interval between pings. - Align naming with the behavior of the timeout mechanism to improve developer understanding and consistency. Changes: - Renamed `pingInterval` to `pingTimeout` across all gRPC event bus configurations and implementations. - Updated related documentation, comments, and tests to reflect the rename. - Adjusted server-side options to use `maxPingTimeout` instead of `maxPingInterval` for clearer meaning. - Refactored peer timeout calculations to prioritize `pingTimeout` values. - Maintained backward-compatible validation for server `maxPingTimeout` configurations.

…oved clarity and consistency Motivation: - Replace ambiguous term `Peer` with `RemoteEndpoint` to provide better semantic meaning and align with gRPC terminology. - Enhance code readability and reduce confusion regarding naming of variables, methods, and objects. Changes: - Renamed `Peer` to `RemoteEndpoint` across all relevant classes, methods, and variables. - Updated method names such as `handlePeerDown` to `handleRemoteEndpointDown` to reflect the terminology change. - Refactored `EventBusStreamEndpoint` to `EventBusGrpcEndpoint` for better alignment with gRPC-based functionality. - Updated documentation, comments, and README to reflect new terminology. - Ensured tests and related logic were updated to support the new naming convention.

vietj

Minor changes :-)

…s and remove redundant event bus parameter Motivation: - Simplify the API by removing the unnecessary `EventBus` parameter from client and server constructors. - Enhance usability by relying on the event bus instance already accessible through `Vertx`. - Reduce duplication and improve code clarity across gRPC implementations. Changes: - Refactored `EventBusGrpcClientImpl` and `EventBusGrpcServerImpl` to construct clients and servers using `Vertx` and options only. - Removed overloaded methods that explicitly required an event bus instance. - Updated documentation comments in `EventBusGrpcServerOptions` and `EventBusGrpcClientOptions` to clarify configurations. - Cleaned up redundant imports and adjusted method signatures in affected classes.

zZHorizonZz force-pushed the grpc-eventbus-streaming branch 2 times, most recently from ad3f1ff to 782c880 Compare June 9, 2026 14:43

zZHorizonZz requested a review from vietj June 16, 2026 05:54

vietj added this to the 5.2.0 milestone Jun 17, 2026

vietj added the enhancement New feature or request label Jun 17, 2026

zZHorizonZz force-pushed the grpc-eventbus-streaming branch from 782c880 to 06fce04 Compare June 17, 2026 16:41

vietj marked this pull request as ready for review June 23, 2026 07:24

vietj requested changes Jun 23, 2026

View reviewed changes

vietj requested changes Jun 29, 2026

View reviewed changes

zZHorizonZz force-pushed the grpc-eventbus-streaming branch from cbf9ee9 to 5697d2f Compare June 30, 2026 14:31

zZHorizonZz mentioned this pull request Jun 30, 2026

Support server-streaming in transcoding #284

Draft

zZHorizonZz requested a review from vietj July 2, 2026 07:18

vietj reviewed Jul 15, 2026

View reviewed changes

Comment thread vertx-grpc-common/src/main/java/io/vertx/grpc/common/MethodType.java Outdated

vietj reviewed Jul 15, 2026

View reviewed changes

Comment thread ...grpc-eventbus/src/main/java/io/vertx/grpc/eventbus/impl/EventBusGrpcClientStreamingCall.java Outdated

vietj reviewed Jul 15, 2026

View reviewed changes

Comment thread ...grpc-eventbus/src/main/java/io/vertx/grpc/eventbus/impl/EventBusGrpcClientStreamingCall.java Outdated

vietj reviewed Jul 15, 2026

View reviewed changes

Comment thread vertx-grpc-eventbus/src/main/java/io/vertx/grpc/eventbus/impl/EventBusGrpcStreamBase.java Outdated

vietj requested changes Jul 15, 2026

View reviewed changes

zZHorizonZz force-pushed the grpc-eventbus-streaming branch from 8e25a27 to 62edcd5 Compare July 15, 2026 16:22

zZHorizonZz added 12 commits July 20, 2026 17:02

Add experimental gRPC streaming over the event bus

3a92e33

Signed-off-by: Daniel Fiala <danfiala23@gmail.com>