Remove routing table when it is stale for too long.

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/internal/async/pool/ConnectionFactory.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2002-2019 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * 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 org.neo4j.driver.internal.async.pool;
+
+import io.netty.channel.Channel;
+
+import org.neo4j.driver.internal.spi.Connection;
+
+public interface ConnectionFactory
+{
+    Connection createConnection( Channel channel, ExtendedChannelPool pool );
+}

driver/src/main/java/org/neo4j/driver/internal/async/pool/ConnectionPoolImpl.java

@@ -40,7 +40,6 @@
 import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.internal.BoltServerAddress;
 import org.neo4j.driver.internal.async.connection.ChannelConnector;
-import org.neo4j.driver.internal.async.NetworkConnection;
 import org.neo4j.driver.internal.metrics.ListenerEvent;
 import org.neo4j.driver.internal.metrics.MetricsListener;
 import org.neo4j.driver.internal.spi.Connection;
@@ -57,32 +56,33 @@ public class ConnectionPoolImpl implements ConnectionPool
     private final NettyChannelTracker nettyChannelTracker;
     private final NettyChannelHealthChecker channelHealthChecker;
     private final PoolSettings settings;
-    private final Clock clock;
     private final Logger log;
     private final MetricsListener metricsListener;
     private final boolean ownsEventLoopGroup;
 
     private final ConcurrentMap<BoltServerAddress,ExtendedChannelPool> pools = new ConcurrentHashMap<>();
     private final AtomicBoolean closed = new AtomicBoolean();
+    private final ConnectionFactory connectionFactory;
 
-    public ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, PoolSettings settings, MetricsListener metricsListener, Logging logging, Clock clock, boolean ownsEventLoopGroup )
+    public ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, PoolSettings settings, MetricsListener metricsListener, Logging logging,
+            Clock clock, boolean ownsEventLoopGroup )
     {
-
-        this( connector, bootstrap, new NettyChannelTracker( metricsListener, bootstrap.config().group().next(), logging ), settings, metricsListener, logging, clock, ownsEventLoopGroup );
+        this( connector, bootstrap, new NettyChannelTracker( metricsListener, bootstrap.config().group().next(), logging ), settings, metricsListener, logging,
+                clock, ownsEventLoopGroup, new NetworkConnectionFactory( clock, metricsListener ) );
     }
 
-    ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, NettyChannelTracker nettyChannelTracker,
-            PoolSettings settings, MetricsListener metricsListener, Logging logging, Clock clock, boolean ownsEventLoopGroup )
+    public ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, NettyChannelTracker nettyChannelTracker, PoolSettings settings,
+            MetricsListener metricsListener, Logging logging, Clock clock, boolean ownsEventLoopGroup, ConnectionFactory connectionFactory )
     {
         this.connector = connector;
         this.bootstrap = bootstrap;
         this.nettyChannelTracker = nettyChannelTracker;
         this.channelHealthChecker = new NettyChannelHealthChecker( settings, clock, logging );
         this.settings = settings;
         this.metricsListener = metricsListener;
-        this.clock = clock;
         this.log = logging.getLog( ConnectionPool.class.getSimpleName() );
         this.ownsEventLoopGroup = ownsEventLoopGroup;
+        this.connectionFactory = connectionFactory;
     }
 
     @Override
@@ -103,7 +103,7 @@ public CompletionStage<Connection> acquire( BoltServerAddress address )
             {
                 processAcquisitionError( pool, address, error );
                 assertNotClosed( address, channel, pool );
-                Connection connection = new NetworkConnection( channel, pool, clock, metricsListener );
+                Connection connection = connectionFactory.createConnection( channel, pool );
 
                 metricsListener.afterAcquiredOrCreated( address, acquireEvent );
                 return connection;

driver/src/main/java/org/neo4j/driver/internal/async/pool/NetworkConnectionFactory.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2002-2019 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * 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 org.neo4j.driver.internal.async.pool;
+
+import io.netty.channel.Channel;
+
+import org.neo4j.driver.internal.async.NetworkConnection;
+import org.neo4j.driver.internal.metrics.MetricsListener;
+import org.neo4j.driver.internal.spi.Connection;
+import org.neo4j.driver.internal.util.Clock;
+
+public class NetworkConnectionFactory implements ConnectionFactory
+{
+    private final Clock clock;
+    private final MetricsListener metricsListener;
+
+    public NetworkConnectionFactory( Clock clock, MetricsListener metricsListener )
+    {
+        this.clock = clock;
+        this.metricsListener = metricsListener;
+    }
+
+    @Override
+    public Connection createConnection( Channel channel, ExtendedChannelPool pool )
+    {
+        return new NetworkConnection( channel, pool, clock, metricsListener );
+    }
+}

driver/src/main/java/org/neo4j/driver/internal/cluster/ClusterComposition.java

@@ -21,24 +21,16 @@
 import java.util.LinkedHashSet;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Function;
 
-import org.neo4j.driver.internal.BoltServerAddress;
 import org.neo4j.driver.Record;
 import org.neo4j.driver.Value;
-import java.util.function.Function;
+import org.neo4j.driver.internal.BoltServerAddress;
 
 public final class ClusterComposition
 {
     private static final long MAX_TTL = Long.MAX_VALUE / 1000L;
-    private static final Function<Value,BoltServerAddress> OF_BoltServerAddress =
-            new Function<Value,BoltServerAddress>()
-            {
-                @Override
-                public BoltServerAddress apply( Value value )
-                {
-                    return new BoltServerAddress( value.asString() );
-                }
-            };
+    private static final Function<Value,BoltServerAddress> OF_BoltServerAddress = value -> new BoltServerAddress( value.asString() );
 
     private final Set<BoltServerAddress> readers;
     private final Set<BoltServerAddress> writers;

driver/src/main/java/org/neo4j/driver/internal/cluster/ClusterRoutingTable.java

@@ -67,6 +67,12 @@ public boolean isStaleFor( AccessMode mode )
                mode == AccessMode.WRITE && writers.size() == 0;
     }
 
+    @Override
+    public boolean isStale( long staleRoutingTableTimeout )
+    {
+        return expirationTimeout + staleRoutingTableTimeout < clock.millis();
+    }
+
     @Override
     public synchronized void update( ClusterComposition cluster )
     {