Merge branch 'master' into feat/add_http_header_size_config

Author: ST-DDT

grpc-client-spring-boot-autoconfigure/src/main/java/net/devh/boot/grpc/client/autoconfigure/GrpcClientAutoConfiguration.java

@@ -46,6 +46,9 @@
 import net.devh.boot.grpc.client.interceptor.AnnotationGlobalClientInterceptorConfigurer;
 import net.devh.boot.grpc.client.interceptor.GlobalClientInterceptorRegistry;
 import net.devh.boot.grpc.client.nameresolver.NameResolverRegistration;
+import net.devh.boot.grpc.client.stubfactory.AsyncStubFactory;
+import net.devh.boot.grpc.client.stubfactory.BlockingStubFactory;
+import net.devh.boot.grpc.client.stubfactory.FutureStubFactory;
 import net.devh.boot.grpc.common.autoconfigure.GrpcCommonCodecAutoConfiguration;
 
 /**
@@ -65,6 +68,21 @@ static GrpcClientBeanPostProcessor grpcClientBeanPostProcessor(final Application
         return new GrpcClientBeanPostProcessor(applicationContext);
     }
 
+    @Bean
+    AsyncStubFactory asyncStubFactory() {
+        return new AsyncStubFactory();
+    }
+
+    @Bean
+    BlockingStubFactory blockingStubFactory() {
+        return new BlockingStubFactory();
+    }
+
+    @Bean
+    FutureStubFactory futureStubFactory() {
+        return new FutureStubFactory();
+    }
+
     @ConditionalOnMissingBean
     @Bean
     GrpcChannelsProperties grpcChannelsProperties() {

grpc-client-spring-boot-autoconfigure/src/main/java/net/devh/boot/grpc/client/channelfactory/AbstractChannelFactory.java

@@ -19,10 +19,12 @@
 
 import static java.util.Objects.requireNonNull;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import javax.annotation.PreDestroy;
@@ -132,6 +134,10 @@ protected ManagedChannel newManagedChannel(final String name) {
         final T builder = newChannelBuilder(name);
         configure(builder, name);
         final ManagedChannel channel = builder.build();
+        final Duration timeout = properties.getChannel(name).getImmediateConnectTimeout();
+        if (!timeout.isZero()) {
+            connectOnStartup(name, channel, timeout);
+        }
         watchConnectivityState(name, channel);
         return channel;
     }
@@ -254,6 +260,36 @@ protected void watchConnectivityState(final String name, final ManagedChannel ch
         }
     }
 
+    private void connectOnStartup(String name, ManagedChannel channel, Duration timeout) {
+        log.debug("Initiating connection to channel {}", name);
+        channel.getState(true);
+
+        final CountDownLatch readyLatch = new CountDownLatch(1);
+        waitForReady(channel, readyLatch);
+        boolean connected;
+        try {
+            log.debug("Waiting for connection to channel {}", name);
+            connected = !readyLatch.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            connected = false;
+        }
+        if (connected) {
+            throw new IllegalStateException("Can't connect to channel " + name);
+        }
+        log.info("Successfully connected to channel {}", name);
+    }
+
+    private void waitForReady(ManagedChannel channel, CountDownLatch readySignal) {
+        final ConnectivityState state = channel.getState(false);
+        log.debug("Waiting for ready state. Currently in {}", state);
+        if (state == ConnectivityState.READY) {
+            readySignal.countDown();
+        } else {
+            channel.notifyWhenStateChanged(state, () -> waitForReady(channel, readySignal));
+        }
+    }
+
     /**
      * Closes this channel factory and the channels created by this instance. The shutdown happens in two phases, first
      * an orderly shutdown is initiated on all channels and then the method waits for all channels to terminate. If the

grpc-client-spring-boot-autoconfigure/src/main/java/net/devh/boot/grpc/client/config/GrpcChannelProperties.java

@@ -361,6 +361,38 @@ public void setNegotiationType(final NegotiationType negotiationType) {
 
     // --------------------------------------------------
 
+    private Duration immediateConnectTimeout;
+    private static final Duration DEFAULT_IMMEDIATE_CONNECT = Duration.ZERO;
+
+    /**
+     * Get the connection timeout at application startup.
+     *
+     * @return connection timeout at application startup.
+     *
+     * @see #setImmediateConnectTimeout(Duration)
+     */
+    public Duration getImmediateConnectTimeout() {
+        return this.immediateConnectTimeout == null ? DEFAULT_IMMEDIATE_CONNECT : this.immediateConnectTimeout;
+    }
+
+    /**
+     * If set to a positive duration instructs a client to connect to GRPC-endpoint when GRPC stub is created. If it's
+     * set to a positive timeout application startup will be slower due to connection process will be executed
+     * synchronously with maximum to connection timeout. If connection fails stub will fail to create with an exception
+     * which in turn causes context startup to If connection fails stub will fail to create with an exception which in
+     * turn causes context fail. Defaults to false.
+     *
+     * @param immediateConnectTimeout Connection timeout at application startup.
+     */
+    public void setImmediateConnectTimeout(final Duration immediateConnectTimeout) {
+        if (immediateConnectTimeout.isNegative()) {
+            throw new IllegalArgumentException("Timeout can't be negative");
+        }
+        this.immediateConnectTimeout = immediateConnectTimeout;
+    }
+
+    // --------------------------------------------------
+
     private final Security security = new Security();
 
     /**
@@ -409,6 +441,9 @@ public void copyDefaultsFrom(final GrpcChannelProperties config) {
         if (this.negotiationType == null) {
             this.negotiationType = config.negotiationType;
         }
+        if (this.immediateConnectTimeout == null) {
+            this.immediateConnectTimeout = config.immediateConnectTimeout;
+        }
         this.security.copyDefaultsFrom(config.security);
     }
 

grpc-client-spring-boot-autoconfigure/src/main/java/net/devh/boot/grpc/client/inject/GrpcClientBeanPostProcessor.java

@@ -19,7 +19,6 @@
 
 import static java.util.Objects.requireNonNull;
 
-import java.lang.reflect.Constructor;
 import java.lang.reflect.Field;
 import java.lang.reflect.Member;
 import java.lang.reflect.Method;
@@ -41,12 +40,10 @@
 
 import io.grpc.Channel;
 import io.grpc.ClientInterceptor;
-import io.grpc.stub.AbstractAsyncStub;
-import io.grpc.stub.AbstractBlockingStub;
-import io.grpc.stub.AbstractFutureStub;
 import io.grpc.stub.AbstractStub;
 import net.devh.boot.grpc.client.channelfactory.GrpcChannelFactory;
 import net.devh.boot.grpc.client.nameresolver.NameResolverRegistration;
+import net.devh.boot.grpc.client.stubfactory.StubFactory;
 
 /**
  * This {@link BeanPostProcessor} searches for fields and methods in beans that are annotated with {@link GrpcClient}
@@ -63,6 +60,7 @@ public class GrpcClientBeanPostProcessor implements BeanPostProcessor {
     // which could lead to problems with the correct bean setup.
     private GrpcChannelFactory channelFactory = null;
     private List<StubTransformer> stubTransformers = null;
+    private List<StubFactory> stubFactories = null;
 
     /**
      * Creates a new GrpcClientBeanPostProcessor with the given ApplicationContext.
@@ -217,7 +215,8 @@ protected <T> T valueForMember(final String name, final Member injectionTarget,
         } else if (AbstractStub.class.isAssignableFrom(injectionType)) {
 
             @SuppressWarnings("unchecked") // Eclipse incorrectly marks this as not required
-            AbstractStub<?> stub = createStub(injectionType.asSubclass(AbstractStub.class), channel);
+            AbstractStub<?> stub = createStub(
+                    (Class<? extends AbstractStub<?>>) injectionType.asSubclass(AbstractStub.class), channel);
             for (final StubTransformer stubTransformer : getStubTransformers()) {
                 stub = stubTransformer.transform(name, stub);
             }
@@ -229,53 +228,38 @@ protected <T> T valueForMember(final String name, final Member injectionTarget,
     }
 
     /**
-     * Creates a stub of the given type.
-     *
-     * @param <T> The type of the instance to be injected.
-     * @param stubType The type of the stub to create.
-     * @param channel The channel used to create the stub.
-     * @return The newly created stub.
-     *
-     * @throws BeanInstantiationException If the stub couldn't be created.
+     * Creates a stub instance for the specified stub type using the resolved {@link StubFactory}.
+     * 
+     * @param stubClass The stub class that needs to be created.
+     * @param channel The gRPC channel associated with the created stub, passed as a parameter to the stub factory.
+     * @throws BeanInstantiationException If the stub couldn't be created, either because the type isn't supported or
+     *         because of a failure in creation.
+     * @return A newly created gRPC stub.
      */
-    protected <T extends AbstractStub<T>> T createStub(final Class<T> stubType, final Channel channel) {
+    private AbstractStub<?> createStub(Class<? extends AbstractStub<?>> stubClass, Channel channel) {
+        final StubFactory factory = getStubFactories().stream()
+                .filter(stubFactory -> stubFactory.isApplicable(stubClass))
+                .findFirst()
+                .orElseThrow(() -> new BeanInstantiationException(stubClass,
+                        "Unsupported stub type: " + stubClass.getName() + " -> Please report this issue."));
+
         try {
-            // First try the public static factory method
-            final String methodName = deriveStubFactoryMethodName(stubType);
-            final Class<?> enclosingClass = stubType.getEnclosingClass();
-            final Method factoryMethod = enclosingClass.getMethod(methodName, Channel.class);
-            return stubType.cast(factoryMethod.invoke(null, channel));
-        } catch (final Exception e) {
-            try {
-                // Use the private constructor as backup
-                final Constructor<T> constructor = stubType.getDeclaredConstructor(Channel.class);
-                constructor.setAccessible(true);
-                return constructor.newInstance(channel);
-            } catch (final Exception e1) {
-                e.addSuppressed(e1);
-            }
-            throw new BeanInstantiationException(stubType, "Failed to create gRPC client", e);
+            return factory.createStub(stubClass, channel);
+        } catch (Exception exception) {
+            throw new BeanInstantiationException(stubClass, "Failed to create gRPC stub of type " + stubClass.getName(),
+                    exception);
         }
     }
 
     /**
-     * Derives the name of the factory method from the given stub type.
+     * Lazy getter for the list of defined {@link StubFactory} beans.
      *
-     * @param stubType The type of the stub to get it for.
-     * @return The name of the factory method.
-     * @throws IllegalArgumentException If the method was called with an unsupported stub type.
+     * @return A list of all defined {@link StubFactory} beans.
      */
-    protected String deriveStubFactoryMethodName(final Class<? extends AbstractStub<?>> stubType) {
-        if (AbstractAsyncStub.class.isAssignableFrom(stubType)) {
-            return "newStub";
-        } else if (AbstractBlockingStub.class.isAssignableFrom(stubType)) {
-            return "newBlockingStub";
-        } else if (AbstractFutureStub.class.isAssignableFrom(stubType)) {
-            return "newFutureStub";
-        } else {
-            throw new IllegalArgumentException(
-                    "Unsupported stub type: " + stubType.getName() + " -> Please report this issue.");
+    private List<StubFactory> getStubFactories() {
+        if (this.stubFactories == null) {
+            stubFactories = new ArrayList<>(applicationContext.getBeansOfType(StubFactory.class).values());
         }
+        return stubFactories;
     }
-
 }

grpc-client-spring-boot-autoconfigure/src/main/java/net/devh/boot/grpc/client/stubfactory/AsyncStubFactory.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2016-2020 Michael Zhang <[email protected]>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+ * WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package net.devh.boot.grpc.client.stubfactory;
+
+import io.grpc.stub.AbstractAsyncStub;
+import io.grpc.stub.AbstractStub;
+
+public class AsyncStubFactory extends StandardJavaGrpcStubFactory {
+
+    @Override
+    public boolean isApplicable(Class<? extends AbstractStub<?>> stubType) {
+        return AbstractAsyncStub.class.isAssignableFrom(stubType);
+    }
+
+    @Override
+    protected String getFactoryMethodName() {
+        return "newStub";
+    }
+}

grpc-client-spring-boot-autoconfigure/src/main/java/net/devh/boot/grpc/client/stubfactory/BlockingStubFactory.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2016-2020 Michael Zhang <[email protected]>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+ * WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package net.devh.boot.grpc.client.stubfactory;
+
+import io.grpc.stub.AbstractBlockingStub;
+import io.grpc.stub.AbstractStub;
+
+public class BlockingStubFactory extends StandardJavaGrpcStubFactory {
+
+    @Override
+    public boolean isApplicable(Class<? extends AbstractStub<?>> stubType) {
+        return AbstractBlockingStub.class.isAssignableFrom(stubType);
+    }
+
+    @Override
+    protected String getFactoryMethodName() {
+        return "newBlockingStub";
+    }
+}

grpc-client-spring-boot-autoconfigure/src/main/java/net/devh/boot/grpc/client/stubfactory/FutureStubFactory.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2016-2020 Michael Zhang <[email protected]>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+ * WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package net.devh.boot.grpc.client.stubfactory;
+
+import io.grpc.stub.AbstractFutureStub;
+import io.grpc.stub.AbstractStub;
+
+public class FutureStubFactory extends StandardJavaGrpcStubFactory {
+
+    @Override
+    public boolean isApplicable(Class<? extends AbstractStub<?>> stubType) {
+        return AbstractFutureStub.class.isAssignableFrom(stubType);
+    }
+
+    @Override
+    protected String getFactoryMethodName() {
+        return "newFutureStub";
+    }
+}