netty,okhttp: Allow keepalive without calls

Again, the server can enforce this, so clients shouldn't go wild with
their new-found freedom.

netty/src/main/java/io/grpc/netty/NettyChannelBuilder.java

@@ -88,6 +88,7 @@ public final class NettyChannelBuilder
   private int maxHeaderListSize = GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE;
   private long keepAliveTimeNanos = KEEPALIVE_TIME_NANOS_DISABLED;
   private long keepAliveTimeoutNanos = DEFAULT_KEEPALIVE_TIMEOUT_NANOS;
+  private boolean keepAliveWithoutCalls;
   private TransportCreationParamsFilterFactory dynamicParamsFactory;
 
   /**
@@ -308,12 +309,28 @@ public final class NettyChannelBuilder
     return this;
   }
 
+  /**
+   * Sets whether keepalive will be performed when there are no outstanding RPC on a connection.
+   * Defaults to {@code false}.
+   *
+   * <p>Clients must receive permission from the service owner before enabling this option.
+   * Keepalives on unused connections can easilly accidentally consume a considerable amount of
+   * bandwidth and CPU.
+   *
+   * @since 1.3.0
+   * @see #keepAliveTime(long, TimeUnit)
+   */
+  public NettyChannelBuilder keepAliveWithoutCalls(boolean enable) {
+    keepAliveWithoutCalls = enable;
+    return this;
+  }
+
   @Override
   @CheckReturnValue
   protected ClientTransportFactory buildTransportFactory() {
     return new NettyTransportFactory(dynamicParamsFactory, channelType, channelOptions,
         negotiationType, sslContext, eventLoopGroup, flowControlWindow, maxInboundMessageSize(),
-        maxHeaderListSize, keepAliveTimeNanos, keepAliveTimeoutNanos);
+        maxHeaderListSize, keepAliveTimeNanos, keepAliveTimeoutNanos, keepAliveWithoutCalls);
   }
 
   @Override
@@ -436,6 +453,7 @@ public final class NettyChannelBuilder
     private final int maxHeaderListSize;
     private final long keepAliveDelayNanos;
     private final long keepAliveTimeoutNanos;
+    private final boolean keepAliveWithoutCalls;
 
     private boolean closed;
 
@@ -443,7 +461,7 @@ public final class NettyChannelBuilder
         Class<? extends Channel> channelType, Map<ChannelOption<?>, ?> channelOptions,
         NegotiationType negotiationType, SslContext sslContext, EventLoopGroup group,
         int flowControlWindow, int maxMessageSize, int maxHeaderListSize,
-        long keepAliveDelayNanos, long keepAliveTimeoutNanos) {
+        long keepAliveDelayNanos, long keepAliveTimeoutNanos, boolean keepAliveWithoutCalls) {
       this.channelType = channelType;
       this.negotiationType = negotiationType;
       this.channelOptions = new HashMap<ChannelOption<?>, Object>(channelOptions);
@@ -465,6 +483,7 @@ public final class NettyChannelBuilder
       this.maxHeaderListSize = maxHeaderListSize;
       this.keepAliveDelayNanos = keepAliveDelayNanos;
       this.keepAliveTimeoutNanos = keepAliveTimeoutNanos;
+      this.keepAliveWithoutCalls = keepAliveWithoutCalls;
       usingSharedGroup = group == null;
       if (usingSharedGroup) {
         // The group was unspecified, using the shared group.
@@ -486,7 +505,7 @@ public final class NettyChannelBuilder
           dparams.getTargetServerAddress(), channelType, channelOptions, group,
           dparams.getProtocolNegotiator(), flowControlWindow,
           maxMessageSize, maxHeaderListSize, keepAliveDelayNanos, keepAliveTimeoutNanos,
-          dparams.getAuthority(), dparams.getUserAgent());
+          keepAliveWithoutCalls, dparams.getAuthority(), dparams.getUserAgent());
       return transport;
     }
 

netty/src/main/java/io/grpc/netty/NettyClientTransport.java

@@ -84,8 +84,9 @@ class NettyClientTransport implements ConnectionClientTransport {
   private final int maxMessageSize;
   private final int maxHeaderListSize;
   private KeepAliveManager keepAliveManager;
-  private long keepAliveDelayNanos;
-  private long keepAliveTimeoutNanos;
+  private final long keepAliveDelayNanos;
+  private final long keepAliveTimeoutNanos;
+  private final boolean keepAliveWithoutCalls;
 
   private ProtocolNegotiator.Handler negotiationHandler;
   private NettyClientHandler handler;
@@ -102,7 +103,7 @@ class NettyClientTransport implements ConnectionClientTransport {
       Map<ChannelOption<?>, ?> channelOptions, EventLoopGroup group,
       ProtocolNegotiator negotiator, int flowControlWindow, int maxMessageSize,
       int maxHeaderListSize, long keepAliveDelayNanos, long keepAliveTimeoutNanos,
-      String authority, @Nullable String userAgent) {
+      boolean keepAliveWithoutCalls, String authority, @Nullable String userAgent) {
     this.negotiator = Preconditions.checkNotNull(negotiator, "negotiator");
     this.address = Preconditions.checkNotNull(address, "address");
     this.group = Preconditions.checkNotNull(group, "group");
@@ -113,6 +114,7 @@ class NettyClientTransport implements ConnectionClientTransport {
     this.maxHeaderListSize = maxHeaderListSize;
     this.keepAliveDelayNanos = keepAliveDelayNanos;
     this.keepAliveTimeoutNanos = keepAliveTimeoutNanos;
+    this.keepAliveWithoutCalls = keepAliveWithoutCalls;
     this.authority = new AsciiString(authority);
     this.userAgent = new AsciiString(GrpcUtil.getGrpcUserAgent("netty", userAgent));
   }
@@ -178,7 +180,7 @@ class NettyClientTransport implements ConnectionClientTransport {
     if (keepAliveDelayNanos != KEEPALIVE_TIME_NANOS_DISABLED) {
       keepAliveManager = new KeepAliveManager(
           new ClientKeepAlivePinger(this), eventLoop, keepAliveDelayNanos, keepAliveTimeoutNanos,
-          false);
+          keepAliveWithoutCalls);
     }
 
     handler = NettyClientHandler.newHandler(lifecycleManager, keepAliveManager, flowControlWindow,

netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java

@@ -173,7 +173,7 @@ public class NettyClientTransportTest {
     NettyClientTransport transport = new NettyClientTransport(
         address, NioSocketChannel.class, channelOptions, group, newNegotiator(),
         DEFAULT_WINDOW_SIZE, DEFAULT_MAX_MESSAGE_SIZE, GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE,
-        KEEPALIVE_TIME_NANOS_DISABLED, 1L, authority, null /* user agent */);
+        KEEPALIVE_TIME_NANOS_DISABLED, 1L, false, authority, null /* user agent */);
     transports.add(transport);
     callMeMaybe(transport.start(clientTransportListener));
 
@@ -299,7 +299,8 @@ public class NettyClientTransportTest {
     NettyClientTransport transport = new NettyClientTransport(
         address, CantConstructChannel.class, new HashMap<ChannelOption<?>, Object>(), group,
         newNegotiator(), DEFAULT_WINDOW_SIZE, DEFAULT_MAX_MESSAGE_SIZE,
-        GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE, KEEPALIVE_TIME_NANOS_DISABLED, 1, authority, null);
+        GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE, KEEPALIVE_TIME_NANOS_DISABLED, 1, false, authority,
+        null);
     transports.add(transport);
 
     // Should not throw
@@ -464,7 +465,7 @@ public class NettyClientTransportTest {
     }
     NettyClientTransport transport = new NettyClientTransport(
         address, NioSocketChannel.class, new HashMap<ChannelOption<?>, Object>(), group, negotiator,
-        DEFAULT_WINDOW_SIZE, maxMsgSize, maxHeaderListSize, keepAliveTimeNano, 1L, authority,
+        DEFAULT_WINDOW_SIZE, maxMsgSize, maxHeaderListSize, keepAliveTimeNano, 1L, false, authority,
         userAgent);
     transports.add(transport);
     return transport;

okhttp/src/main/java/io/grpc/okhttp/OkHttpChannelBuilder.java

@@ -120,6 +120,7 @@ public class OkHttpChannelBuilder extends
   private NegotiationType negotiationType = NegotiationType.TLS;
   private long keepAliveTimeNanos = KEEPALIVE_TIME_NANOS_DISABLED;
   private long keepAliveTimeoutNanos = DEFAULT_KEEPALIVE_TIMEOUT_NANOS;
+  private boolean keepAliveWithoutCalls;
 
   protected OkHttpChannelBuilder(String host, int port) {
     this(GrpcUtil.authorityFromHostAndPort(host, port));
@@ -222,6 +223,22 @@ public class OkHttpChannelBuilder extends
     return this;
   }
 
+  /**
+   * Sets whether keepalive will be performed when there are no outstanding RPC on a connection.
+   * Defaults to {@code false}.
+   *
+   * <p>Clients must receive permission from the service owner before enabling this option.
+   * Keepalives on unused connections can easilly accidentally consume a considerable amount of
+   * bandwidth and CPU.
+   *
+   * @since 1.3.0
+   * @see #keepAliveTime(long, TimeUnit)
+   */
+  public OkHttpChannelBuilder keepAliveWithoutCalls(boolean enable) {
+    keepAliveWithoutCalls = enable;
+    return this;
+  }
+
   /**
    * Override the default {@link SSLSocketFactory} and enable {@link NegotiationType#TLS}
    * negotiation.
@@ -272,7 +289,7 @@ public class OkHttpChannelBuilder extends
     boolean enableKeepAlive = keepAliveTimeNanos != KEEPALIVE_TIME_NANOS_DISABLED;
     return new OkHttpTransportFactory(transportExecutor,
         createSocketFactory(), connectionSpec, maxInboundMessageSize(), enableKeepAlive,
-        keepAliveTimeNanos, keepAliveTimeoutNanos);
+        keepAliveTimeNanos, keepAliveTimeoutNanos, keepAliveWithoutCalls);
   }
 
   @Override
@@ -342,9 +359,10 @@ public class OkHttpChannelBuilder extends
     private final SSLSocketFactory socketFactory;
     private final ConnectionSpec connectionSpec;
     private final int maxMessageSize;
-    private boolean enableKeepAlive;
-    private long keepAliveDelayNanos;
-    private long keepAliveTimeoutNanos;
+    private final boolean enableKeepAlive;
+    private final long keepAliveDelayNanos;
+    private final long keepAliveTimeoutNanos;
+    private final boolean keepAliveWithoutCalls;
     private boolean closed;
 
     private OkHttpTransportFactory(Executor executor,
@@ -353,13 +371,15 @@ public class OkHttpChannelBuilder extends
         int maxMessageSize,
         boolean enableKeepAlive,
         long keepAliveDelayNanos,
-        long keepAliveTimeoutNanos) {
+        long keepAliveTimeoutNanos,
+        boolean keepAliveWithoutCalls) {
       this.socketFactory = socketFactory;
       this.connectionSpec = connectionSpec;
       this.maxMessageSize = maxMessageSize;
       this.enableKeepAlive = enableKeepAlive;
       this.keepAliveDelayNanos = keepAliveDelayNanos;
       this.keepAliveTimeoutNanos = keepAliveTimeoutNanos;
+      this.keepAliveWithoutCalls = keepAliveWithoutCalls;
 
       usingSharedExecutor = executor == null;
       if (usingSharedExecutor) {
@@ -391,7 +411,8 @@ public class OkHttpChannelBuilder extends
           userAgent, executor, socketFactory, Utils.convertSpec(connectionSpec), maxMessageSize,
           proxyAddress, null, null);
       if (enableKeepAlive) {
-        transport.enableKeepAlive(true, keepAliveDelayNanos, keepAliveTimeoutNanos);
+        transport.enableKeepAlive(
+            true, keepAliveDelayNanos, keepAliveTimeoutNanos, keepAliveWithoutCalls);
       }
       return transport;
     }

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java

@@ -183,6 +183,7 @@ class OkHttpClientTransport implements ConnectionClientTransport {
   private boolean enableKeepAlive;
   private long keepAliveDelayNanos;
   private long keepAliveTimeoutNanos;
+  private boolean keepAliveWithoutCalls;
   @Nullable
   private final InetSocketAddress proxyAddress;
   @Nullable
@@ -246,10 +247,11 @@ class OkHttpClientTransport implements ConnectionClientTransport {
    * Enable keepalive with custom delay and timeout.
    */
   void enableKeepAlive(boolean enable, long keepAliveDelayNanos,
-      long keepAliveTimeoutNanos) {
+      long keepAliveTimeoutNanos, boolean keepAliveWithoutCalls) {
     enableKeepAlive = enable;
     this.keepAliveDelayNanos = keepAliveDelayNanos;
     this.keepAliveTimeoutNanos = keepAliveTimeoutNanos;
+    this.keepAliveWithoutCalls = keepAliveWithoutCalls;
   }
 
   private boolean isForTest() {
@@ -372,7 +374,7 @@ class OkHttpClientTransport implements ConnectionClientTransport {
       scheduler = SharedResourceHolder.get(TIMER_SERVICE);
       keepAliveManager = new KeepAliveManager(
           new ClientKeepAlivePinger(this), scheduler, keepAliveDelayNanos, keepAliveTimeoutNanos,
-          false);
+          keepAliveWithoutCalls);
       keepAliveManager.onTransportStarted();
     }