Merge branch 'master' into master

Author: LeoBarykin

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

@@ -17,12 +17,16 @@
 
 package net.devh.boot.grpc.client.channelfactory;
 
+import static java.util.Comparator.comparingLong;
 import static java.util.Objects.requireNonNull;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -134,7 +138,7 @@ 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();
+        final Duration timeout = this.properties.getChannel(name).getImmediateConnectTimeout();
         if (!timeout.isZero()) {
             connectOnStartup(name, channel, timeout);
         }
@@ -260,7 +264,7 @@ protected void watchConnectivityState(final String name, final ManagedChannel ch
         }
     }
 
-    private void connectOnStartup(String name, ManagedChannel channel, Duration timeout) {
+    private void connectOnStartup(final String name, final ManagedChannel channel, final Duration timeout) {
         log.debug("Initiating connection to channel {}", name);
         channel.getState(true);
 
@@ -270,7 +274,7 @@ private void connectOnStartup(String name, ManagedChannel channel, Duration time
         try {
             log.debug("Waiting for connection to channel {}", name);
             connected = !readyLatch.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
-        } catch (InterruptedException e) {
+        } catch (final InterruptedException e) {
             Thread.currentThread().interrupt();
             connected = false;
         }
@@ -280,7 +284,7 @@ private void connectOnStartup(String name, ManagedChannel channel, Duration time
         log.info("Successfully connected to channel {}", name);
     }
 
-    private void waitForReady(ManagedChannel channel, CountDownLatch readySignal) {
+    private void waitForReady(final ManagedChannel channel, final CountDownLatch readySignal) {
         final ConnectivityState state = channel.getState(false);
         log.debug("Waiting for ready state. Currently in {}", state);
         if (state == ConnectivityState.READY) {
@@ -302,16 +306,30 @@ public synchronized void close() {
             return;
         }
         this.shutdown = true;
-        for (final ManagedChannel channel : this.channels.values()) {
+        final List<ShutdownRecord> shutdownEntries = new ArrayList<>();
+        for (final Entry<String, ManagedChannel> entry : this.channels.entrySet()) {
+            final ManagedChannel channel = entry.getValue();
             channel.shutdown();
+            final long gracePeriod = this.properties.getChannel(entry.getKey()).getShutdownGracePeriod().toMillis();
+            shutdownEntries.add(new ShutdownRecord(entry.getKey(), channel, gracePeriod));
         }
         try {
-            final long waitLimit = System.currentTimeMillis() + 60_000; // wait 60 seconds at max
-            for (final ManagedChannel channel : this.channels.values()) {
-                int i = 0;
-                do {
-                    log.debug("Awaiting channel shutdown: {} ({}s)", channel, i++);
-                } while (System.currentTimeMillis() < waitLimit && !channel.awaitTermination(1, TimeUnit.SECONDS));
+            final long start = System.currentTimeMillis();
+            shutdownEntries.sort(comparingLong(ShutdownRecord::getGracePeriod));
+
+            for (final ShutdownRecord entry : shutdownEntries) {
+                if (!entry.channel.isTerminated()) {
+                    log.debug("Awaiting channel termination: {}", entry.name);
+
+                    final long waitedTime = System.currentTimeMillis() - start;
+                    final long waitTime = entry.gracePeriod - waitedTime;
+
+                    if (waitTime > 0) {
+                        entry.channel.awaitTermination(waitTime, MILLISECONDS);
+                    }
+                    entry.channel.shutdownNow();
+                }
+                log.debug("Completed channel termination: {}", entry.name);
             }
         } catch (final InterruptedException e) {
             Thread.currentThread().interrupt();
@@ -327,7 +345,26 @@ public synchronized void close() {
         final int channelCount = this.channels.size();
         this.channels.clear();
         this.channelStates.clear();
-        log.debug("GrpcCannelFactory closed (including {} channels)", channelCount);
+        log.debug("GrpcChannelFactory closed (including {} channels)", channelCount);
+    }
+
+    private static class ShutdownRecord {
+
+        private final String name;
+        private final ManagedChannel channel;
+        private final long gracePeriod;
+
+        public ShutdownRecord(final String name, final ManagedChannel channel, final long gracePeriod) {
+            this.name = name;
+            this.channel = channel;
+            // gracePeriod < 0 => Infinite
+            this.gracePeriod = gracePeriod < 0 ? Long.MAX_VALUE : gracePeriod;
+        }
+
+        long getGracePeriod() {
+            return this.gracePeriod;
+        }
+
     }
 
 }

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

@@ -263,6 +263,33 @@ public void setKeepAliveWithoutCalls(final Boolean keepAliveWithoutCalls) {
         this.keepAliveWithoutCalls = keepAliveWithoutCalls;
     }
 
+    // --------------------------------------------------
+
+    @DurationUnit(ChronoUnit.SECONDS)
+    private Duration shutdownGracePeriod;
+    private static final Duration DEFAULT_SHUTDOWN_GRACE_PERIOD = Duration.ofSeconds(30);
+
+    /**
+     * Gets the time to wait for the channel to gracefully shutdown. If set to a negative value, the channel waits
+     * forever. If set to {@code 0} the channel will force shutdown immediately. Defaults to {@code 30s}.
+     *
+     * @return The time to wait for a graceful shutdown.
+     */
+    public Duration getShutdownGracePeriod() {
+        return this.shutdownGracePeriod == null ? DEFAULT_SHUTDOWN_GRACE_PERIOD : this.shutdownGracePeriod;
+    }
+
+    /**
+     * Sets the time to wait for the channel to gracefully shutdown (completing all requests). If set to a negative
+     * value, the channel waits forever. If set to {@code 0} the channel will force shutdown immediately. Defaults to
+     * {@code 30s}.
+     *
+     * @param shutdownGracePeriod The time to wait for a graceful shutdown.
+     */
+    public void setShutdownGracePeriod(final Duration shutdownGracePeriod) {
+        this.shutdownGracePeriod = shutdownGracePeriod;
+    }
+
     // --------------------------------------------------
     // Message Transfer
     // --------------------------------------------------
@@ -432,6 +459,9 @@ public void copyDefaultsFrom(final GrpcChannelProperties config) {
         if (this.keepAliveWithoutCalls == null) {
             this.keepAliveWithoutCalls = config.keepAliveWithoutCalls;
         }
+        if (this.shutdownGracePeriod == null) {
+            this.shutdownGracePeriod = config.shutdownGracePeriod;
+        }
         if (this.maxInboundMessageSize == null) {
             this.maxInboundMessageSize = config.maxInboundMessageSize;
         }

grpc-client-spring-boot-autoconfigure/src/test/java/net/devh/boot/grpc/client/config/GrpcChannelPropertiesGivenUnitTest.java

@@ -29,7 +29,7 @@
 import org.springframework.util.unit.DataSize;
 
 /**
- * Tests whether the property resolution using suffixes works.
+ * Tests whether the property resolution works when using suffixes.
  */
 @ExtendWith(SpringExtension.class)
 @SpringBootTest(properties = {

grpc-client-spring-boot-autoconfigure/src/test/java/net/devh/boot/grpc/client/config/GrpcChannelPropertiesNegativeGivenUnitTest.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2016-2021 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.config;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+/**
+ * Tests whether the property resolution works with negative values and when using suffixes.
+ */
+@ExtendWith(SpringExtension.class)
+@SpringBootTest(properties = {
+        "grpc.client.test.shutdownGracePeriod=-1ms"
+})
+class GrpcChannelPropertiesNegativeGivenUnitTest {
+
+    @Autowired
+    private GrpcChannelsProperties grpcChannelsProperties;
+
+    @Test
+    void test() {
+        final GrpcChannelProperties properties = this.grpcChannelsProperties.getChannel("test");
+        assertEquals(Duration.ofMillis(-1), properties.getShutdownGracePeriod());
+    }
+
+}

grpc-client-spring-boot-autoconfigure/src/test/java/net/devh/boot/grpc/client/config/GrpcChannelPropertiesNegativeNoUnitTest.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2016-2021 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.config;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+/**
+ * Tests whether the property resolution works with negative values and without suffixes.
+ */
+@ExtendWith(SpringExtension.class)
+@SpringBootTest(properties = {
+        "grpc.client.test.shutdownGracePeriod=-1"
+})
+class GrpcChannelPropertiesNegativeNoUnitTest {
+
+    @Autowired
+    private GrpcChannelsProperties grpcChannelsProperties;
+
+    @Test
+    void test() {
+        final GrpcChannelProperties properties = this.grpcChannelsProperties.getChannel("test");
+        assertEquals(Duration.ofSeconds(-1), properties.getShutdownGracePeriod());
+    }
+
+}

grpc-client-spring-boot-autoconfigure/src/test/java/net/devh/boot/grpc/client/config/GrpcChannelPropertiesNoUnitTest.java

@@ -29,7 +29,7 @@
 import org.springframework.util.unit.DataSize;
 
 /**
- * Tests whether the property resolution without suffixes works (backwards compatibility).
+ * Tests whether the property resolution works without suffixes (backwards compatibility).
  */
 @ExtendWith(SpringExtension.class)
 @SpringBootTest(properties = {

grpc-server-spring-boot-autoconfigure/src/main/java/net/devh/boot/grpc/server/autoconfigure/GrpcServerAutoConfiguration.java

@@ -135,8 +135,8 @@ public List<GrpcServerConfigurer> defaultServerConfigurers() {
     @ConditionalOnMissingBean
     @ConditionalOnBean(GrpcServerFactory.class)
     @Bean
-    public GrpcServerLifecycle grpcServerLifecycle(final GrpcServerFactory factory) {
-        return new GrpcServerLifecycle(factory);
+    public GrpcServerLifecycle grpcServerLifecycle(final GrpcServerFactory factory, GrpcServerProperties properties) {
+        return new GrpcServerLifecycle(factory, properties.getShutdownGracePeriod());
     }
 
 }

grpc-server-spring-boot-autoconfigure/src/main/java/net/devh/boot/grpc/server/autoconfigure/GrpcServerFactoryAutoConfiguration.java

@@ -82,12 +82,16 @@ public ShadedNettyGrpcServerFactory shadedNettyGrpcServerFactory(
      * The server lifecycle bean for a shaded netty based server.
      *
      * @param factory The factory used to create the lifecycle.
+     * @param properties The server properties to use.
      * @return The inter-process server lifecycle bean.
      */
     @ConditionalOnBean(ShadedNettyGrpcServerFactory.class)
     @Bean
-    public GrpcServerLifecycle shadedNettyGrpcServerLifecycle(final ShadedNettyGrpcServerFactory factory) {
-        return new GrpcServerLifecycle(factory);
+    public GrpcServerLifecycle shadedNettyGrpcServerLifecycle(
+            final ShadedNettyGrpcServerFactory factory,
+            final GrpcServerProperties properties) {
+
+        return new GrpcServerLifecycle(factory, properties.getShutdownGracePeriod());
     }
 
     // Then try the normal netty server
@@ -120,12 +124,16 @@ public NettyGrpcServerFactory nettyGrpcServerFactory(
      * The server lifecycle bean for netty based server.
      *
      * @param factory The factory used to create the lifecycle.
+     * @param properties The server properties to use.
      * @return The inter-process server lifecycle bean.
      */
     @ConditionalOnBean(NettyGrpcServerFactory.class)
     @Bean
-    public GrpcServerLifecycle nettyGrpcServerLifecycle(final NettyGrpcServerFactory factory) {
-        return new GrpcServerLifecycle(factory);
+    public GrpcServerLifecycle nettyGrpcServerLifecycle(
+            final NettyGrpcServerFactory factory,
+            final GrpcServerProperties properties) {
+
+        return new GrpcServerLifecycle(factory, properties.getShutdownGracePeriod());
     }
 
     /**
@@ -153,12 +161,16 @@ public InProcessGrpcServerFactory inProcessGrpcServerFactory(
      * The server lifecycle bean for the in-process-server.
      *
      * @param factory The factory used to create the lifecycle.
+     * @param properties The server properties to use.
      * @return The in-process server lifecycle bean.
      */
     @ConditionalOnBean(InProcessGrpcServerFactory.class)
     @Bean
-    public GrpcServerLifecycle inProcessGrpcServerLifecycle(final InProcessGrpcServerFactory factory) {
-        return new GrpcServerLifecycle(factory);
+    public GrpcServerLifecycle inProcessGrpcServerLifecycle(
+            final InProcessGrpcServerFactory factory,
+            final GrpcServerProperties properties) {
+
+        return new GrpcServerLifecycle(factory, properties.getShutdownGracePeriod());
     }
 
 }

grpc-server-spring-boot-autoconfigure/src/main/java/net/devh/boot/grpc/server/config/GrpcServerProperties.java

@@ -93,6 +93,17 @@ public class GrpcServerProperties {
      */
     private String inProcessName;
 
+    /**
+     * The time to wait for the server to gracefully shutdown (completing all requests after the server started to
+     * shutdown). If set to a negative value, the server waits forever. If set to {@code 0} the server will force
+     * shutdown immediately. Defaults to {@code 30s}.
+     * 
+     * @param gracefullShutdownTimeout The time to wait for a graceful shutdown.
+     * @return The time to wait for a graceful shutdown.
+     */
+    @DurationUnit(ChronoUnit.SECONDS)
+    private Duration shutdownGracePeriod = Duration.of(30, ChronoUnit.SECONDS);
+
     /**
      * Setting to enable keepAlive. Default to {@code false}.
      *