diff --git a/webserver/grpc/pom.xml b/webserver/grpc/pom.xml index a42a6f533cf..7a05ede4a52 100644 --- a/webserver/grpc/pom.xml +++ b/webserver/grpc/pom.xml @@ -150,6 +150,23 @@ + + org.xolstice.maven.plugins + protobuf-maven-plugin + + + + test-compile + + + + + com.google.protobuf:protoc:${version.lib.google-protobuf}:exe:${os.detected.classifier} + grpc-java + io.grpc:protoc-gen-grpc-java:${version.lib.grpc}:exe:${os.detected.classifier} + + + diff --git a/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcProtocolHandler.java b/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcProtocolHandler.java index 73b7968cf9b..0e4d62f4a29 100644 --- a/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcProtocolHandler.java +++ b/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcProtocolHandler.java @@ -44,6 +44,7 @@ import io.helidon.http.http2.StreamFlowControl; import io.helidon.webserver.http2.spi.Http2SubProtocolSelector; +import io.grpc.Codec; import io.grpc.Compressor; import io.grpc.CompressorRegistry; import io.grpc.Decompressor; @@ -90,6 +91,7 @@ class GrpcProtocolHandler implements Http2SubProtocolSelector.SubProto private BufferData entityBytes; private Compressor compressor; private Decompressor decompressor; + private boolean isIdentityCompressor; GrpcProtocolHandler(HttpPrologue prologue, Http2Headers headers, @@ -115,41 +117,10 @@ class GrpcProtocolHandler implements Http2SubProtocolSelector.SubProto public void init() { try { ServerCall serverCall = createServerCall(); - Headers httpHeaders = headers.httpHeaders(); - // check for encoding and respond using same algorithm - if (httpHeaders.contains(GRPC_ENCODING)) { - Header grpcEncoding = httpHeaders.get(GRPC_ENCODING); - String encoding = grpcEncoding.asString().get(); - decompressor = DECOMPRESSOR_REGISTRY.lookupDecompressor(encoding); - compressor = COMPRESSOR_REGISTRY.lookupCompressor(encoding); - - // report encoding not supported - if (decompressor == null || compressor == null) { - Metadata metadata = new Metadata(); - Set encodings = DECOMPRESSOR_REGISTRY.getAdvertisedMessageEncodings(); - metadata.put(Metadata.Key.of(GRPC_ACCEPT_ENCODING.defaultCase(), Metadata.ASCII_STRING_MARSHALLER), - String.join(",", encodings)); - serverCall.close(Status.UNIMPLEMENTED, metadata); - currentStreamState = Http2StreamState.CLOSED; // stops processing - return; - } - } else if (httpHeaders.contains(GRPC_ACCEPT_ENCODING)) { - Header acceptEncoding = httpHeaders.get(GRPC_ACCEPT_ENCODING); - - // check for matching encoding - for (String encoding : acceptEncoding.allValues()) { - compressor = COMPRESSOR_REGISTRY.lookupCompressor(encoding); - if (compressor != null) { - decompressor = DECOMPRESSOR_REGISTRY.lookupDecompressor(encoding); - if (decompressor != null) { - break; // found match - } - compressor = null; - } - } - } + // setup compression + initCompression(serverCall, httpHeaders); // initiate server call ServerCallHandler callHandler = route.callHandler(); @@ -161,10 +132,6 @@ public void init() { } } - private void addNumMessages(int n) { - numMessages.getAndAdd(n); - } - @Override public Http2StreamState streamState() { return currentStreamState; @@ -224,6 +191,52 @@ public void data(Http2FrameHeader header, BufferData data) { } } + void initCompression(ServerCall serverCall, Headers httpHeaders) { + // check for encoding and respond using same algorithm + if (httpHeaders.contains(GRPC_ENCODING)) { + Header grpcEncoding = httpHeaders.get(GRPC_ENCODING); + String encoding = grpcEncoding.asString().get(); + decompressor = DECOMPRESSOR_REGISTRY.lookupDecompressor(encoding); + compressor = COMPRESSOR_REGISTRY.lookupCompressor(encoding); + + // report encoding not supported + if (decompressor == null || compressor == null) { + Metadata metadata = new Metadata(); + Set encodings = DECOMPRESSOR_REGISTRY.getAdvertisedMessageEncodings(); + metadata.put(Metadata.Key.of(GRPC_ACCEPT_ENCODING.defaultCase(), Metadata.ASCII_STRING_MARSHALLER), + String.join(",", encodings)); + serverCall.close(Status.UNIMPLEMENTED, metadata); + currentStreamState = Http2StreamState.CLOSED; // stops processing + return; + } + } else if (httpHeaders.contains(GRPC_ACCEPT_ENCODING)) { + Header acceptEncoding = httpHeaders.get(GRPC_ACCEPT_ENCODING); + + // check for matching encoding + for (String encoding : acceptEncoding.allValues()) { + compressor = COMPRESSOR_REGISTRY.lookupCompressor(encoding); + if (compressor != null) { + decompressor = DECOMPRESSOR_REGISTRY.lookupDecompressor(encoding); + if (decompressor != null) { + break; // found match + } + compressor = null; + } + } + } + + // special handling for identity compressor + isIdentityCompressor = (compressor instanceof Codec.Identity); + } + + boolean isIdentityCompressor() { + return isIdentityCompressor; + } + + private void addNumMessages(int n) { + numMessages.getAndAdd(n); + } + private void flushQueue() { if (listener != null) { while (!listenerQueue.isEmpty() && numMessages.getAndDecrement() > 0) { @@ -268,10 +281,10 @@ public void sendMessage(RES message) { try (InputStream inputStream = route.method().streamResponse(message)) { // prepare buffer for writing BufferData bufferData; - if (compressor == null) { + if (compressor == null || isIdentityCompressor) { byte[] bytes = inputStream.readAllBytes(); bufferData = BufferData.create(5 + bytes.length); - bufferData.write(0); + bufferData.write(0); // off for identity compressor bufferData.writeUnsignedInt32(bytes.length); bufferData.write(bytes); } else { diff --git a/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcProtocolHandlerNotFound.java b/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcProtocolHandlerNotFound.java index 316131404a2..0c0a64cec7b 100644 --- a/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcProtocolHandlerNotFound.java +++ b/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcProtocolHandlerNotFound.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2022, 2023 Oracle and/or its affiliates. + * Copyright (c) 2022, 2024 Oracle and/or its affiliates. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package io.helidon.webserver.grpc; import io.helidon.common.buffers.BufferData; +import io.helidon.http.Status; import io.helidon.http.WritableHeaders; import io.helidon.http.http2.FlowControl; import io.helidon.http.http2.Http2Flag; @@ -45,6 +46,7 @@ class GrpcProtocolHandlerNotFound implements Http2SubProtocolSelector.SubProtoco @Override public void init() { WritableHeaders writable = WritableHeaders.create(); + writable.set(Http2Headers.STATUS_NAME, Status.NOT_FOUND_404.code()); writable.set(GrpcStatus.NOT_FOUND); Http2Headers http2Headers = Http2Headers.create(writable); streamWriter.writeHeaders(http2Headers, @@ -70,5 +72,4 @@ public void windowUpdate(Http2WindowUpdate update) { @Override public void data(Http2FrameHeader header, BufferData data) { } - } diff --git a/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcRouteHandler.java b/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcRouteHandler.java index 92c74008991..471a5f6240e 100644 --- a/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcRouteHandler.java +++ b/webserver/grpc/src/main/java/io/helidon/webserver/grpc/GrpcRouteHandler.java @@ -120,8 +120,14 @@ private static GrpcRouteHandler grpc(Descriptors.FileDe String methodName, ServerCallHandler callHandler) { Descriptors.ServiceDescriptor svc = proto.findServiceByName(serviceName); + if (svc == null) { + throw new IllegalArgumentException("Unable to find gRPC service '" + serviceName + "'"); + } Descriptors.MethodDescriptor mtd = svc.findMethodByName(methodName); - + if (mtd == null) { + throw new IllegalArgumentException("Unable to find gRPC method '" + methodName + + "' in service '" + serviceName + "'"); + } String path = svc.getFullName() + "/" + methodName; /* diff --git a/webserver/grpc/src/test/java/io/helidon/webserver/grpc/GrpcProtocolHandlerNotFoundTest.java b/webserver/grpc/src/test/java/io/helidon/webserver/grpc/GrpcProtocolHandlerNotFoundTest.java new file mode 100644 index 00000000000..bc1cb5fe4cc --- /dev/null +++ b/webserver/grpc/src/test/java/io/helidon/webserver/grpc/GrpcProtocolHandlerNotFoundTest.java @@ -0,0 +1,70 @@ +/* + * Copyright (c) 2024 Oracle and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.helidon.webserver.grpc; + +import io.helidon.http.Status; +import io.helidon.http.http2.FlowControl; +import io.helidon.http.http2.Http2Flag; +import io.helidon.http.http2.Http2FrameData; +import io.helidon.http.http2.Http2Headers; +import io.helidon.http.http2.Http2StreamState; +import io.helidon.http.http2.Http2StreamWriter; +import org.junit.jupiter.api.Test; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.CoreMatchers.is; + +class GrpcProtocolHandlerNotFoundTest { + + private boolean validateHeaders; + + @Test + void testNotFoundHeaders() { + Http2StreamWriter writer = new Http2StreamWriter() { + @Override + public void write(Http2FrameData frame) { + throw new UnsupportedOperationException("Unsupported"); + } + + @Override + public void writeData(Http2FrameData frame, FlowControl.Outbound flowControl) { + throw new UnsupportedOperationException("Unsupported"); + + } + + @Override + public int writeHeaders(Http2Headers headers, int streamId, Http2Flag.HeaderFlags flags, FlowControl.Outbound flowControl) { + validateHeaders = (headers.status() == Status.NOT_FOUND_404); + try { + headers.validateResponse(); + } catch (Exception e) { + validateHeaders = false; + } + return 0; + } + + @Override + public int writeHeaders(Http2Headers headers, int streamId, Http2Flag.HeaderFlags flags, Http2FrameData dataFrame, FlowControl.Outbound flowControl) { + throw new UnsupportedOperationException("Unsupported"); + } + }; + GrpcProtocolHandlerNotFound handler = new GrpcProtocolHandlerNotFound(writer, 1, Http2StreamState.OPEN); + assertThat(validateHeaders, is(false)); + handler.init(); + assertThat(validateHeaders, is(true)); + } +} diff --git a/webserver/grpc/src/test/java/io/helidon/webserver/grpc/GrpcProtocolHandlerTest.java b/webserver/grpc/src/test/java/io/helidon/webserver/grpc/GrpcProtocolHandlerTest.java new file mode 100644 index 00000000000..5638cc38126 --- /dev/null +++ b/webserver/grpc/src/test/java/io/helidon/webserver/grpc/GrpcProtocolHandlerTest.java @@ -0,0 +1,69 @@ +/* + * Copyright (c) 2024 Oracle and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.helidon.webserver.grpc; + +import io.helidon.http.HeaderName; +import io.helidon.http.HeaderNames; +import io.helidon.http.WritableHeaders; +import io.helidon.http.http2.Http2Headers; +import io.helidon.http.http2.Http2Settings; +import io.helidon.http.http2.Http2StreamState; +import org.junit.jupiter.api.Test; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.CoreMatchers.is; + +class GrpcProtocolHandlerTest { + + private static final HeaderName GRPC_ACCEPT_ENCODING = HeaderNames.create("grpc-accept-encoding"); + + @Test + @SuppressWarnings("unchecked") + void testIdentityCompressorFlag() { + WritableHeaders headers = WritableHeaders.create(); + headers.add(GRPC_ACCEPT_ENCODING, "identity"); + GrpcProtocolHandler handler = new GrpcProtocolHandler(null, + Http2Headers.create(headers), + null, + 1, + Http2Settings.builder().build(), + Http2Settings.builder().build(), + null, + Http2StreamState.OPEN, + null); + handler.initCompression(null, headers); + assertThat(handler.isIdentityCompressor(), is(true)); + } + + @Test + @SuppressWarnings("unchecked") + void testGzipCompressor() { + WritableHeaders headers = WritableHeaders.create(); + headers.add(GRPC_ACCEPT_ENCODING, "gzip"); + GrpcProtocolHandler handler = new GrpcProtocolHandler(null, + Http2Headers.create(headers), + null, + 1, + Http2Settings.builder().build(), + Http2Settings.builder().build(), + null, + Http2StreamState.OPEN, + null); + handler.initCompression(null, headers); + assertThat(handler.isIdentityCompressor(), is(false)); + } +} diff --git a/webserver/grpc/src/test/java/io/helidon/webserver/grpc/GrpcRouteHandlerTest.java b/webserver/grpc/src/test/java/io/helidon/webserver/grpc/GrpcRouteHandlerTest.java new file mode 100644 index 00000000000..948d7558963 --- /dev/null +++ b/webserver/grpc/src/test/java/io/helidon/webserver/grpc/GrpcRouteHandlerTest.java @@ -0,0 +1,33 @@ +/* + * Copyright (c) 2024 Oracle and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.helidon.webserver.grpc; + +import com.google.protobuf.Descriptors; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class GrpcRouteHandlerTest { + + @Test + void testBadServiceNames() throws Descriptors.DescriptorValidationException { + assertThrows(IllegalArgumentException.class, + () -> GrpcRouteHandler.unary(Strings.getDescriptor(), "foo", "Upper", null)); + assertThrows(IllegalArgumentException.class, + () -> GrpcRouteHandler.unary(Strings.getDescriptor(), "StringService", "foo", null)); + } +} diff --git a/webserver/grpc/src/test/proto/strings.proto b/webserver/grpc/src/test/proto/strings.proto new file mode 100644 index 00000000000..5242a102593 --- /dev/null +++ b/webserver/grpc/src/test/proto/strings.proto @@ -0,0 +1,27 @@ +/* + * Copyright (c) 2024 Oracle and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + +syntax = "proto3"; +option java_package = "io.helidon.webserver.grpc"; + +service StringService { + rpc Upper (StringMessage) returns (StringMessage) {} +} + +message StringMessage { + string text = 1; +}