This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 0cac563f63f add server interceptors (#7455)
0cac563f63f is described below
commit 0cac563f63f40abe88c9355b516d2c06bbda3b8b
Author: Mohamad Hadi El Hajj <[email protected]>
AuthorDate: Mon Apr 18 23:59:59 2022 -0500
add server interceptors (#7455)
---
.../component/grpc/GrpcEndpointConfigurer.java | 6 +
.../component/grpc/GrpcEndpointUriFactory.java | 3 +-
.../org/apache/camel/component/grpc/grpc.json | 3 +-
.../apache/camel/component/grpc/GrpcComponent.java | 11 ++
.../camel/component/grpc/GrpcConfiguration.java | 32 +++-
.../apache/camel/component/grpc/GrpcConsumer.java | 4 +
.../grpc/GrpcConsumerServerInterceptorTest.java | 167 +++++++++++++++++++++
.../component/grpc/GrpcMockServerInterceptor.java | 39 +++++
8 files changed, 262 insertions(+), 3 deletions(-)
diff --git
a/components/camel-grpc/src/generated/java/org/apache/camel/component/grpc/GrpcEndpointConfigurer.java
b/components/camel-grpc/src/generated/java/org/apache/camel/component/grpc/GrpcEndpointConfigurer.java
index 3e0c0234396..9a849bc6836 100644
---
a/components/camel-grpc/src/generated/java/org/apache/camel/component/grpc/GrpcEndpointConfigurer.java
+++
b/components/camel-grpc/src/generated/java/org/apache/camel/component/grpc/GrpcEndpointConfigurer.java
@@ -25,6 +25,8 @@ public class GrpcEndpointConfigurer extends
PropertyConfigurerSupport implements
case "authenticationType":
target.getConfiguration().setAuthenticationType(property(camelContext,
org.apache.camel.component.grpc.GrpcAuthType.class, value)); return true;
case "autodiscoverclientinterceptors":
case "autoDiscoverClientInterceptors":
target.getConfiguration().setAutoDiscoverClientInterceptors(property(camelContext,
boolean.class, value)); return true;
+ case "autodiscoverserverinterceptors":
+ case "autoDiscoverServerInterceptors":
target.getConfiguration().setAutoDiscoverServerInterceptors(property(camelContext,
boolean.class, value)); return true;
case "bridgeerrorhandler":
case "bridgeErrorHandler":
target.setBridgeErrorHandler(property(camelContext, boolean.class, value));
return true;
case "consumerstrategy":
@@ -86,6 +88,8 @@ public class GrpcEndpointConfigurer extends
PropertyConfigurerSupport implements
case "authenticationType": return
org.apache.camel.component.grpc.GrpcAuthType.class;
case "autodiscoverclientinterceptors":
case "autoDiscoverClientInterceptors": return boolean.class;
+ case "autodiscoverserverinterceptors":
+ case "autoDiscoverServerInterceptors": return boolean.class;
case "bridgeerrorhandler":
case "bridgeErrorHandler": return boolean.class;
case "consumerstrategy":
@@ -148,6 +152,8 @@ public class GrpcEndpointConfigurer extends
PropertyConfigurerSupport implements
case "authenticationType": return
target.getConfiguration().getAuthenticationType();
case "autodiscoverclientinterceptors":
case "autoDiscoverClientInterceptors": return
target.getConfiguration().isAutoDiscoverClientInterceptors();
+ case "autodiscoverserverinterceptors":
+ case "autoDiscoverServerInterceptors": return
target.getConfiguration().isAutoDiscoverServerInterceptors();
case "bridgeerrorhandler":
case "bridgeErrorHandler": return target.isBridgeErrorHandler();
case "consumerstrategy":
diff --git
a/components/camel-grpc/src/generated/java/org/apache/camel/component/grpc/GrpcEndpointUriFactory.java
b/components/camel-grpc/src/generated/java/org/apache/camel/component/grpc/GrpcEndpointUriFactory.java
index 40cf2ecf0ac..d4350e5c7cc 100644
---
a/components/camel-grpc/src/generated/java/org/apache/camel/component/grpc/GrpcEndpointUriFactory.java
+++
b/components/camel-grpc/src/generated/java/org/apache/camel/component/grpc/GrpcEndpointUriFactory.java
@@ -21,9 +21,10 @@ public class GrpcEndpointUriFactory extends
org.apache.camel.support.component.E
private static final Set<String> SECRET_PROPERTY_NAMES;
private static final Set<String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(31);
+ Set<String> props = new HashSet<>(32);
props.add("authenticationType");
props.add("autoDiscoverClientInterceptors");
+ props.add("autoDiscoverServerInterceptors");
props.add("bridgeErrorHandler");
props.add("consumerStrategy");
props.add("exceptionHandler");
diff --git
a/components/camel-grpc/src/generated/resources/org/apache/camel/component/grpc/grpc.json
b/components/camel-grpc/src/generated/resources/org/apache/camel/component/grpc/grpc.json
index 8162da88480..42c4d5ef9d1 100644
---
a/components/camel-grpc/src/generated/resources/org/apache/camel/component/grpc/grpc.json
+++
b/components/camel-grpc/src/generated/resources/org/apache/camel/component/grpc/grpc.json
@@ -35,9 +35,9 @@
"host": { "kind": "path", "displayName": "Host", "group": "common",
"label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.grpc.GrpcConfiguration", "configurationField":
"configuration", "description": "The gRPC server host name. This is localhost
or 0.0.0.0 when being a consumer or remote server host name when using
producer." },
"port": { "kind": "path", "displayName": "Port", "group": "common",
"label": "", "required": true, "type": "integer", "javaType": "int",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "configurationClass":
"org.apache.camel.component.grpc.GrpcConfiguration", "configurationField":
"configuration", "description": "The gRPC local or remote server port" },
"service": { "kind": "path", "displayName": "Service", "group": "common",
"label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.grpc.GrpcConfiguration", "configurationField":
"configuration", "description": "Fully qualified service name from the protocol
buffer descriptor file (package dot service definition name)" },
- "autoDiscoverClientInterceptors": { "kind": "parameter", "displayName":
"Auto Discover Client Interceptors", "group": "common", "label": "common",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration",
"configurationField": "configuration", "description": "Setting the
autoDiscoverClientInterceptors mechanism, if true, the com [...]
"flowControlWindow": { "kind": "parameter", "displayName": "Flow Control
Window", "group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 1048576, "configurationClass":
"org.apache.camel.component.grpc.GrpcConfiguration", "configurationField":
"configuration", "description": "The HTTP\/2 flow control window size (MiB)" },
"maxMessageSize": { "kind": "parameter", "displayName": "Max Message
Size", "group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 4194304, "configurationClass":
"org.apache.camel.component.grpc.GrpcConfiguration", "configurationField":
"configuration", "description": "The maximum message size allowed to be
received\/sent (MiB)" },
+ "autoDiscoverServerInterceptors": { "kind": "parameter", "displayName":
"Auto Discover Server Interceptors", "group": "consumer", "label": "consumer",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration",
"configurationField": "configuration", "description": "Setting the
autoDiscoverServerInterceptors mechanism, if true, the [...]
"bridgeErrorHandler": { "kind": "parameter", "displayName": "Bridge Error
Handler", "group": "consumer", "label": "consumer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "Allows for bridging the
consumer to the Camel routing Error Handler, which mean any exceptions occurred
while the consumer is trying to pickup incoming messages, or the likes, will
now be processed as a m [...]
"consumerStrategy": { "kind": "parameter", "displayName": "Consumer
Strategy", "group": "consumer", "label": "consumer", "required": false, "type":
"object", "javaType": "org.apache.camel.component.grpc.GrpcConsumerStrategy",
"enum": [ "AGGREGATION", "PROPAGATION" ], "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "PROPAGATION", "configurationClass":
"org.apache.camel.component.grpc.GrpcConfiguration", "configurationField":
"configuration", "description": "T [...]
"forwardOnCompleted": { "kind": "parameter", "displayName": "Forward On
Completed", "group": "consumer", "label": "consumer", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.grpc.GrpcConfiguration", "configurationField":
"configuration", "description": "Determines if onCompleted events should be
pushed to the Camel route." },
@@ -46,6 +46,7 @@
"routeControlledStreamObserver": { "kind": "parameter", "displayName":
"Route Controlled Stream Observer", "group": "consumer", "label": "consumer",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration",
"configurationField": "configuration", "description": "Lets the route to take
control over stream observer. If this value [...]
"exceptionHandler": { "kind": "parameter", "displayName": "Exception
Handler", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By default the
con [...]
"exchangePattern": { "kind": "parameter", "displayName": "Exchange
Pattern", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "object", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut",
"InOptionalOut" ], "deprecated": false, "autowired": false, "secret": false,
"description": "Sets the exchange pattern when the consumer creates an
exchange." },
+ "autoDiscoverClientInterceptors": { "kind": "parameter", "displayName":
"Auto Discover Client Interceptors", "group": "producer", "label": "producer",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration",
"configurationField": "configuration", "description": "Setting the
autoDiscoverClientInterceptors mechanism, if true, the [...]
"lazyStartProducer": { "kind": "parameter", "displayName": "Lazy Start
Producer", "group": "producer", "label": "producer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "Whether the producer
should be started lazy (on the first message). By starting lazy you can use
this to allow CamelContext and routes to startup in situations where a producer
may otherwise fail during sta [...]
"method": { "kind": "parameter", "displayName": "Method", "group":
"producer", "label": "producer", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.grpc.GrpcConfiguration", "configurationField":
"configuration", "description": "gRPC method name" },
"producerStrategy": { "kind": "parameter", "displayName": "Producer
Strategy", "group": "producer", "label": "producer", "required": false, "type":
"object", "javaType": "org.apache.camel.component.grpc.GrpcProducerStrategy",
"enum": [ "SIMPLE", "STREAMING" ], "deprecated": false, "autowired": false,
"secret": false, "defaultValue": "SIMPLE", "configurationClass":
"org.apache.camel.component.grpc.GrpcConfiguration", "configurationField":
"configuration", "description": "The mode used [...]
diff --git
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcComponent.java
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcComponent.java
index 5d691da1ea9..9279635e242 100644
---
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcComponent.java
+++
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcComponent.java
@@ -22,6 +22,7 @@ import java.util.Map;
import java.util.Set;
import io.grpc.ClientInterceptor;
+import io.grpc.ServerInterceptor;
import org.apache.camel.Endpoint;
import org.apache.camel.spi.annotations.Component;
import org.apache.camel.support.DefaultComponent;
@@ -39,6 +40,9 @@ public class GrpcComponent extends DefaultComponent {
if (config.isAutoDiscoverClientInterceptors()) {
checkAndSetRegistryClientInterceptors(config);
}
+ if (config.isAutoDiscoverServerInterceptors()) {
+ checkAndSetRegistryServerInterceptors(config);
+ }
return endpoint;
}
@@ -58,4 +62,11 @@ public class GrpcComponent extends DefaultComponent {
configuration.setClientInterceptors(new
ArrayList<>(clientInterceptors));
}
}
+
+ private void checkAndSetRegistryServerInterceptors(GrpcConfiguration
configuration) {
+ Set<ServerInterceptor> serverInterceptors =
getCamelContext().getRegistry().findByType(ServerInterceptor.class);
+ if (!serverInterceptors.isEmpty()) {
+ configuration.setServerInterceptors(new
ArrayList<>(serverInterceptors));
+ }
+ }
}
diff --git
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConfiguration.java
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConfiguration.java
index 422de2812ce..0ebcf9111be 100644
---
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConfiguration.java
+++
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConfiguration.java
@@ -21,6 +21,7 @@ import java.util.Collections;
import java.util.List;
import io.grpc.ClientInterceptor;
+import io.grpc.ServerInterceptor;
import io.grpc.internal.GrpcUtil;
import io.grpc.netty.NegotiationType;
import io.grpc.netty.NettyChannelBuilder;
@@ -111,9 +112,14 @@ public class GrpcConfiguration {
@UriParam(label = "consumer", defaultValue = "false")
private boolean routeControlledStreamObserver;
+ private List<ServerInterceptor> serverInterceptors =
Collections.emptyList();
+
+ @UriParam(label = "consumer", defaultValue = "true")
+ private boolean autoDiscoverServerInterceptors = true;
+
private List<ClientInterceptor> clientInterceptors =
Collections.emptyList();
- @UriParam(label = "common", defaultValue = "true")
+ @UriParam(label = "producer", defaultValue = "true")
private boolean autoDiscoverClientInterceptors = true;
@UriParam(defaultValue = "false", label = "advanced",
@@ -404,6 +410,30 @@ public class GrpcConfiguration {
this.maxConcurrentCallsPerConnection = maxConcurrentCallsPerConnection;
}
+ public List<ServerInterceptor> getServerInterceptors() {
+ return serverInterceptors;
+ }
+
+ /**
+ * Setting the server interceptors on the netty channel in order to
intercept incoming calls before they are
+ * received by the server.
+ */
+ public void setServerInterceptors(List<ServerInterceptor>
serverInterceptors) {
+ this.serverInterceptors = serverInterceptors;
+ }
+
+ public boolean isAutoDiscoverServerInterceptors() {
+ return autoDiscoverServerInterceptors;
+ }
+
+ /**
+ * Setting the autoDiscoverServerInterceptors mechanism, if true, the
component will look for a ServerInterceptor
+ * instance in the registry automatically otherwise it will skip that
checking.
+ */
+ public void setAutoDiscoverServerInterceptors(boolean
autoDiscoverServerInterceptors) {
+ this.autoDiscoverServerInterceptors = autoDiscoverServerInterceptors;
+ }
+
public List<ClientInterceptor> getClientInterceptors() {
return clientInterceptors;
}
diff --git
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConsumer.java
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConsumer.java
index c8b4e6943af..38ecabea11a 100644
---
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConsumer.java
+++
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConsumer.java
@@ -138,6 +138,10 @@ public class GrpcConsumer extends DefaultConsumer {
configuration.getJwtIssuer(),
configuration.getJwtSubject()));
}
+ for (ServerInterceptor si : configuration.getServerInterceptors()) {
+ serverBuilder.intercept(si);
+ }
+
server =
serverBuilder.addService(ServerInterceptors.intercept(bindableService,
headerInterceptor))
.maxInboundMessageSize(configuration.getMaxMessageSize())
.flowControlWindow(configuration.getFlowControlWindow())
diff --git
a/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcConsumerServerInterceptorTest.java
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcConsumerServerInterceptorTest.java
new file mode 100644
index 00000000000..e1309e01799
--- /dev/null
+++
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcConsumerServerInterceptorTest.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.component.grpc;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.Server;
+import io.grpc.stub.StreamObserver;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class GrpcConsumerServerInterceptorTest extends CamelTestSupport {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(GrpcConsumerServerInterceptorTest.class);
+
+ private static final int GRPC_REQUEST_INTERCEPT_TEST_PORT =
AvailablePortFinder.getNextAvailable();
+ private static final int GRPC_REQUEST_NO_INTERCEPT_TEST_PORT =
AvailablePortFinder.getNextAvailable();
+ private static final int GRPC_TEST_PING_ID = 1;
+ private static final String GRPC_TEST_PING_VALUE = "PING";
+ private static final String GRPC_TEST_PONG_VALUE = "PONG";
+
+ private static Server grpcServer;
+ private final GrpcMockServerInterceptor mockServerInterceptor =
mock(GrpcMockServerInterceptor.class);
+ private final GrpcMockServerInterceptor mockServerInterceptor2 =
mock(GrpcMockServerInterceptor.class);
+
+ private ManagedChannel interceptRequestChannel;
+ private ManagedChannel nointerceptRequestChannel;
+ private PingPongGrpc.PingPongBlockingStub interceptBlockingStub;
+ private PingPongGrpc.PingPongBlockingStub nointerceptBlockingStub;
+
+ @BeforeEach
+ public void startGrpcChannels() {
+ interceptRequestChannel =
ManagedChannelBuilder.forAddress("localhost",
GRPC_REQUEST_INTERCEPT_TEST_PORT).usePlaintext().build();
+ nointerceptRequestChannel =
ManagedChannelBuilder.forAddress("localhost",
GRPC_REQUEST_NO_INTERCEPT_TEST_PORT).usePlaintext().build();
+ interceptBlockingStub =
PingPongGrpc.newBlockingStub(interceptRequestChannel);
+ nointerceptBlockingStub =
PingPongGrpc.newBlockingStub(nointerceptRequestChannel);
+ }
+
+ @AfterEach
+ public void stopGrpcChannels() {
+ interceptRequestChannel.shutdown().shutdownNow();
+ nointerceptRequestChannel.shutdown().shutdownNow();
+ }
+
+ @Test
+ public void testServerInterceptors() throws Exception {
+ when(mockServerInterceptor.interceptCall(any(), any(),
any())).thenCallRealMethod();
+ when(mockServerInterceptor2.interceptCall(any(), any(),
any())).thenCallRealMethod();
+ LOG.info("gRPC pingSyncSync method blocking test start");
+ PingRequest pingRequest
+ =
PingRequest.newBuilder().setPingName(GRPC_TEST_PING_VALUE).setPingId(GRPC_TEST_PING_ID).build();
+ PongResponse pongResponse =
interceptBlockingStub.pingSyncSync(pingRequest);
+
+ assertNotNull(pongResponse);
+ assertEquals(GRPC_TEST_PING_ID, pongResponse.getPongId());
+ assertEquals(GRPC_TEST_PING_VALUE + GRPC_TEST_PONG_VALUE,
pongResponse.getPongName());
+ verify(mockServerInterceptor, times(1)).interceptCall(any(), any(),
any());
+ verify(mockServerInterceptor2, times(1)).interceptCall(any(), any(),
any());
+ }
+
+ @Test
+ public void testNoAutoDiscover() throws Exception {
+ when(mockServerInterceptor.interceptCall(any(), any(),
any())).thenCallRealMethod();
+ when(mockServerInterceptor2.interceptCall(any(), any(),
any())).thenCallRealMethod();
+ LOG.info("gRPC pingSyncSync method blocking test start");
+ PingRequest pingRequest
+ =
PingRequest.newBuilder().setPingName(GRPC_TEST_PING_VALUE).setPingId(GRPC_TEST_PING_ID).build();
+ PongResponse pongResponse =
nointerceptBlockingStub.pingSyncSync(pingRequest);
+
+ assertNotNull(pongResponse);
+ assertEquals(GRPC_TEST_PING_ID, pongResponse.getPongId());
+ assertEquals(GRPC_TEST_PING_VALUE + GRPC_TEST_PONG_VALUE,
pongResponse.getPongName());
+ verify(mockServerInterceptor, times(0)).interceptCall(any(), any(),
any());
+ verify(mockServerInterceptor2, times(0)).interceptCall(any(), any(),
any());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ context.getRegistry().bind("grpcMockServerInterceptor",
mockServerInterceptor);
+ context.getRegistry().bind("grpcMockServerInterceptor2",
mockServerInterceptor2);
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("grpc://localhost:" + GRPC_REQUEST_INTERCEPT_TEST_PORT
+ +
"/org.apache.camel.component.grpc.PingPong?synchronous=true&consumerStrategy=AGGREGATION")
+ .bean(new
GrpcConsumerServerInterceptorTest.GrpcMessageBuilder(), "buildPongResponse");
+
+ from("grpc://localhost:" + GRPC_REQUEST_NO_INTERCEPT_TEST_PORT
+ +
"/org.apache.camel.component.grpc.PingPong?synchronous=true&consumerStrategy=AGGREGATION"
+ + "&autoDiscoverServerInterceptors=false")
+ .bean(new
GrpcConsumerServerInterceptorTest.GrpcMessageBuilder(), "buildPongResponse");
+ }
+ };
+ }
+
+ public class PongResponseStreamObserver implements
StreamObserver<PongResponse> {
+ private PongResponse pongResponse;
+ private final CountDownLatch latch;
+
+ public PongResponseStreamObserver(CountDownLatch latch) {
+ this.latch = latch;
+ }
+
+ public PongResponse getPongResponse() {
+ return pongResponse;
+ }
+
+ @Override
+ public void onNext(PongResponse value) {
+ pongResponse = value;
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ LOG.info("Exception", t);
+ latch.countDown();
+ }
+
+ @Override
+ public void onCompleted() {
+ latch.countDown();
+ }
+ }
+
+ public class GrpcMessageBuilder {
+ public PongResponse buildPongResponse(PingRequest pingRequest) {
+ return
PongResponse.newBuilder().setPongName(pingRequest.getPingName() +
GRPC_TEST_PONG_VALUE)
+ .setPongId(pingRequest.getPingId()).build();
+ }
+
+ public PongResponse buildAsyncPongResponse(List<PingRequest>
pingRequests) {
+ return
PongResponse.newBuilder().setPongName(pingRequests.get(0).getPingName() +
GRPC_TEST_PONG_VALUE)
+ .setPongId(pingRequests.get(0).getPingId()).build();
+ }
+ }
+}
diff --git
a/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcMockServerInterceptor.java
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcMockServerInterceptor.java
new file mode 100644
index 00000000000..3fdd9b298f1
--- /dev/null
+++
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcMockServerInterceptor.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.component.grpc;
+
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCall.Listener;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GrpcMockServerInterceptor implements ServerInterceptor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(GrpcMockServerInterceptor.class);
+
+ @Override
+ public <ReqT, RespT> Listener<ReqT> interceptCall(
+ ServerCall<ReqT, RespT> call, Metadata headers,
+ ServerCallHandler<ReqT, RespT> next) {
+ LOG.info("Grpc Server Interceptor intercepted server call, forwarding
now.");
+ return next.startCall(call, headers);
+ }
+
+}
