eclipse-vertx / vertx-grpc Goto Github PK
View Code? Open in Web Editor NEWDevelopment of the gRPC component for Eclipse Vert.x
License: Eclipse Public License 2.0
Development of the gRPC component for Eclipse Vert.x
License: Eclipse Public License 2.0
vertx-grpc documentation page is showing a rendering error for the new vertx-grpc-context-storage module
Last line on the page https://vertx.io/docs/vertx-grpc/java/
Unresolved directive in index.adoc - include::storage.adoc[]
4.3.5
Encountering exception when querying grpc reflection service
Testing with grpcurl:
grpcurl -plaintext localhost:8080list
Exception:
15:01:35.948 [vert.x-eventloop-thread-2] ERROR io.vertx.core.impl.ContextBase - Unhandled exception java.lang.NullPointerException: Cannot invoke "io.grpc.Server.getImmutableServices()" because "server" is null at io.grpc.protobuf.services.ProtoReflectionService.getRefreshedIndex(ProtoReflectionService.java:93) at io.grpc.protobuf.services.ProtoReflectionService.serverReflectionInfo(ProtoReflectionService.java:134) at io.grpc.reflection.v1alpha.ServerReflectionGrpc$MethodHandlers.invoke(ServerReflectionGrpc.java:208) at io.grpc.stub.ServerCalls$StreamingServerCallHandler.startCall(ServerCalls.java:235) at io.vertx.grpc.server.impl.GrpcServiceBridgeImpl.lambda$bind$2(GrpcServiceBridgeImpl.java:62) at io.vertx.grpc.server.impl.GrpcServerImpl$MethodCallHandler.handle(GrpcServerImpl.java:95) at io.vertx.grpc.server.impl.GrpcServerImpl.handle(GrpcServerImpl.java:62)
val grpcServer: GrpcServer = GrpcServer.server(vertx)
val mainBridge = GrpcServiceBridge.bridge(MainBridge())
val reflectionBridge = GrpcServiceBridge.bridge(ProtoReflectionService.newInstance())
mainBridge.bind(grpcServer)
reflectionBridge.bind(grpcServer)
httpServer = vertx
.createHttpServer(httpOptions)
.requestHandler(grpcServer)
//.requestHandler(router)
.listen(appConfig.webPort).await()
See VertxClientCall class.
To reproduce
public void testNewSmoke() throws Exception {
Vertx vertx = vertx();
try {
HttpClientOptions options = new HttpClientOptions();
Buffer buffer;
try (InputStream stream = getClass().getClassLoader().getResourceAsStream("tls/ca.pem")) {
buffer = Buffer.buffer(stream.readAllBytes());
}
options.setTrustOptions(new PemTrustOptions().addCertValue(buffer));
GrpcClient client = GrpcClient.client(vertx, options);
try {
GrpcClientChannel channel = new GrpcClientChannel(client, SocketAddress.inetSocketAddress(port(), "localhost"));
GreeterGrpc.GreeterBlockingStub stub = GreeterGrpc.newBlockingStub(channel);
HelloReply reply = stub.sayHello(HelloRequest.newBuilder().setName("neo-blocking").build());
We need a client builder API to let user configure load balancing / discovery in a friendly fashion.
Define threading model and expectations.
Using this client code:
client
.request(server, GreeterGrpc.getSayHelloMethod()).compose(request -> {
request.end(HelloRequest
.newBuilder()
.setName("Bob")
.build());
return request.response().compose(response -> response.last());
}).onSuccess(reply -> {
System.out.println("Received " + reply.getMessage());
});
I have observed that RST_STREAM (8) is being sent after each request. However, my understanding of the RST_STREAM frame is that it should only be used when there is an error that prevents the sender or receiver from completing a particular stream.
To confirm my understanding, I also used the older VertxGrpcStub and found no instances of RST_STREAM being used.
I attempted to change request.end()
to request.write()
but this did not resolve the issue. Is there a way to send a request without including the RST_STREAM frame?
To gain further insights, I also performed a debug session using this unit test to trace the code that produces the RST_STREAM frame. I think it probably comes from here.
gRPC interceptors are useful for platform code like logging, monitoring or tracing.
The futurized stubs provided by the Vert.x protoc plugin cannot be integrated with gRPC Java interceptor code because they don't implement BindableService
.
Users should be able to write interceptors for their futurized gRPC services.
4.3.7
The GrpcClient constructors allow you to pass in custom HttpOptions.
Because httpOptions.setHttp2ClearTextUpgrade(false)
must be set, the constructor should do it for you, like the default constructor does.
GrpcClient.client(vertx, new HttpClientOptions())
.GrpcClient.client(vertx, new HttpClientOptions().setHttp2ClearTextUpgrade(false))
resolves the issue.This is a known issue. It is never valid to have this set to true
(or the default) on a GrpcClient.
Headers/trailers are not always correctly handled. gRPC specific metadata keys (beginning with -grpc) are sometimes skipped and binary keys (ending with -bin) are not properly handled.
When I using GrpcServiceBridgeImpl
, some of the behavior of vertx-grpc is not as expected when the network is abnormal.
if the network between client and server was interrupted, listener.onCancel()
should be invoked.
nothing happened.
4.5.7
The errorHandler
is an method from GrpcReadStream
, it was invoked only when the httpStream has a StreamResetException
, but when client was killed, an IOException
occurred.
java.io.IOException: Broken pipe
at java.base/sun.nio.ch.SocketDispatcher.writev0(Native Method)
at java.base/sun.nio.ch.SocketDispatcher.writev(SocketDispatcher.java:66)
at java.base/sun.nio.ch.IOUtil.write(IOUtil.java:227)
at java.base/sun.nio.ch.IOUtil.write(IOUtil.java:158)
at java.base/sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:574)
at io.netty.channel.socket.nio.NioSocketChannel.doWrite(NioSocketChannel.java:430)
at io.netty.channel.AbstractChannel$AbstractUnsafe.flush0(AbstractChannel.java:931)
at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.flush0(AbstractNioChannel.java:359)
at io.netty.channel.AbstractChannel$AbstractUnsafe.flush(AbstractChannel.java:895)
at io.netty.channel.DefaultChannelPipeline$HeadContext.flush(DefaultChannelPipeline.java:1372)
at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:935)
at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:921)
at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:907)
at io.netty.handler.codec.http2.Http2ConnectionHandler.flush(Http2ConnectionHandler.java:197)
at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:941)
at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:921)
at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:907)
at io.netty.channel.DefaultChannelPipeline.flush(DefaultChannelPipeline.java:967)
at io.netty.channel.AbstractChannel.flush(AbstractChannel.java:254)
at io.vertx.core.http.impl.VertxHttp2ConnectionHandler.checkFlush(VertxHttp2ConnectionHandler.java:247)
at io.vertx.core.http.impl.VertxHttp2ConnectionHandler.writeData(VertxHttp2ConnectionHandler.java:242)
at io.vertx.core.http.impl.VertxHttp2Stream.doWriteData(VertxHttp2Stream.java:244)
at io.vertx.core.http.impl.VertxHttp2Stream.writeData(VertxHttp2Stream.java:216)
at io.vertx.core.http.impl.Http2ServerResponse.write(Http2ServerResponse.java:469)
at io.vertx.core.http.impl.Http2ServerResponse.write(Http2ServerResponse.java:347)
at io.vertx.core.http.impl.Http2ServerResponse.write(Http2ServerResponse.java:48)
at io.vertx.grpc.server.impl.GrpcServerResponseImpl.writeMessage(GrpcServerResponseImpl.java:242)
at io.vertx.grpc.server.impl.GrpcServerResponseImpl.writeMessage(GrpcServerResponseImpl.java:112)
at io.vertx.grpc.server.impl.GrpcServerResponseImpl.write(GrpcServerResponseImpl.java:102)
at io.vertx.grpc.server.impl.GrpcServerResponseImpl.write(GrpcServerResponseImpl.java:248)
at io.vertx.core.streams.impl.PipeImpl.lambda$to$1(PipeImpl.java:81)
at io.vertx.grpc.server.impl.GrpcServerRequestImpl.lambda$handler$0(GrpcServerRequestImpl.java:79)
at io.vertx.grpc.common.impl.GrpcReadStreamBase.handleMessage(GrpcReadStreamBase.java:207)
at io.vertx.grpc.common.impl.GrpcReadStreamBase.lambda$init$3(GrpcReadStreamBase.java:89)
at io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:255)
at io.vertx.core.streams.impl.InboundBuffer.write(InboundBuffer.java:134)
at io.vertx.grpc.common.impl.GrpcReadStreamBase.handle(GrpcReadStreamBase.java:125)
at io.vertx.grpc.common.impl.GrpcReadStreamBase.handle(GrpcReadStreamBase.java:39)
at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:279)
at io.vertx.core.http.impl.HttpEventHandler.handleChunk(HttpEventHandler.java:51)
at io.vertx.core.http.impl.Http2ServerRequest.handleData(Http2ServerRequest.java:148)
at io.vertx.core.http.impl.Http2ServerStream.handleData(Http2ServerStream.java:206)
at io.vertx.core.http.impl.VertxHttp2Stream.lambda$new$1(VertxHttp2Stream.java:75)
at io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:255)
at io.vertx.core.streams.impl.InboundBuffer.drain(InboundBuffer.java:242)
at io.vertx.core.streams.impl.InboundBuffer.lambda$fetch$0(InboundBuffer.java:295)
at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:279)
at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:261)
at io.vertx.core.impl.ContextInternal.lambda$runOnContext$0(ContextInternal.java:59)
at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:1583)
how to get remoteAddress from io.vertx.grpc.server.GrpcServerRequest,i can`t find any method to get remoteAddress
GrpcServer grpcServer = GrpcServer.server(vertx);
grpcServer.callHandler(StreamingGrpc.getPipeMethod(), request -> {
// how to get remoteAddress from request
});
if use socket,i can use this method
io.vertx.core.http.WebSocketBase#remoteAddress
4.3.7
Add support for withCallCredentials
on gRPC stubs.
SomeStub.newStub(new GrpcClientChannel(...)).withCallCredentials
does not propagate call credentials.
It's both.
Caught this in CI:
SEVERE: LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records:
Created at:
io.netty.buffer.UnpooledByteBufAllocator.newDirectBuffer(UnpooledByteBufAllocator.java:96)
io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:188)
io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:116)
io.netty.handler.codec.base64.Base64.encode(Base64.java:108)
io.netty.handler.codec.base64.Base64.encode(Base64.java:100)
io.netty.handler.codec.base64.Base64.encode(Base64.java:80)
io.netty.handler.codec.base64.Base64.encode(Base64.java:74)
io.vertx.grpc.server.impl.GrpcServerResponseImpl.encodeMessage(GrpcServerResponseImpl.java:274)
io.vertx.grpc.server.impl.GrpcServerResponseImpl.writeMessage(GrpcServerResponseImpl.java:267)
io.vertx.grpc.server.impl.GrpcServerResponseImpl.writeMessage(GrpcServerResponseImpl.java:123)
io.vertx.grpc.common.impl.WriteStreamAdapter.write(WriteStreamAdapter.java:47)
io.vertx.grpc.server.impl.GrpcServiceBridgeImpl$ServerCallImpl.sendMessage(GrpcServiceBridgeImpl.java:169)
io.grpc.ForwardingServerCall.sendMessage(ForwardingServerCall.java:32)
io.grpc.stub.ServerCalls$ServerCallStreamObserverImpl.onNext(ServerCalls.java:380)
io.vertx.grpc.server.web.TestServiceImpl.streamingCall(TestServiceImpl.java:58)
io.vertx.grpcweb.TestServiceGrpc$MethodHandlers.invoke(TestServiceGrpc.java:335)
io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:182)
io.vertx.grpc.server.impl.GrpcServiceBridgeImpl$ServerCallImpl$1.handleClose(GrpcServiceBridgeImpl.java:100)
io.vertx.grpc.common.impl.ReadStreamAdapter.lambda$init$1(ReadStreamAdapter.java:33)
io.vertx.grpc.common.impl.GrpcReadStreamBase.handleEnd(GrpcReadStreamBase.java:195)
io.vertx.grpc.common.impl.GrpcReadStreamBase.lambda$init$3(GrpcReadStreamBase.java:83)
io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:255)
io.vertx.core.streams.impl.InboundBuffer.write(InboundBuffer.java:134)
io.vertx.grpc.common.impl.GrpcReadStreamBase.lambda$init$0(GrpcReadStreamBase.java:72)
io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:237)
io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:219)
io.vertx.core.http.impl.HttpEventHandler.handleEnd(HttpEventHandler.java:76)
io.vertx.core.http.impl.Http1xServerRequest.onEnd(Http1xServerRequest.java:541)
io.vertx.core.http.impl.Http1xServerRequest$1.handleMessage(Http1xServerRequest.java:104)
io.vertx.core.net.impl.InboundMessageQueue.test(InboundMessageQueue.java:73)
io.vertx.core.streams.impl.InboundReadQueue.drain(InboundReadQueue.java:250)
io.vertx.core.streams.impl.InboundReadQueue.drain(InboundReadQueue.java:224)
io.vertx.core.net.impl.InboundMessageQueue.drainInternal(InboundMessageQueue.java:164)
io.vertx.core.net.impl.InboundMessageQueue.drain(InboundMessageQueue.java:144)
io.vertx.core.http.impl.Http1xServerRequest.handleEnd(Http1xServerRequest.java:153)
io.vertx.core.http.impl.Http1xServerConnection.onEnd(Http1xServerConnection.java:197)
io.vertx.core.http.impl.Http1xServerConnection.onContent(Http1xServerConnection.java:188)
io.vertx.core.http.impl.Http1xServerConnection.handleOther(Http1xServerConnection.java:173)
io.vertx.core.http.impl.Http1xServerConnection.handleMessage(Http1xServerConnection.java:166)
io.vertx.core.net.impl.VertxConnection.read(VertxConnection.java:239)
io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:153)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:93)
io.netty.handler.codec.http.websocketx.extensions.WebSocketServerExtensionHandler.channelRead(WebSocketServerExtensionHandler.java:87)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:318)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
io.vertx.core.http.impl.Http1xOrH2CHandler.end(Http1xOrH2CHandler.java:61)
io.vertx.core.http.impl.Http1xOrH2CHandler.channelRead(Http1xOrH2CHandler.java:38)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1407)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:918)
io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:994)
io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
java.base/java.lang.Thread.run(Thread.java:829)
The Message level API as outlined here: https://vertx.io/docs/vertx-grpc/java/#_message_level_api does not work.
The reason is that ServiceName does not provide a custom equals method.
I encountered this bug in 4.4.2.
Support grpc-web protocol in Vert.x gRPC Server
The recent changs regarding the metadata/trailer seem to be causing issues:
2023-02-13T13:15:31.1187617Z 2023-02-13 13:15:31,036 ERROR [io.qua.ver.cor.run.VertxCoreRecorder] (vert.x-eventloop-thread-1) Uncaught exception received by Vert.x: java.lang.ArrayIndexOutOfBoundsException: Index 10 out of bounds for length 10
2023-02-13T13:15:31.1188475Z at io.vertx.grpc.common.impl.Utils.readMetadata(Utils.java:34)
2023-02-13T13:15:31.1189648Z at io.vertx.grpc.client.VertxClientCall.lambda$null$2(VertxClientCall.java:90)
2023-02-13T13:15:31.1190216Z at io.vertx.core.impl.future.FutureImpl$3.onSuccess(FutureImpl.java:141)
2023-02-13T13:15:31.1190754Z at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:60)
2023-02-13T13:15:31.1191353Z at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:211)
2023-02-13T13:15:31.1192254Z at io.vertx.core.impl.future.Mapping.onSuccess(Mapping.java:40)
2023-02-13T13:15:31.1192716Z at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:60)
2023-02-13T13:15:31.1193146Z at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:211)
2023-02-13T13:15:31.1193585Z at io.vertx.core.impl.future.PromiseImpl.tryComplete(PromiseImpl.java:23)
2023-02-13T13:15:31.1193972Z at io.vertx.core.Promise.complete(Promise.java:66)
2023-02-13T13:15:31.1194411Z at io.vertx.core.http.impl.HttpClientRequestImpl.handleResponse(HttpClientRequestImpl.java:346)
2023-02-13T13:15:31.1194955Z at io.vertx.core.http.impl.HttpClientRequestBase.handleResponse(HttpClientRequestBase.java:182)
2023-02-13T13:15:31.1195481Z at io.vertx.core.http.impl.HttpClientRequestBase.lambda$new$0(HttpClientRequestBase.java:71)
2023-02-13T13:15:31.1195926Z at io.vertx.core.impl.EventLoopContext.emit(EventLoopContext.java:55)
2023-02-13T13:15:31.1196306Z at io.vertx.core.impl.ContextBase.emit(ContextBase.java:239)
2023-02-13T13:15:31.1196736Z at io.vertx.core.http.impl.Http2ClientConnection$Stream.onHeaders(Http2ClientConnection.java:331)
2023-02-13T13:15:31.1197244Z at io.vertx.core.http.impl.Http2ClientConnection.onHeadersRead(Http2ClientConnection.java:189)
2023-02-13T13:15:31.1197749Z at io.vertx.core.http.impl.Http2ConnectionBase.onHeadersRead(Http2ConnectionBase.java:206)
2023-02-13T13:15:31.1198331Z at io.netty.handler.codec.http2.Http2FrameListenerDecorator.onHeadersRead(Http2FrameListenerDecorator.java:48)
2023-02-13T13:15:31.1198958Z at io.netty.handler.codec.http2.Http2EmptyDataFrameListener.onHeadersRead(Http2EmptyDataFrameListener.java:63)
2023-02-13T13:15:31.1199599Z at io.netty.handler.codec.http2.DefaultHttp2ConnectionDecoder$FrameReadListener.onHeadersRead(DefaultHttp2ConnectionDecoder.java:409)
2023-02-13T13:15:31.1200202Z at io.netty.handler.codec.http2.DefaultHttp2FrameReader$1.processFragment(DefaultHttp2FrameReader.java:450)
2023-02-13T13:15:31.1200773Z at io.netty.handler.codec.http2.DefaultHttp2FrameReader.readHeadersFrame(DefaultHttp2FrameReader.java:457)
2023-02-13T13:15:31.1201603Z at io.netty.handler.codec.http2.DefaultHttp2FrameReader.processPayloadState(DefaultHttp2FrameReader.java:253)
2023-02-13T13:15:31.1202200Z at io.netty.handler.codec.http2.DefaultHttp2FrameReader.readFrame(DefaultHttp2FrameReader.java:159)
2023-02-13T13:15:31.1202797Z at io.netty.handler.codec.http2.DefaultHttp2ConnectionDecoder.decodeFrame(DefaultHttp2ConnectionDecoder.java:173)
2023-02-13T13:15:31.1203487Z at io.netty.handler.codec.http2.DecoratingHttp2ConnectionDecoder.decodeFrame(DecoratingHttp2ConnectionDecoder.java:63)
2023-02-13T13:15:31.1204102Z at io.netty.handler.codec.http2.Http2ConnectionHandler$FrameDecoder.decode(Http2ConnectionHandler.java:393)
2023-02-13T13:15:31.1204641Z at io.netty.handler.codec.http2.Http2ConnectionHandler.decode(Http2ConnectionHandler.java:453)
2023-02-13T13:15:31.1205217Z at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:529)
2023-02-13T13:15:31.1205789Z at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:468)
2023-02-13T13:15:31.1206292Z at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)
2023-02-13T13:15:31.1206840Z at io.vertx.core.http.impl.VertxHttp2ConnectionHandler.channelRead(VertxHttp2ConnectionHandler.java:408)
The method serviceName() in GrpcMethodCall can throw an index out of bounds exception if the service path doesn't contain a dot.
I encountered this bug in 4.4.2.
The path I tried it with is "/Auth/GetChallenge"
gRPC provides a hook io.grpc.override.ContextStorageOverride
to let implementations provide their own storage.
This should be in a standalone jar, to let users decide or not to use this storage (as the io.grpc.override.ContextStorageOverride
is an hardcoded loaded class by gRPC for the moment).
4.4.1
When using vertx grpc client it seems that callOptions that holds deadline configuration is not being propagated to VertxClientCall.
Create a vertx stub:
GrpcClientChannel channel =
new GrpcClientChannel(
GrpcClient.client(vertx),
SocketAddress.inetSocketAddress(
port, host));
SomeVertxStub stub = VertxSomeServiceGrpc.newVertxStub(
channel);
Use this stub to make calls with deadline:
stub
.withDeadlineAfter(1, TimeUnit.MILLISECONDS)
The deadline which is a part callOptions is not being sent to VertxClientCall and not taken into account when making grpc calls so there is no way to set request timeout for grpc client calls.
A stub waiting for a response will not properly handle all error cases.
The Netty variant of the gRPC server allows configuring the keep-alive-time. However, the Vert.x variant does not.
Related to: quarkusio/quarkus#36033
The current implementation only implements grpc-status
currently. The protocol allows grpc-message
as well.
Any gRPC call could benefit from it. Especially useful during manual testing and logging.
The documentation and the 4.3 release blog post indicate that a GrpcServer
can be mounted in a Vert.x Web router.
In fact, it is not possible because the gRPC server is a Handler<HttpServerRequest>
not a Handler<RoutingContext>
. So this blog post example is wrong:
GreeterGrpc.GreeterImplBase service = new GreeterGrpc.GreeterImplBase() {
@Override
public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
responseObserver.onNext(HelloReply.newBuilder().setMessage("Hello " + request.getName()).build());
responseObserver.onCompleted();
}
};
GrpcServer grpcServer = GrpcServer.server(vertx);
GrpcServiceBridge serverStub = GrpcServiceBridge.bridge(service);
serverStub.bind(grpcServer);
router.consumes("application/grpc").handler(grpcServer); // does not compile
I am Grpc Server Use Vertx Grpc Serve build.Grpc Client do not use Vertx Grpc Client.Then I Loop 10 call ,but blocked at first。
4.3.7
public class VertxStart extends AbstractVerticle {
@Override
public void start() {
final var server = GrpcServer.server(vertx);
final var greeter = new GreeterGrpc.GreeterImplBase() {
@Override
public void sayHello(final HelloRequest request, final StreamObserver<HelloReply> responseObserver) {
System.out.println("Hello " + request.getName());
final var reply = HelloReply.newBuilder().setMessage(request.getName()).build();
responseObserver.onNext(reply);
responseObserver.onCompleted();
}
};
GrpcServiceBridge
.bridge(greeter)
.bind(server);
vertx.createHttpServer().requestHandler(server).listen(4240);
}
}
public static void main(String[] args) {
new GreeterClient().start();
}
public void start() {
final var channel = NettyChannelBuilder.forAddress("127.0.0.1", 4240).usePlaintext()
.build();
final var sub = GreeterGrpc.newBlockingStub(channel);
for (int i = 0; i < 10; i++) {
System.out.println("开始idx\t" + i);
final var data = HelloRequest.newBuilder().setName("qwx" + i).build();
final var resp = sub.sayHello(data);
System.out.println("resp\t" + resp.getMessage());
System.out.println("结束idx\t" + i);
}
channel.shutdown() ;
}
}
4.3.1 and 4.3.3-SNAPSHOT
My server is interrupted while running, the client does not have any exceptions or logs, It is not easy to locate the problem in time.
I looked at the source code and found that in the vertx-grpc-client
module io.vertx.grpc.client.VertxClientCall #start()
does not handle the client.request error.
4.3.7
Using request.exceptionHandler(...)
, bidirectional streams can respond to ungraceful stream closure, which could be killing an app, a browser crashing, a phone running out of battery, network disconnection or other failure.
This package does not correctly receive exceptions the way Netty does. It lacks the configuration that the Netty server has for keep alives. This appears to be the root issue.
jprotoc
based code generator. You will have a method of the signature:void bidiCall(ReadStream<X> request, WriteStream<Y> response);
VertxServer
. Configure it with timeouts:var builder = VertxServerBuilder.forPort(vertx, port);
var nettyServerBuilder = builder.nettyBuilder();
nettyServerBuilder
.maxConnectionIdle(Long.MAX_VALUE, TimeUnit.NANOSECONDS)
.maxConnectionAge(29, TimeUnit.DAYS)
.keepAliveTime(400, TimeUnit.MILLISECONDS)
.keepAliveTimeout(8000, TimeUnit.MILLISECONDS)
.permitKeepAliveTime(100, TimeUnit.MILLISECONDS)
.permitKeepAliveWithoutCalls(true);
builder.build();
StatusRuntimeException
with status Status.CANCELLED
is delivered to request.exceptionHandler
.GrpcServer
.endHandler
called in a short while.This is reproduced in the tests of my own open source program. I can share that one. Otherwise I will need help authoring tests in this package, since Maven in my IntelliJ does not appear to generate the protos / reference them correctly and it's impossible to code. Additionally I suggest adding TestContainers as a dependency for using ToxiProxy.
Currently the generator provides creation of both client and server, it should be possible to generate only of them when configuring the plugin
Thanks to your awesome work, now I can build a grpc reversed proxy using this library.
But the proxyResponse.errorHandler
is not called when the inner grpc service returns code like 2. Is this a bug or expected behavior? I can use the code:
// if (proxyResponse.status() != null) {
// clientReq.response().status(proxyResponse.status());
// }
to make a workaround, but it seems not that good.
private void startGrpcGatewayService() {
GrpcServer server = GrpcServer.server(vertx);
GrpcClient client = GrpcClient.client(vertx);
server.callHandler(clientReq -> {
clientReq.pause();
System.out.println(clientReq.fullMethodName());
SocketAddress address = SocketAddress.inetSocketAddress(8081, "127.0.0.1");
client.request(address)
.onSuccess(proxyRequest -> {
System.out.println("on grpc proxy success");
proxyRequest.response()
.onSuccess(proxyResponse -> {
System.out.println("proxy response success: " + proxyResponse.status());
// if (proxyResponse.status() != null) {
// clientReq.response().status(proxyResponse.status());
// }
proxyResponse.messageHandler(clientReq.response()::writeMessage);
proxyResponse.errorHandler(error -> {
// TODO why we cannot get proxy call fail code here?
System.out.println("proxy call error happened: " + proxyResponse.status());
clientReq.response().status(error.status);
});
proxyResponse.endHandler(voids -> clientReq.response().end());
})
.onFailure(exception -> {
System.out.println("proxy response fail: " + exception.getMessage());
exception.printStackTrace();
clientReq.response().status(GrpcStatus.UNKNOWN).end();
});
proxyRequest.fullMethodName(clientReq.fullMethodName());
clientReq.messageHandler(proxyRequest::writeMessage);
clientReq.endHandler(v -> proxyRequest.end());
clientReq.resume();
})
.onFailure(exception -> {
System.out.println("on grpc proxy fail");
clientReq.response().status(GrpcStatus.UNKNOWN).end();
clientReq.resume();
});
});
https://github.com/LangInteger/vertx-grpc-reversed-proxy
git clone [email protected]:LangInteger/vertx-grpc-reversed-proxy.git
cd vertx-grpc-reversed-proxy
./gradlew clean build
take a look at the helloWorldBrokenTest
This feature request proposes gRPC transcoding capabilities for the server. This would allow the server to accept json formatted grpc requests and transcode them according to the specifications defined in google.api.http.
While grpc web is already implemented, it utilizes a unique format that differs from the transcoding of REST requests into grpc requests. Implementing this feature would be particularly useful when publicly exposing a grpc service, where supporting web based requests or providing users with a REST api option is desired.
I am willing to contribute to the implementation of this feature. I have already implemented this in quarkus repository. After discussion in this PR, the maintainers concluded that this should be implemented directly in grpc server implementation.
The protoc plugin is not configurable. Consequently, it generates code for clients and servers, forcing users to add dependencies to both vertx-grpc-client
and vertx-grpc-server
.
The plugin should be configurable, in particular it should be easy to switch off code generation for clients or server from Maven.
We are using the older (Netty) version of the gRPC Server and implemented custom Snappy compression/decompression. We extended the (De)Compressor registry and registered it in the GrpcServer builder via
I couldn't find any option to register custom (de)compressors in grpc-netty
This feature is usefull when you need to compress the payload and don't want to use gzip because it's very heavy on CPU. I our case we decided to use Snappy because is comes with a nice balance between CPU usage and compression ratio
I'm happy to implement this feature and it might be enough to just allow a custom Compressor/Decompressor Registry to be added to the GrpcServiceBridge and use them instead of the default registries to find the compressor needed.
But I'm not sure
Thanks you
Markus
Futures-style stub generator backward compatible with https://github.com/vert-x3/vertx-grpc/blob/master/vertx-grpc-protoc-plugin/src/main/resources/VertxStub.mustache
Migrating to this package from the old grpc package.
I am blocked by being able to import this into my build pipeline.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.