Skip to content

Commit 5697d2f

Browse files
zZHorizonZzzZHorizonZz
committed
Refactor gRPC transport to simplify streaming protocol and remove max 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.
1 parent 437a6b9 commit 5697d2f

26 files changed

Lines changed: 111 additions & 162 deletions

File tree

‎vertx-grpc-common/src/main/java/io/vertx/grpc/common/MethodType.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,16 @@ public enum MethodType {
2525
*/
2626
BIDI;
2727

28+
/**
29+
* @return the method type for the given cardinality on each side
30+
*/
31+
public static MethodType of(boolean clientStreaming, boolean serverStreaming) {
32+
if (clientStreaming) {
33+
return serverStreaming ? BIDI : CLIENT_STREAMING;
34+
}
35+
return serverStreaming ? SERVER_STREAMING : UNARY;
36+
}
37+
2838
/**
2939
* @return whether the client sends a stream of requests
3040
*/

‎vertx-grpc-common/src/main/java/io/vertx/grpc/common/ServiceMethod.java‎

Lines changed: 34 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,14 @@
1919
public interface ServiceMethod<I, O> {
2020

2121
static <Req, Resp> ServiceMethod<Resp, Req> client(ServiceName serviceName, String methodName, GrpcMessageEncoder<Req> encoder, GrpcMessageDecoder<Resp> decoder) {
22-
return client(serviceName, methodName, MethodType.UNARY, encoder, decoder);
22+
return client(serviceName, methodName, null, encoder, decoder);
2323
}
2424

25-
static <Req, Resp> ServiceMethod<Resp, Req> client(ServiceName serviceName, String methodName, MethodType type, GrpcMessageEncoder<Req> encoder, GrpcMessageDecoder<Resp> decoder) {
25+
static <Req, Resp> ServiceMethod<Resp, Req> client(ServiceName serviceName, String methodName, boolean clientStreaming, boolean serverStreaming, GrpcMessageEncoder<Req> encoder, GrpcMessageDecoder<Resp> decoder) {
26+
return client(serviceName, methodName, MethodType.of(clientStreaming, serverStreaming), encoder, decoder);
27+
}
28+
29+
private static <Req, Resp> ServiceMethod<Resp, Req> client(ServiceName serviceName, String methodName, MethodType type, GrpcMessageEncoder<Req> encoder, GrpcMessageDecoder<Resp> decoder) {
2630
return new ServiceMethod<>() {
2731
@Override
2832
public ServiceName serviceName() {
@@ -33,8 +37,12 @@ public String methodName() {
3337
return methodName;
3438
}
3539
@Override
36-
public MethodType type() {
37-
return type;
40+
public boolean clientStreaming() {
41+
return type.clientStreaming();
42+
}
43+
@Override
44+
public boolean serverStreaming() {
45+
return type.serverStreaming();
3846
}
3947
@Override
4048
public GrpcMessageDecoder<Resp> decoder() {
@@ -48,10 +56,14 @@ public GrpcMessageEncoder<Req> encoder() {
4856
}
4957

5058
static <Req, Resp> ServiceMethod<Req, Resp> server(ServiceName serviceName, String methodName, GrpcMessageEncoder<Resp> encoder, GrpcMessageDecoder<Req> decoder) {
51-
return server(serviceName, methodName, MethodType.UNARY, encoder, decoder);
59+
return server(serviceName, methodName, null, encoder, decoder);
5260
}
5361

54-
static <Req, Resp> ServiceMethod<Req, Resp> server(ServiceName serviceName, String methodName, MethodType type, GrpcMessageEncoder<Resp> encoder, GrpcMessageDecoder<Req> decoder) {
62+
static <Req, Resp> ServiceMethod<Req, Resp> server(ServiceName serviceName, String methodName, boolean clientStreaming, boolean serverStreaming, GrpcMessageEncoder<Resp> encoder, GrpcMessageDecoder<Req> decoder) {
63+
return server(serviceName, methodName, MethodType.of(clientStreaming, serverStreaming), encoder, decoder);
64+
}
65+
66+
private static <Req, Resp> ServiceMethod<Req, Resp> server(ServiceName serviceName, String methodName, MethodType type, GrpcMessageEncoder<Resp> encoder, GrpcMessageDecoder<Req> decoder) {
5567
return new ServiceMethod<>() {
5668
@Override
5769
public ServiceName serviceName() {
@@ -62,8 +74,12 @@ public String methodName() {
6274
return methodName;
6375
}
6476
@Override
65-
public MethodType type() {
66-
return type;
77+
public boolean clientStreaming() {
78+
return type.clientStreaming();
79+
}
80+
@Override
81+
public boolean serverStreaming() {
82+
return type.serverStreaming();
6783
}
6884
@Override
6985
public GrpcMessageDecoder<Req> decoder() {
@@ -87,14 +103,17 @@ public GrpcMessageEncoder<Resp> encoder() {
87103
String methodName();
88104

89105
/**
90-
* Retrieves the cardinality of the gRPC service method.
91-
* By default, it will return {@code MethodType.UNARY}, indicating
92-
* a single request and a single response.
93-
*
94-
* @return the {@code MethodType} of the service method
106+
* @return whether the client side sends a stream of requests
107+
*/
108+
default boolean clientStreaming() {
109+
return false;
110+
}
111+
112+
/**
113+
* @return whether the server side sends a stream of responses
95114
*/
96-
default MethodType type() {
97-
return MethodType.UNARY;
115+
default boolean serverStreaming() {
116+
return false;
98117
}
99118

100119
/**

‎vertx-grpc-docs/src/main/java/examples/grpc/GreeterGrpcClient.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,8 @@ public interface GreeterGrpcClient extends GreeterClient {
2525
ServiceMethod<examples.grpc.HelloReply, examples.grpc.HelloRequest> SayHello = ServiceMethod.client(
2626
ServiceName.create("examples.grpc", "Greeter"),
2727
"SayHello",
28-
io.vertx.grpc.common.MethodType.UNARY,
28+
false,
29+
false,
2930
GrpcMessageEncoder.encoder(),
3031
GrpcMessageDecoder.decoder(examples.grpc.HelloReply.newBuilder()));
3132

‎vertx-grpc-docs/src/main/java/examples/grpc/StreamingGrpcClient.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,8 @@ public interface StreamingGrpcClient extends StreamingClient {
2525
ServiceMethod<examples.grpc.Item, examples.grpc.Empty> Source = ServiceMethod.client(
2626
ServiceName.create("examples.grpc", "Streaming"),
2727
"Source",
28-
io.vertx.grpc.common.MethodType.SERVER_STREAMING,
28+
false,
29+
true,
2930
GrpcMessageEncoder.encoder(),
3031
GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));
3132

@@ -35,7 +36,8 @@ public interface StreamingGrpcClient extends StreamingClient {
3536
ServiceMethod<examples.grpc.Empty, examples.grpc.Item> Sink = ServiceMethod.client(
3637
ServiceName.create("examples.grpc", "Streaming"),
3738
"Sink",
38-
io.vertx.grpc.common.MethodType.CLIENT_STREAMING,
39+
true,
40+
false,
3941
GrpcMessageEncoder.encoder(),
4042
GrpcMessageDecoder.decoder(examples.grpc.Empty.newBuilder()));
4143

@@ -45,7 +47,8 @@ public interface StreamingGrpcClient extends StreamingClient {
4547
ServiceMethod<examples.grpc.Item, examples.grpc.Item> Pipe = ServiceMethod.client(
4648
ServiceName.create("examples.grpc", "Streaming"),
4749
"Pipe",
48-
io.vertx.grpc.common.MethodType.BIDI,
50+
true,
51+
true,
4952
GrpcMessageEncoder.encoder(),
5053
GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));
5154

‎vertx-grpc-docs/src/main/java/examples/grpc/StreamingGrpcService.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -71,7 +71,8 @@ public static Service of(StreamingService service) {
7171
public static final ServiceMethod<examples.grpc.Empty, examples.grpc.Item> Source = ServiceMethod.server(
7272
SERVICE_NAME,
7373
"Source",
74-
io.vertx.grpc.common.MethodType.SERVER_STREAMING,
74+
false,
75+
true,
7576
GrpcMessageEncoder.encoder(),
7677
GrpcMessageDecoder.decoder(examples.grpc.Empty.newBuilder()));
7778

@@ -81,7 +82,8 @@ public static Service of(StreamingService service) {
8182
public static final ServiceMethod<examples.grpc.Item, examples.grpc.Empty> Sink = ServiceMethod.server(
8283
SERVICE_NAME,
8384
"Sink",
84-
io.vertx.grpc.common.MethodType.CLIENT_STREAMING,
85+
true,
86+
false,
8587
GrpcMessageEncoder.encoder(),
8688
GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));
8789

@@ -91,7 +93,8 @@ public static Service of(StreamingService service) {
9193
public static final ServiceMethod<examples.grpc.Item, examples.grpc.Item> Pipe = ServiceMethod.server(
9294
SERVICE_NAME,
9395
"Pipe",
94-
io.vertx.grpc.common.MethodType.BIDI,
96+
true,
97+
true,
9598
GrpcMessageEncoder.encoder(),
9699
GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));
97100

‎vertx-grpc-eventbus/src/main/generated/io/vertx/grpc/eventbus/EventBusGrpcClientOptionsConverter.java‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -12,11 +12,6 @@ public class EventBusGrpcClientOptionsConverter {
1212
static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, EventBusGrpcClientOptions obj) {
1313
for (java.util.Map.Entry<String, Object> member : json) {
1414
switch (member.getKey()) {
15-
case "wireFormat":
16-
if (member.getValue() instanceof String) {
17-
obj.setWireFormat(io.vertx.grpc.common.WireFormat.valueOf((String)member.getValue()));
18-
}
19-
break;
2015
}
2116
}
2217
}
@@ -26,8 +21,5 @@ static void toJson(EventBusGrpcClientOptions obj, JsonObject json) {
2621
}
2722

2823
static void toJson(EventBusGrpcClientOptions obj, java.util.Map<String, Object> json) {
29-
if (obj.getWireFormat() != null) {
30-
json.put("wireFormat", obj.getWireFormat().name());
31-
}
3224
}
3325
}

‎vertx-grpc-eventbus/src/main/generated/io/vertx/grpc/eventbus/EventBusGrpcServerOptionsConverter.java‎

Lines changed: 0 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -12,21 +12,6 @@ public class EventBusGrpcServerOptionsConverter {
1212
static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, EventBusGrpcServerOptions obj) {
1313
for (java.util.Map.Entry<String, Object> member : json) {
1414
switch (member.getKey()) {
15-
case "maxConcurrentStreams":
16-
if (member.getValue() instanceof Number) {
17-
obj.setMaxConcurrentStreams(((Number)member.getValue()).intValue());
18-
}
19-
break;
20-
case "supportedWireFormats":
21-
if (member.getValue() instanceof JsonArray) {
22-
java.util.LinkedHashSet<io.vertx.grpc.common.WireFormat> list = new java.util.LinkedHashSet<>();
23-
((Iterable<Object>)member.getValue()).forEach( item -> {
24-
if (item instanceof String)
25-
list.add(io.vertx.grpc.common.WireFormat.valueOf((String)item));
26-
});
27-
obj.setSupportedWireFormats(list);
28-
}
29-
break;
3015
}
3116
}
3217
}
@@ -36,11 +21,5 @@ static void toJson(EventBusGrpcServerOptions obj, JsonObject json) {
3621
}
3722

3823
static void toJson(EventBusGrpcServerOptions obj, java.util.Map<String, Object> json) {
39-
json.put("maxConcurrentStreams", obj.getMaxConcurrentStreams());
40-
if (obj.getSupportedWireFormats() != null) {
41-
JsonArray array = new JsonArray();
42-
obj.getSupportedWireFormats().forEach(item -> array.add(item.name()));
43-
json.put("supportedWireFormats", array);
44-
}
4524
}
4625
}

‎vertx-grpc-eventbus/src/main/java/io/vertx/grpc/eventbus/EventBusGrpcServerOptions.java‎

Lines changed: 7 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,9 @@
66
import io.vertx.core.json.JsonObject;
77
import io.vertx.grpc.common.WireFormat;
88

9+
import java.util.Arrays;
910
import java.util.Collections;
10-
import java.util.EnumSet;
11+
import java.util.LinkedHashSet;
1112
import java.util.Set;
1213

1314
@DataObject
@@ -16,32 +17,24 @@
1617
public class EventBusGrpcServerOptions {
1718

1819
/**
19-
* The default maximum number of streams multiplexed concurrently over the server's private address = {@code 1000}
20+
* The default set of wire formats the server accepts = {@code [proto, json]}
2021
*/
21-
public static final int DEFAULT_MAX_CONCURRENT_STREAMS = 1000;
22+
public static final Set<WireFormat> DEFAULT_SUPPORTED_WIRE_FORMATS = Collections.unmodifiableSet(new LinkedHashSet<>(Arrays.asList(WireFormat.PROTOBUF, WireFormat.JSON)));
2223

23-
/**
24-
* The default set of wire formats the server accepts = {@code [PROTOBUF, JSON]}
25-
*/
26-
public static final Set<WireFormat> DEFAULT_SUPPORTED_WIRE_FORMATS = Collections.unmodifiableSet(EnumSet.allOf(WireFormat.class));
27-
28-
private int maxConcurrentStreams;
2924
private Set<WireFormat> supportedWireFormats;
3025

3126
/**
3227
* Default options.
3328
*/
3429
public EventBusGrpcServerOptions() {
35-
maxConcurrentStreams = DEFAULT_MAX_CONCURRENT_STREAMS;
36-
supportedWireFormats = EnumSet.copyOf(DEFAULT_SUPPORTED_WIRE_FORMATS);
30+
supportedWireFormats = new LinkedHashSet<>(DEFAULT_SUPPORTED_WIRE_FORMATS);
3731
}
3832

3933
/**
4034
* Copy constructor.
4135
*/
4236
public EventBusGrpcServerOptions(EventBusGrpcServerOptions other) {
43-
maxConcurrentStreams = other.maxConcurrentStreams;
44-
supportedWireFormats = EnumSet.copyOf(other.supportedWireFormats);
37+
supportedWireFormats = new LinkedHashSet<>(other.supportedWireFormats);
4538
}
4639

4740
/**
@@ -52,29 +45,6 @@ public EventBusGrpcServerOptions(JsonObject json) {
5245
EventBusGrpcServerOptionsConverter.fromJson(json, this);
5346
}
5447

55-
/**
56-
* @return the maximum number of streams the server multiplexes concurrently over its private address
57-
*/
58-
public int getMaxConcurrentStreams() {
59-
return maxConcurrentStreams;
60-
}
61-
62-
/**
63-
* Set the maximum number of streams the server multiplexes concurrently over its private address. Opening requests
64-
* beyond this bound are rejected with {@code RESOURCE_EXHAUSTED}, so a flood of opens cannot grow the demux map
65-
* without limit.
66-
*
67-
* @param maxConcurrentStreams the maximum number of concurrent streams, must be {@code > 0}
68-
* @return a reference to this, so the API can be used fluently
69-
*/
70-
public EventBusGrpcServerOptions setMaxConcurrentStreams(int maxConcurrentStreams) {
71-
if (maxConcurrentStreams <= 0) {
72-
throw new IllegalArgumentException("maxConcurrentStreams must be > 0");
73-
}
74-
this.maxConcurrentStreams = maxConcurrentStreams;
75-
return this;
76-
}
77-
7848
/**
7949
* @return the set of wire formats the server accepts
8050
*/
@@ -101,7 +71,7 @@ public EventBusGrpcServerOptions setSupportedWireFormats(Set<WireFormat> support
10171
if (supportedWireFormats == null || supportedWireFormats.isEmpty()) {
10272
throw new IllegalArgumentException("supportedWireFormats must not be empty");
10373
}
104-
this.supportedWireFormats = EnumSet.copyOf(supportedWireFormats);
74+
this.supportedWireFormats = new LinkedHashSet<>(supportedWireFormats);
10575
return this;
10676
}
10777

‎vertx-grpc-eventbus/src/main/java/io/vertx/grpc/eventbus/impl/EventBusGrpcClientImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ public <Req, Resp> Future<GrpcClientRequest<Req, Resp>> invoker(ServiceMethod<Re
3131

3232
@Override
3333
public <Req, Resp> Future<GrpcClientRequest<Req, Resp>> request(ServiceMethod<Resp, Req> method) {
34-
EventBusGrpcClientInvoker invoker = new EventBusGrpcClientInvoker(context(), this, method.type());
34+
EventBusGrpcClientInvoker invoker = new EventBusGrpcClientInvoker(context(), this, method.clientStreaming() || method.serverStreaming());
3535
GrpcClientRequestImpl<Req, Resp> request = new GrpcClientRequestImpl<>(
3636
context(),
3737
invoker,

‎vertx-grpc-eventbus/src/main/java/io/vertx/grpc/eventbus/impl/EventBusGrpcClientInvoker.java‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,25 +2,24 @@
22

33
import io.vertx.core.internal.ContextInternal;
44
import io.vertx.grpc.client.impl.GrpcClientInvoker;
5-
import io.vertx.grpc.common.MethodType;
65
import io.vertx.grpc.common.ServiceName;
76
import io.vertx.grpc.common.impl.GrpcStream;
87

98
public class EventBusGrpcClientInvoker implements GrpcClientInvoker {
109

1110
private final ContextInternal context;
1211
private final EventBusGrpcClientImpl client;
13-
private final MethodType type;
12+
private final boolean streaming;
1413

15-
public EventBusGrpcClientInvoker(ContextInternal context, EventBusGrpcClientImpl client, MethodType type) {
14+
public EventBusGrpcClientInvoker(ContextInternal context, EventBusGrpcClientImpl client, boolean streaming) {
1615
this.context = context;
1716
this.client = client;
18-
this.type = type;
17+
this.streaming = streaming;
1918
}
2019

2120
@Override
2221
public GrpcStream invoke(ServiceName serviceName, String methodName) {
23-
if (type.streaming()) {
22+
if (streaming) {
2423
return new EventBusGrpcClientStreamingCall(context, client, serviceName, methodName);
2524
}
2625
return new EventBusGrpcClientUnaryCall(context, client.eventBus(), serviceName, methodName);

0 commit comments

Comments
 (0)