Unify behavior around configuration for replica set discovery

And fix handling of wrong set name in single topology

JAVA-3623 JAVA-3699

Author: rozza

driver-core/src/main/com/mongodb/ConnectionString.java

@@ -31,6 +31,7 @@
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
@@ -226,6 +227,7 @@
  * <li>{@code uuidRepresentation=unspecified|standard|javaLegacy|csharpLegacy|pythonLegacy}.  See
  * {@link MongoClientSettings#getUuidRepresentation()} for documentation of semantics of this parameter.  Defaults to "javaLegacy", but
  * will change to "unspecified" in the next major release.</li>
+ * <li>{@code directConnection=true|false}. If true the driver will set the connection to be a direct connection to the host.</li>
  * </ul>
  *
  * @mongodb.driver.manual reference/connection-string Connection String Format
@@ -247,6 +249,7 @@ public class ConnectionString {
     private final String collection;
     private final String connectionString;
 
+    private Boolean directConnection;
     private ReadPreference readPreference;
     private WriteConcern writeConcern;
     private Boolean retryWrites;
@@ -389,6 +392,15 @@ public ConnectionString(final String connectionString) {
             combinedOptionsMaps.put("ssl", singletonList("true"));
         }
         translateOptions(combinedOptionsMaps);
+
+        if (directConnection != null && directConnection) {
+            if (isSrvProtocol) {
+                throw new IllegalArgumentException("Direct connections are not supported when using mongodb+srv protocol");
+            } else if (hosts.size() > 1) {
+                throw new IllegalArgumentException("Direct connections are not supported when using multiple hosts");
+            }
+        }
+
         credential = createCredentials(combinedOptionsMaps, userName, password);
         warnOnUnsupportedOptions(combinedOptionsMaps);
     }
@@ -435,6 +447,8 @@ public ConnectionString(final String connectionString) {
 
         GENERAL_OPTIONS_KEYS.add("uuidrepresentation");
 
+        GENERAL_OPTIONS_KEYS.add("directconnection");
+
         COMPRESSOR_KEYS.add("compressors");
         COMPRESSOR_KEYS.add("zlibcompressionlevel");
 
@@ -535,6 +549,8 @@ private void translateOptions(final Map<String, List<String>> optionsMap) {
                 retryReads = parseBoolean(value, "retryreads");
             } else if (key.equals("uuidrepresentation")) {
                 uuidRepresentation = createUuidRepresentation(value);
+            } else if (key.equals("directconnection")) {
+                directConnection = parseBoolean(value, "directconnection");
             }
         }
 
@@ -1082,7 +1098,6 @@ public String getDatabase() {
         return database;
     }
 
-
     /**
      * Gets the collection name
      *
@@ -1093,6 +1108,17 @@ public String getCollection() {
         return collection;
     }
 
+    /**
+     * Indicates if the connection should be a direct connection
+     *
+     * @return true if a direct connection
+     * @since 4.1
+     */
+    @Nullable
+    public Boolean isDirectConnection() {
+        return directConnection;
+    }
+
     /**
      * Get the unparsed connection string.
      *
@@ -1337,91 +1363,45 @@ public boolean equals(final Object o) {
         if (this == o) {
             return true;
         }
-        if (!(o instanceof ConnectionString)) {
+        if (o == null || getClass() != o.getClass()) {
             return false;
         }
-
-        ConnectionString that = (ConnectionString) o;
-
-        if (collection != null ? !collection.equals(that.collection) : that.collection != null) {
-            return false;
-        }
-        if (connectTimeout != null ? !connectTimeout.equals(that.connectTimeout) : that.connectTimeout != null) {
-            return false;
-        }
-        if (credential != null ? !credential.equals(that.credential) : that.credential != null) {
-            return false;
-        }
-        if (database != null ? !database.equals(that.database) : that.database != null) {
-            return false;
-        }
-        if (!hosts.equals(that.hosts)) {
-            return false;
-        }
-        if (maxConnectionIdleTime != null ? !maxConnectionIdleTime.equals(that.maxConnectionIdleTime)
-                                          : that.maxConnectionIdleTime != null) {
-            return false;
-        }
-        if (maxConnectionLifeTime != null ? !maxConnectionLifeTime.equals(that.maxConnectionLifeTime)
-                                          : that.maxConnectionLifeTime != null) {
-            return false;
-        }
-        if (maxConnectionPoolSize != null ? !maxConnectionPoolSize.equals(that.maxConnectionPoolSize)
-                                          : that.maxConnectionPoolSize != null) {
-            return false;
-        }
-        if (maxWaitTime != null ? !maxWaitTime.equals(that.maxWaitTime) : that.maxWaitTime != null) {
-            return false;
-        }
-        if (minConnectionPoolSize != null ? !minConnectionPoolSize.equals(that.minConnectionPoolSize)
-                                          : that.minConnectionPoolSize != null) {
-            return false;
-        }
-        if (readPreference != null ? !readPreference.equals(that.readPreference) : that.readPreference != null) {
-            return false;
-        }
-        if (requiredReplicaSetName != null ? !requiredReplicaSetName.equals(that.requiredReplicaSetName)
-                                           : that.requiredReplicaSetName != null) {
-            return false;
-        }
-        if (socketTimeout != null ? !socketTimeout.equals(that.socketTimeout) : that.socketTimeout != null) {
-            return false;
-        }
-        if (sslEnabled != null ? !sslEnabled.equals(that.sslEnabled) : that.sslEnabled != null) {
-            return false;
-        }
-        if (writeConcern != null ? !writeConcern.equals(that.writeConcern) : that.writeConcern != null) {
-            return false;
-        }
-        if (applicationName != null ? !applicationName.equals(that.applicationName) : that.applicationName != null) {
-            return false;
-        }
-        if (!compressorList.equals(that.compressorList)) {
-            return false;
-        }
-
-        return true;
+        final ConnectionString that = (ConnectionString) o;
+        return isSrvProtocol == that.isSrvProtocol
+                && Objects.equals(directConnection, that.directConnection)
+                && Objects.equals(credential, that.credential)
+                && Objects.equals(hosts, that.hosts)
+                && Objects.equals(database, that.database)
+                && Objects.equals(collection, that.collection)
+                && Objects.equals(readPreference, that.readPreference)
+                && Objects.equals(writeConcern, that.writeConcern)
+                && Objects.equals(retryWrites, that.retryWrites)
+                && Objects.equals(retryReads, that.retryReads)
+                && Objects.equals(readConcern, that.readConcern)
+                && Objects.equals(minConnectionPoolSize, that.minConnectionPoolSize)
+                && Objects.equals(maxConnectionPoolSize, that.maxConnectionPoolSize)
+                && Objects.equals(maxWaitTime, that.maxWaitTime)
+                && Objects.equals(maxConnectionIdleTime, that.maxConnectionIdleTime)
+                && Objects.equals(maxConnectionLifeTime, that.maxConnectionLifeTime)
+                && Objects.equals(connectTimeout, that.connectTimeout)
+                && Objects.equals(socketTimeout, that.socketTimeout)
+                && Objects.equals(sslEnabled, that.sslEnabled)
+                && Objects.equals(sslInvalidHostnameAllowed, that.sslInvalidHostnameAllowed)
+                && Objects.equals(requiredReplicaSetName, that.requiredReplicaSetName)
+                && Objects.equals(serverSelectionTimeout, that.serverSelectionTimeout)
+                && Objects.equals(localThreshold, that.localThreshold)
+                && Objects.equals(heartbeatFrequency, that.heartbeatFrequency)
+                && Objects.equals(applicationName, that.applicationName)
+                && Objects.equals(compressorList, that.compressorList)
+                && Objects.equals(uuidRepresentation, that.uuidRepresentation);
     }
 
     @Override
     public int hashCode() {
-        int result = credential != null ? credential.hashCode() : 0;
-        result = 31 * result + hosts.hashCode();
-        result = 31 * result + (database != null ? database.hashCode() : 0);
-        result = 31 * result + (collection != null ? collection.hashCode() : 0);
-        result = 31 * result + (readPreference != null ? readPreference.hashCode() : 0);
-        result = 31 * result + (writeConcern != null ? writeConcern.hashCode() : 0);
-        result = 31 * result + (minConnectionPoolSize != null ? minConnectionPoolSize.hashCode() : 0);
-        result = 31 * result + (maxConnectionPoolSize != null ? maxConnectionPoolSize.hashCode() : 0);
-        result = 31 * result + (maxWaitTime != null ? maxWaitTime.hashCode() : 0);
-        result = 31 * result + (maxConnectionIdleTime != null ? maxConnectionIdleTime.hashCode() : 0);
-        result = 31 * result + (maxConnectionLifeTime != null ? maxConnectionLifeTime.hashCode() : 0);
-        result = 31 * result + (connectTimeout != null ? connectTimeout.hashCode() : 0);
-        result = 31 * result + (socketTimeout != null ? socketTimeout.hashCode() : 0);
-        result = 31 * result + (sslEnabled != null ? sslEnabled.hashCode() : 0);
-        result = 31 * result + (requiredReplicaSetName != null ? requiredReplicaSetName.hashCode() : 0);
-        result = 31 * result + (applicationName != null ? applicationName.hashCode() : 0);
-        result = 31 * result + compressorList.hashCode();
-        return result;
+        return Objects.hash(credential, isSrvProtocol, hosts, database, collection, directConnection, readPreference,
+                writeConcern, retryWrites, retryReads, readConcern, minConnectionPoolSize, maxConnectionPoolSize, maxWaitTime,
+                maxConnectionIdleTime, maxConnectionLifeTime, connectTimeout, socketTimeout, sslEnabled, sslInvalidHostnameAllowed,
+                requiredReplicaSetName, serverSelectionTimeout, localThreshold, heartbeatFrequency, applicationName, compressorList,
+                uuidRepresentation);
     }
 }

driver-core/src/main/com/mongodb/connection/ClusterSettings.java

@@ -22,6 +22,7 @@
 import com.mongodb.annotations.NotThreadSafe;
 import com.mongodb.event.ClusterListener;
 import com.mongodb.internal.connection.Cluster;
+import com.mongodb.internal.connection.ServerAddressHelper;
 import com.mongodb.internal.selector.LatencyMinimizingServerSelector;
 import com.mongodb.selector.CompositeServerSelector;
 import com.mongodb.selector.ServerSelector;
@@ -31,6 +32,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import static com.mongodb.assertions.Assertions.isTrueArgument;
 import static com.mongodb.assertions.Assertions.notNull;
@@ -262,25 +264,24 @@ public Builder addClusterListener(final ClusterListener clusterListener) {
          * @return this
          */
         public Builder applyConnectionString(final ConnectionString connectionString) {
+            Boolean directConnection = connectionString.isDirectConnection();
+
             if (connectionString.isSrvProtocol()) {
                 mode(ClusterConnectionMode.MULTIPLE);
                 srvHost(connectionString.getHosts().get(0));
-            }
-            else if (connectionString.getHosts().size() == 1 && connectionString.getRequiredReplicaSetName() == null) {
+            } else if ((directConnection != null && directConnection)
+                    || (directConnection == null && connectionString.getHosts().size() == 1
+                        && connectionString.getRequiredReplicaSetName() == null)) {
                 mode(ClusterConnectionMode.SINGLE)
-                .hosts(singletonList(createServerAddress(connectionString.getHosts().get(0))));
+                        .hosts(singletonList(createServerAddress(connectionString.getHosts().get(0))));
             } else {
-                List<ServerAddress> seedList = new ArrayList<ServerAddress>();
-                for (final String cur : connectionString.getHosts()) {
-                    seedList.add(createServerAddress(cur));
-                }
+                List<ServerAddress> seedList = connectionString.getHosts().stream()
+                        .map(ServerAddressHelper::createServerAddress)
+                        .collect(Collectors.toList());
                 mode(ClusterConnectionMode.MULTIPLE).hosts(seedList);
             }
             requiredReplicaSetName(connectionString.getRequiredReplicaSetName());
 
-            Integer maxConnectionPoolSize = connectionString.getMaxConnectionPoolSize();
-            int maxSize = maxConnectionPoolSize != null ? maxConnectionPoolSize : 100;
-
             Integer serverSelectionTimeout = connectionString.getServerSelectionTimeout();
             if (serverSelectionTimeout != null) {
                 serverSelectionTimeout(serverSelectionTimeout, MILLISECONDS);
@@ -562,7 +563,6 @@ private ClusterSettings(final Builder builder) {
             if (builder.mode == ClusterConnectionMode.SINGLE && builder.hosts.size() > 1) {
                 throw new IllegalArgumentException("Can not directly connect to more than one server");
             }
-
             mode = builder.mode != null ? builder.mode : hosts.size() == 1 ? ClusterConnectionMode.SINGLE : ClusterConnectionMode.MULTIPLE;
         }
         requiredReplicaSetName = builder.requiredReplicaSetName;

driver-core/src/main/com/mongodb/connection/ServerDescription.java

@@ -101,6 +101,17 @@ public static Builder builder() {
         return new Builder();
     }
 
+    /**
+     * Creates a Builder for a ServerDescription instance based on a previous serverDescription.
+     *
+     * @param serverDescription the ServerDescription to base the builder from
+     * @return a new Builder for ServerDescription.
+     * @since 4.1
+     */
+    public static Builder builder(final ServerDescription serverDescription) {
+        return new Builder(serverDescription);
+    }
+
     /**
      * Gets the string representing the host name and port that this member of a replica set was configured with,
      * e.g. {@code "somehost:27019"}. This is typically derived from the "me" field from the "isMaster" command response.
@@ -150,6 +161,33 @@ public static class Builder {
 
         private Throwable exception;
 
+        Builder() {
+        }
+
+        Builder(final ServerDescription serverDescription) {
+            this.address = serverDescription.address;
+            this.type = serverDescription.type;
+            this.canonicalAddress = serverDescription.canonicalAddress;
+            this.hosts = serverDescription.hosts;
+            this.passives = serverDescription.passives;
+            this.arbiters = serverDescription.arbiters;
+            this.primary = serverDescription.primary;
+            this.maxDocumentSize = serverDescription.maxDocumentSize;
+            this.tagSet = serverDescription.tagSet;
+            this.setName = serverDescription.setName;
+            this.roundTripTimeNanos = serverDescription.roundTripTimeNanos;
+            this.ok = serverDescription.ok;
+            this.state = serverDescription.state;
+            this.minWireVersion = serverDescription.minWireVersion;
+            this.maxWireVersion = serverDescription.maxWireVersion;
+            this.electionId = serverDescription.electionId;
+            this.setVersion = serverDescription.setVersion;
+            this.lastWriteDate = serverDescription.lastWriteDate;
+            this.lastUpdateTimeNanos = serverDescription.lastUpdateTimeNanos;
+            this.logicalSessionTimeoutMinutes = serverDescription.logicalSessionTimeoutMinutes;
+            this.exception = serverDescription.exception;
+        }
+
         /**
          * Sets the address of the server.
          *

driver-core/src/main/com/mongodb/internal/connection/AbstractMultiServerCluster.java

@@ -197,7 +197,7 @@ private void onChange(final ServerDescriptionChangedEvent event) {
                 return;
             }
 
-            if (event.getNewDescription().isOk()) {
+            if (newDescription.isOk()) {
                 if (clusterType == UNKNOWN && newDescription.getType() != REPLICA_SET_GHOST) {
                     clusterType = newDescription.getClusterType();
                     if (LOGGER.isInfoEnabled()) {