UnframedGrpcService extends SimpleDecoratingHttpService implements GrpcService { @Override public HttpResponse serve(ServiceRequestContext ctx, HttpRequest req) throws Exception { final RequestHeaders clientHeaders = req.headers(); final MediaType contentType = clientHeaders.contentType(); if (contentType == null) { return unwrap().serve(ctx, req); } for (SerializationFormat format : GrpcSerializationFormats.values()) { if (format.isAccepted(contentType)) { return unwrap().serve(ctx, req); } } final String methodName = GrpcRequestUtil.determineMethod(ctx); final MethodDescriptor<?, ?> method = methodName != null ? methodsByName.get(methodName) : null; if (method == null) { return unwrap().serve(ctx, req); } if (method.getType() != MethodType.UNARY) { return HttpResponse.of(HttpStatus.BAD_REQUEST, MediaType.PLAIN_TEXT_UTF_8, "Only unary methods can be used with non-framed requests."); } final RequestHeadersBuilder grpcHeaders = clientHeaders.toBuilder(); final MediaType framedContentType; if (contentType.is(MediaType.PROTOBUF)) { framedContentType = GrpcSerializationFormats.PROTO.mediaType(); } else if (contentType.is(MediaType.JSON_UTF_8)) { framedContentType = GrpcSerializationFormats.JSON.mediaType(); } else { return HttpResponse.of(HttpStatus.UNSUPPORTED_MEDIA_TYPE, MediaType.PLAIN_TEXT_UTF_8, "Unsupported media type. Only application/protobuf is supported."); } grpcHeaders.contentType(framedContentType); if (grpcHeaders.get(GrpcHeaderNames.GRPC_ENCODING) != null) { return HttpResponse.of(HttpStatus.UNSUPPORTED_MEDIA_TYPE, MediaType.PLAIN_TEXT_UTF_8, "gRPC encoding is not supported for non-framed requests."); } grpcHeaders.remove(GrpcHeaderNames.GRPC_ACCEPT_ENCODING); ctx.logBuilder().defer(RequestLogProperty.REQUEST_CONTENT, RequestLogProperty.RESPONSE_CONTENT); final CompletableFuture<HttpResponse> responseFuture = new CompletableFuture<>(); req.aggregateWithPooledObjects(ctx.eventLoop(), ctx.alloc()).handle((clientRequest, t) -> { if (t != null) { responseFuture.completeExceptionally(t); } else { frameAndServe(ctx, grpcHeaders.build(), clientRequest, responseFuture); } return null; }); return HttpResponse.from(responseFuture); } UnframedGrpcService(GrpcService delegate); @Override boolean isFramed(); @Override List<ServerServiceDefinition> services(); @Override Set<SerializationFormat> supportedSerializationFormats(); @Override HttpResponse serve(ServiceRequestContext ctx, HttpRequest req); @Override Set<Route> routes();  }
@Test public void statusOk() throws Exception { final UnframedGrpcService unframedGrpcService = buildUnframedGrpcService(testService); final HttpResponse response = unframedGrpcService.serve(ctx, request); final AggregatedHttpResponse res = response.aggregate().get(); assertThat(res.status()).isEqualTo(HttpStatus.OK); assertThat(res.contentUtf8()).isEqualTo("{}"); }
@Test public void statusCancelled() throws Exception { final TestService spyTestService = spy(testService); doThrow(Status.CANCELLED.withDescription("grpc error message").asRuntimeException()) .when(spyTestService) .emptyCall(any(), any()); final UnframedGrpcService unframedGrpcService = buildUnframedGrpcService(spyTestService); final HttpResponse response = unframedGrpcService.serve(ctx, request); final AggregatedHttpResponse res = response.aggregate().get(); assertThat(res.status()).isEqualTo(HttpStatus.CLIENT_CLOSED_REQUEST); assertThat(res.contentUtf8()) .isEqualTo("http-status: 499, Client Closed Request\n" + "Caused by: \n" + "grpc-status: 1, CANCELLED, grpc error message"); }
@Test public void noContent() throws Exception { final UnframedGrpcService unframedGrpcService = buildUnframedGrpcService(new TestServiceImplBase() { @Override public void emptyCall(Empty request, StreamObserver<Empty> responseObserver) { responseObserver.onCompleted(); } }); final HttpResponse response = unframedGrpcService.serve(ctx, request); final AggregatedHttpResponse res = response.aggregate().get(); assertThat(res.status()).isEqualTo(HttpStatus.OK); assertThat(res.content().isEmpty()).isTrue(); }