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

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

Motivation:

Add experimental gRPC streaming over the event bus

vietj · 2026-06-23T07:12:08Z

+      EventBusHeaders.encodeMultiMap(HEADER_PREFIX, requestHeaders, options.getHeaders());
+    }
+
+    long clientStreamId = endpoint.register(this);


we should split registration and stream ID generation in two steps, so we register only when the interaction on the following EventBus request is successful

we should have a monotonic stream ID sequence that is allocated when the stream is created and sent when the result of the event-bus interaction is received

vietj · 2026-06-23T07:16:09Z

+    HalfClose half_close = 7; // client to server, end of the request stream
+    Trailers trailers = 8; // server to client, terminates the call
+    Cancel cancel = 9; // either direction, abnormal termination
+    Headers headers = 10; // server to client, response metadata, before the first message


I think we don't need this frame because we can make this part of the initial request/response interaction along with the initialized response in clear format (like for unary)

vietj · 2026-06-23T07:16:57Z

+  uint64 stream_sequence = 2; // per-stream, monotonic, advances on Message frames only
+
+  oneof frame {
+    Initialize initialize = 3; // client to server, the opening request body


I don't think we need initialize/initialized frames here, we should keep protobuf frames only for the private address communication

vietj · 2026-06-23T07:20:56Z

+    return id;
+  }
+
+  void closed(FrameHandler stream) {


are we testing that an an initial or streaming failure correctly remove the stream from the map ?

vietj · 2026-06-29T07:34:48Z

+import io.vertx.core.eventbus.Message;
+import io.vertx.grpc.eventbus.transport.v1alpha.TransportFrame;
+
+interface FrameHandler extends Closeable {


for the moment we don't need that interface, we should just make in an abstract method of EventBusStreamEndpoint

vietj · 2026-06-29T07:37:24Z

+abstract class EventBusStreamEndpoint {
+
+  protected final ConcurrentMap<Long, FrameHandler> streams = new ConcurrentHashMap<>();
+  protected volatile int maxConcurrentStreams = EventBusGrpcServerOptions.DEFAULT_MAX_CONCURRENT_STREAMS;


it should be removed.

I don't think we can support this because there is no way to identify the remote side we will handle when using the event-bus due to load balancing, the main purpose of max concurrent streams is to protect the producer and when using load balancing we just use an event-bus address.

vietj · 2026-06-29T12:06:26Z

 public interface ServiceMethod<I, O> {

  static <Req, Resp> ServiceMethod<Resp, Req> client(ServiceName serviceName, String methodName, GrpcMessageEncoder<Req> encoder, GrpcMessageDecoder<Resp> decoder) {
+    return client(serviceName, methodName, MethodType.UNARY, encoder, decoder);


we should set null instead of MethodType.UNARY instead because this is not available and unknown. I'd prefer a bug with an NPE rather than a misbehavior.

vietj · 2026-06-29T12:07:39Z

+    return client(serviceName, methodName, MethodType.UNARY, encoder, decoder);
+  }
+
+  static <Req, Resp> ServiceMethod<Resp, Req> client(ServiceName serviceName, String methodName, MethodType type, GrpcMessageEncoder<Req> encoder, GrpcMessageDecoder<Resp> decoder) {


instead of this we should instead have booleain indicating cardinality for both sides of the service and avoid expose MethodType at the moment, it should remain internal.

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.

vietj · 2026-07-15T08:53:02Z

+/**
+ * The cardinality of a gRPC service method.
+ */
+public enum MethodType {


I think for now I would like to avoid introducing this type as it is only used in this module and instead stick with the two booleans used, probably those booleans should be Boolean to indicate when the information is unknown

vietj · 2026-07-15T09:34:47Z

+  private final EventBus eventBus;
+  private final ServiceName serviceName;
+  private final String methodName;
+  private final Deque<GrpcMessage> pending = new ArrayDeque<>();


init that in the constructor along with other inits

vietj · 2026-07-15T09:34:53Z

+  private Duration timeout;
+
+  private boolean ended;
+  private State state = State.IDLE;


init that in the constructor along with other inits

vietj · 2026-07-15T09:47:36Z

+    }
+  }
+
+  protected void writeMessage(GrpcMessage message) {


We need to add message write future return with the following modification:

the EventBusGrpcStreamBase#writeMessage(GrpcMessage) should return a Future<Void>, to implement this:

the EventBusGrpcStreamBase#outboundQueue should be modified to contain an object that has a Promise in addition of the message itself, e.g. a MessageWrite internal class

the method EventBusGrpcStreamBase#sendTransportFrame should be modified to return a Future<Void>

The implementation of sendTransportFrame should use a MessageProducer instead of the event bus directly, MessageProducer#write will return a Future<Void>, this producer for sending can be obtained with EventBus#sender(String address), before writing the message delivery options should be set on the producer (and then set to null). I think this is OK but not convenient, so I think we should overload MessageProducer#write(T) with MessageProducer#write(T, DeliveryOptions) in vertx-core at some point.

…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.

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 added 9 commits June 30, 2026 16:16

Add experimental gRPC streaming over the event bus

3a0d364

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

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

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Add experimental gRPC streaming over the event bus#281

Add experimental gRPC streaming over the event bus#281
zZHorizonZz wants to merge 11 commits into
eclipse-vertx:mainfrom
zZHorizonZz:grpc-eventbus-streaming

zZHorizonZz commented Jun 9, 2026

Uh oh!

vietj Jun 23, 2026

Uh oh!

vietj Jun 23, 2026

Uh oh!

vietj Jun 23, 2026

Uh oh!

vietj Jun 23, 2026

Uh oh!

vietj Jun 29, 2026

Uh oh!

vietj Jun 29, 2026

Uh oh!

vietj Jun 29, 2026

Uh oh!

vietj Jun 29, 2026

Uh oh!

vietj Jul 15, 2026

Uh oh!

vietj Jul 15, 2026

Uh oh!

vietj Jul 15, 2026

Uh oh!

vietj Jul 15, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

zZHorizonZz commented Jun 9, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants