JAVA-1217: Instead of increasing the retry frequency for an unreachable server, always use the same frequency, and force an attempt to connect only when there is a failure of the server selector to find a matching server.

Author: jyemin

src/main/com/mongodb/BaseCluster.java

@@ -76,15 +76,13 @@ public Server getServer(final ServerSelector serverSelector, final long maxWaitT
                     }
                 }
 
-                if (!curDescription.isConnecting()) {
-                    throw new MongoServerSelectionException(format("Unable to connect to any server that matches %s", serverSelector));
-                }
-
                 final long timeout = endTime - System.nanoTime();
 
                 LOGGER.info(format("No server chosen by %s from cluster description %s. Waiting for %d ms before timing out",
                                    serverSelector, curDescription, MILLISECONDS.convert(timeout, NANOSECONDS)));
 
+                connect();
+
                 if (!currentPhase.await(timeout, NANOSECONDS)) {
                     throw new MongoTimeoutException(format("Timed out while waiting for a server that matches %s after %d ms",
                                                            serverSelector, MILLISECONDS.convert(timeout, NANOSECONDS)));
@@ -107,11 +105,6 @@ public ClusterDescription getDescription(final long maxWaitTime, final TimeUnit
             ClusterDescription curDescription = description;
             final long endTime = System.nanoTime() + NANOSECONDS.convert(maxWaitTime, timeUnit);
             while (curDescription.getType() == ClusterType.Unknown) {
-
-                if (!curDescription.isConnecting()) {
-                    throw new MongoServerSelectionException(format("Unable to connect to any servers"));
-                }
-
                 final long timeout = endTime - System.nanoTime();
 
                 LOGGER.info(format("Cluster description not yet available. Waiting for %d ms before timing out",
@@ -156,6 +149,11 @@ public boolean isClosed() {
      */
     protected abstract ClusterableServer getServer(final ServerAddress serverAddress);
 
+    /**
+     * Try to connect to all servers
+     */
+    protected abstract void connect();
+
     protected synchronized void updateDescription(final ClusterDescription newDescription) {
         LOGGER.fine(format("Updating cluster description to  %s", newDescription.getShortDescription()));
 

src/main/com/mongodb/ClusterDescription.java

@@ -25,7 +25,6 @@
 import java.util.Set;
 import java.util.TreeSet;
 
-import static com.mongodb.ServerConnectionState.Connecting;
 import static java.lang.String.format;
 import static org.bson.util.Assertions.notNull;
 
@@ -75,20 +74,6 @@ public ClusterType getType() {
         return type;
     }
 
-    /**
-     * Returns true if the application has been unsuccessful in its last attempt to connect to any of the servers in the cluster.
-     *
-     * @return true if connecting, false otherwise
-     */
-    public boolean isConnecting() {
-        for (final ServerDescription cur : all) {
-            if (cur.getState() == Connecting) {
-                return true;
-            }
-        }
-        return false;
-    }
-
     /**
      * Returns the Set of all server descriptions in this cluster, sorted by the String value of the ServerAddress of each one.
      *

src/main/com/mongodb/ClusterableServer.java

@@ -47,5 +47,10 @@ interface ClusterableServer extends Server {
      * * @return whether the server is closed
      */
     boolean isClosed();
+
+    /**
+     * Attempt to connect to the server.
+     */
+    void connect();
 }
 

src/main/com/mongodb/DefaultServer.java

@@ -18,65 +18,32 @@
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 
 import static com.mongodb.ServerConnectionState.Connecting;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.bson.util.Assertions.isTrue;
 import static org.bson.util.Assertions.notNull;
 
 class DefaultServer implements ClusterableServer {
-
-    private enum HeartbeatFrequency {
-        NORMAL {
-            @Override
-            long getFrequencyMS(final ServerSettings settings) {
-                return settings.getHeartbeatFrequency(MILLISECONDS);
-            }
-        },
-
-        RETRY {
-            @Override
-            long getFrequencyMS(final ServerSettings settings) {
-                return settings.getHeartbeatConnectRetryFrequency(MILLISECONDS);
-            }
-        };
-
-        abstract long getFrequencyMS(final ServerSettings settings);
-    }
-
-    private final String clusterId;
-    private final ScheduledExecutorService scheduledExecutorService;
     private final ServerAddress serverAddress;
-    private final ServerStateNotifier stateNotifier;
+    private final ServerMonitor serverMonitor;
     private final PooledConnectionProvider connectionProvider;
     private final Map<ChangeListener<ServerDescription>, Boolean> changeListeners =
     new ConcurrentHashMap<ChangeListener<ServerDescription>, Boolean>();
-    private final ServerSettings settings;
     private final ChangeListener<ServerDescription> serverStateListener;
     private volatile ServerDescription description;
     private volatile boolean isClosed;
 
-    private ScheduledFuture<?> scheduledFuture;
-    private HeartbeatFrequency currentFrequency;
-
     public DefaultServer(final ServerAddress serverAddress,
                          final ServerSettings settings,
                          final String clusterId, final PooledConnectionProvider connectionProvider,
                          final Mongo mongo) {
-        this.clusterId = notNull("clusterId", clusterId);
-        this.settings = notNull("settings", settings);
         this.serverAddress = notNull("serverAddress", serverAddress);
         this.description = ServerDescription.builder().state(Connecting).address(serverAddress).build();
         serverStateListener = new DefaultServerStateListener();
-        this.stateNotifier = new ServerStateNotifier(serverAddress, serverStateListener,
-                                                     settings.getHeartbeatSocketSettings(), mongo);
-        this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory());
-        setHeartbeat(0, HeartbeatFrequency.NORMAL);
+        this.serverMonitor = new ServerMonitor(serverAddress, serverStateListener,
+                                               settings.getHeartbeatSocketSettings(), settings, clusterId, mongo);
+        this.serverMonitor.start();
         this.connectionProvider = connectionProvider;
     }
 
@@ -107,16 +74,13 @@ public void invalidate() {
         serverStateListener.stateChanged(new ChangeEvent<ServerDescription>(description, ServerDescription.builder()
                                                                                                           .state(Connecting)
                                                                                                           .address(serverAddress).build()));
-        setHeartbeat(0, HeartbeatFrequency.RETRY);
         connectionProvider.invalidate();
     }
 
     @Override
     public void close() {
         if (!isClosed()) {
-            scheduledFuture.cancel(true);
-            scheduledExecutorService.shutdownNow();
-            stateNotifier.close();
+            serverMonitor.close();
             connectionProvider.close();
             isClosed = true;
         }
@@ -127,34 +91,9 @@ public boolean isClosed() {
         return isClosed;
     }
 
-    private void setHeartbeat(final ChangeEvent<ServerDescription> event) {
-        HeartbeatFrequency heartbeatFrequency = event.getNewValue().getState() == Connecting
-                                                ? HeartbeatFrequency.RETRY
-                                                : HeartbeatFrequency.NORMAL;
-        long initialDelay = heartbeatFrequency.getFrequencyMS(settings);
-        setHeartbeat(initialDelay, heartbeatFrequency);
-    }
-
-    private synchronized void setHeartbeat(final long initialDelay, final HeartbeatFrequency newFrequency) {
-        if (currentFrequency != newFrequency) {
-            currentFrequency = newFrequency;
-            if (scheduledFuture != null) {
-                scheduledFuture.cancel(false);
-            }
-            scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(stateNotifier, initialDelay,
-                                                                           newFrequency.getFrequencyMS(settings),
-                                                                           MILLISECONDS);
-        }
-    }
-
-    // Custom thread factory for scheduled executor service that creates daemon threads.  Otherwise,
-    // applications that neglect to close the MongoClient will not exit.
-    class DefaultThreadFactory implements ThreadFactory {
-        public Thread newThread(final Runnable runnable) {
-            Thread t = new Thread(runnable, "cluster-" + clusterId + "-" + serverAddress);
-            t.setDaemon(true);
-            return t;
-        }
+    @Override
+    public void connect() {
+        serverMonitor.connect();
     }
 
     private final class DefaultServerStateListener implements ChangeListener<ServerDescription> {
@@ -164,7 +103,6 @@ public void stateChanged(final ChangeEvent<ServerDescription> event) {
             for (ChangeListener<ServerDescription> listener : changeListeners.keySet()) {
                 listener.stateChanged(event);
             }
-            setHeartbeat(event);
         }
     }
 }

src/main/com/mongodb/MultiServerCluster.java

@@ -75,6 +75,13 @@ public MultiServerCluster(final String clusterId, final ClusterSettings settings
         fireChangeEvent();
     }
 
+    @Override
+    protected void connect() {
+        for (ServerTuple cur : addressToServerTupleMap.values()) {
+            cur.server.connect();
+        }
+    }
+
     @Override
     public void close() {
         if (!isClosed()) {