@Test public void testAppLogicThrowingAnErrorWithDescription() { Status grpcStatus = Status.ALREADY_EXISTS.withDescription("This thing done been here"); GrpcAppLogic<HelloRequest, HelloReply> appLogic = (HelloRequest request) -> { throw new StatusException(grpcStatus); }; subject = new GrpcRequestHandler<>(HelloRequest::parseFrom, appLogic); channel = new EmbeddedChannel( new SimpleChannelInboundHandler<Request>() { @Override protected void channelRead0(ChannelHandlerContext ctx, Request request) { subject.handle(ctx, request, null); } }); ByteBuf grpcRequestBuffer = bufferFor(HelloRequest.getDefaultInstance()); int streamId = 123; SegmentedRequestData segmentedRequest = fullGrpcRequest(grpcRequestBuffer, streamId, true); channel.writeInbound(segmentedRequest); Response response = channel.readOutbound(); SegmentedData segmentedData = channel.readOutbound(); assertEquals(HttpResponseStatus.OK, response.status()); assertEquals(streamId, response.streamId()); assertEquals("application/grpc+proto", response.headers().get(HttpHeaderNames.CONTENT_TYPE)); assertFalse(segmentedData.content().isReadable()); assertEquals( Integer.toString(grpcStatus.getCode().value()), Objects.requireNonNull(segmentedData.trailingHeaders()).get("grpc-status")); String actualMessage = grpcDecodedString( Objects.requireNonNull( Objects.requireNonNull(segmentedData.trailingHeaders()).get("grpc-message"))); assertEquals(grpcStatus.getDescription(), actualMessage); assertEquals(streamId, segmentedData.streamId()); assertTrue(segmentedData.endOfMessage()); }
@Override public void handle(ChannelHandlerContext ctx, Request request, RouteState route) { ByteBuf actualBuffer; if (request instanceof SegmentedRequestData) { SegmentedRequestData segmentedRequest = (SegmentedRequestData) request; actualBuffer = segmentedRequest.content(); } else { actualBuffer = request.body(); } if (actualBuffer == null || !actualBuffer.isReadable()) { return; } HashMap<Integer, GrpcState> session = lazyCreateSession(ctx); GrpcState state = session.get(request.streamId()); if (state == null) { boolean hasMetaData = actualBuffer.isReadable(METADATA_SIZE); if (!hasMetaData) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_NO_METADATA)); return; } boolean isCompressed = actualBuffer.slice(0, 1).readBoolean(); if (isCompressed) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.UNIMPLEMENTED.withDescription(GRPC_MESSAGE_NO_COMPRESSION)); return; } int size = actualBuffer.slice(1, 4).readInt(); if (size >= MAX_PAYLOAD_SIZE) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.RESOURCE_EXHAUSTED.withDescription(GRPC_MESSAGE_LARGE_SIZE)); return; } int packetContentSize = actualBuffer.readableBytes() - METADATA_SIZE; boolean firstChunkIsLargerThanIndicatedSize = packetContentSize > size; if (firstChunkIsLargerThanIndicatedSize) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_WRONG_SIZE)); return; } ByteBuf contentBuffer = UnpooledByteBufAllocator.DEFAULT.buffer(packetContentSize, size); contentBuffer.writeBytes(actualBuffer.slice(METADATA_SIZE, packetContentSize)); state = new GrpcState(size, contentBuffer); } else { boolean accumulatedIsLargerThanIndicatedSize = state.buffer.readableBytes() + actualBuffer.readableBytes() > state.size; if (accumulatedIsLargerThanIndicatedSize) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_WRONG_SIZE)); return; } state.buffer.writeBytes(actualBuffer); } if (request.endOfMessage()) { handleGrpcRequest(ctx, state, request.streamId()); } else { session.put(request.streamId(), state); } }
GrpcRequestHandler implements PipelineRequestHandler { @Override public void handle(ChannelHandlerContext ctx, Request request, RouteState route) { ByteBuf actualBuffer; if (request instanceof SegmentedRequestData) { SegmentedRequestData segmentedRequest = (SegmentedRequestData) request; actualBuffer = segmentedRequest.content(); } else { actualBuffer = request.body(); } if (actualBuffer == null || !actualBuffer.isReadable()) { return; } HashMap<Integer, GrpcState> session = lazyCreateSession(ctx); GrpcState state = session.get(request.streamId()); if (state == null) { boolean hasMetaData = actualBuffer.isReadable(METADATA_SIZE); if (!hasMetaData) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_NO_METADATA)); return; } boolean isCompressed = actualBuffer.slice(0, 1).readBoolean(); if (isCompressed) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.UNIMPLEMENTED.withDescription(GRPC_MESSAGE_NO_COMPRESSION)); return; } int size = actualBuffer.slice(1, 4).readInt(); if (size >= MAX_PAYLOAD_SIZE) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.RESOURCE_EXHAUSTED.withDescription(GRPC_MESSAGE_LARGE_SIZE)); return; } int packetContentSize = actualBuffer.readableBytes() - METADATA_SIZE; boolean firstChunkIsLargerThanIndicatedSize = packetContentSize > size; if (firstChunkIsLargerThanIndicatedSize) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_WRONG_SIZE)); return; } ByteBuf contentBuffer = UnpooledByteBufAllocator.DEFAULT.buffer(packetContentSize, size); contentBuffer.writeBytes(actualBuffer.slice(METADATA_SIZE, packetContentSize)); state = new GrpcState(size, contentBuffer); } else { boolean accumulatedIsLargerThanIndicatedSize = state.buffer.readableBytes() + actualBuffer.readableBytes() > state.size; if (accumulatedIsLargerThanIndicatedSize) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_WRONG_SIZE)); return; } state.buffer.writeBytes(actualBuffer); } if (request.endOfMessage()) { handleGrpcRequest(ctx, state, request.streamId()); } else { session.put(request.streamId(), state); } } }
GrpcRequestHandler implements PipelineRequestHandler { @Override public void handle(ChannelHandlerContext ctx, Request request, RouteState route) { ByteBuf actualBuffer; if (request instanceof SegmentedRequestData) { SegmentedRequestData segmentedRequest = (SegmentedRequestData) request; actualBuffer = segmentedRequest.content(); } else { actualBuffer = request.body(); } if (actualBuffer == null || !actualBuffer.isReadable()) { return; } HashMap<Integer, GrpcState> session = lazyCreateSession(ctx); GrpcState state = session.get(request.streamId()); if (state == null) { boolean hasMetaData = actualBuffer.isReadable(METADATA_SIZE); if (!hasMetaData) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_NO_METADATA)); return; } boolean isCompressed = actualBuffer.slice(0, 1).readBoolean(); if (isCompressed) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.UNIMPLEMENTED.withDescription(GRPC_MESSAGE_NO_COMPRESSION)); return; } int size = actualBuffer.slice(1, 4).readInt(); if (size >= MAX_PAYLOAD_SIZE) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.RESOURCE_EXHAUSTED.withDescription(GRPC_MESSAGE_LARGE_SIZE)); return; } int packetContentSize = actualBuffer.readableBytes() - METADATA_SIZE; boolean firstChunkIsLargerThanIndicatedSize = packetContentSize > size; if (firstChunkIsLargerThanIndicatedSize) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_WRONG_SIZE)); return; } ByteBuf contentBuffer = UnpooledByteBufAllocator.DEFAULT.buffer(packetContentSize, size); contentBuffer.writeBytes(actualBuffer.slice(METADATA_SIZE, packetContentSize)); state = new GrpcState(size, contentBuffer); } else { boolean accumulatedIsLargerThanIndicatedSize = state.buffer.readableBytes() + actualBuffer.readableBytes() > state.size; if (accumulatedIsLargerThanIndicatedSize) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_WRONG_SIZE)); return; } state.buffer.writeBytes(actualBuffer); } if (request.endOfMessage()) { handleGrpcRequest(ctx, state, request.streamId()); } else { session.put(request.streamId(), state); } } GrpcRequestHandler(
      GrpcRequestParser<GrpcRequest> requestParser,
      GrpcAppLogic<GrpcRequest, GrpcResponse> appLogic); }
GrpcRequestHandler implements PipelineRequestHandler { @Override public void handle(ChannelHandlerContext ctx, Request request, RouteState route) { ByteBuf actualBuffer; if (request instanceof SegmentedRequestData) { SegmentedRequestData segmentedRequest = (SegmentedRequestData) request; actualBuffer = segmentedRequest.content(); } else { actualBuffer = request.body(); } if (actualBuffer == null || !actualBuffer.isReadable()) { return; } HashMap<Integer, GrpcState> session = lazyCreateSession(ctx); GrpcState state = session.get(request.streamId()); if (state == null) { boolean hasMetaData = actualBuffer.isReadable(METADATA_SIZE); if (!hasMetaData) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_NO_METADATA)); return; } boolean isCompressed = actualBuffer.slice(0, 1).readBoolean(); if (isCompressed) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.UNIMPLEMENTED.withDescription(GRPC_MESSAGE_NO_COMPRESSION)); return; } int size = actualBuffer.slice(1, 4).readInt(); if (size >= MAX_PAYLOAD_SIZE) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.RESOURCE_EXHAUSTED.withDescription(GRPC_MESSAGE_LARGE_SIZE)); return; } int packetContentSize = actualBuffer.readableBytes() - METADATA_SIZE; boolean firstChunkIsLargerThanIndicatedSize = packetContentSize > size; if (firstChunkIsLargerThanIndicatedSize) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_WRONG_SIZE)); return; } ByteBuf contentBuffer = UnpooledByteBufAllocator.DEFAULT.buffer(packetContentSize, size); contentBuffer.writeBytes(actualBuffer.slice(METADATA_SIZE, packetContentSize)); state = new GrpcState(size, contentBuffer); } else { boolean accumulatedIsLargerThanIndicatedSize = state.buffer.readableBytes() + actualBuffer.readableBytes() > state.size; if (accumulatedIsLargerThanIndicatedSize) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_WRONG_SIZE)); return; } state.buffer.writeBytes(actualBuffer); } if (request.endOfMessage()) { handleGrpcRequest(ctx, state, request.streamId()); } else { session.put(request.streamId(), state); } } GrpcRequestHandler(
      GrpcRequestParser<GrpcRequest> requestParser,
      GrpcAppLogic<GrpcRequest, GrpcResponse> appLogic); GrpcAppLogic<GrpcRequest, GrpcResponse> getAppLogic(); @Override void handle(ChannelHandlerContext ctx, Request request, RouteState route); }
GrpcRequestHandler implements PipelineRequestHandler { @Override public void handle(ChannelHandlerContext ctx, Request request, RouteState route) { ByteBuf actualBuffer; if (request instanceof SegmentedRequestData) { SegmentedRequestData segmentedRequest = (SegmentedRequestData) request; actualBuffer = segmentedRequest.content(); } else { actualBuffer = request.body(); } if (actualBuffer == null || !actualBuffer.isReadable()) { return; } HashMap<Integer, GrpcState> session = lazyCreateSession(ctx); GrpcState state = session.get(request.streamId()); if (state == null) { boolean hasMetaData = actualBuffer.isReadable(METADATA_SIZE); if (!hasMetaData) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_NO_METADATA)); return; } boolean isCompressed = actualBuffer.slice(0, 1).readBoolean(); if (isCompressed) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.UNIMPLEMENTED.withDescription(GRPC_MESSAGE_NO_COMPRESSION)); return; } int size = actualBuffer.slice(1, 4).readInt(); if (size >= MAX_PAYLOAD_SIZE) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.RESOURCE_EXHAUSTED.withDescription(GRPC_MESSAGE_LARGE_SIZE)); return; } int packetContentSize = actualBuffer.readableBytes() - METADATA_SIZE; boolean firstChunkIsLargerThanIndicatedSize = packetContentSize > size; if (firstChunkIsLargerThanIndicatedSize) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_WRONG_SIZE)); return; } ByteBuf contentBuffer = UnpooledByteBufAllocator.DEFAULT.buffer(packetContentSize, size); contentBuffer.writeBytes(actualBuffer.slice(METADATA_SIZE, packetContentSize)); state = new GrpcState(size, contentBuffer); } else { boolean accumulatedIsLargerThanIndicatedSize = state.buffer.readableBytes() + actualBuffer.readableBytes() > state.size; if (accumulatedIsLargerThanIndicatedSize) { sendResponse( ctx, request.streamId(), Unpooled.EMPTY_BUFFER, Status.INTERNAL.withDescription(GRPC_MESSAGE_WRONG_SIZE)); return; } state.buffer.writeBytes(actualBuffer); } if (request.endOfMessage()) { handleGrpcRequest(ctx, state, request.streamId()); } else { session.put(request.streamId(), state); } } GrpcRequestHandler(
      GrpcRequestParser<GrpcRequest> requestParser,
      GrpcAppLogic<GrpcRequest, GrpcResponse> appLogic); GrpcAppLogic<GrpcRequest, GrpcResponse> getAppLogic(); @Override void handle(ChannelHandlerContext ctx, Request request, RouteState route);  }