Default to TlsChannelStreamFactoryFactory in async

when running on Java 8+ and SSL is enabled

JAVA-3038

Author: jyemin

driver-async/src/main/com/mongodb/async/client/MongoClients.java

@@ -17,13 +17,15 @@
 package com.mongodb.async.client;
 
 import com.mongodb.ConnectionString;
+import com.mongodb.MongoClientException;
 import com.mongodb.MongoDriverInformation;
-import com.mongodb.connection.AsynchronousSocketChannelStreamFactory;
+import com.mongodb.MongoInternalException;
+import com.mongodb.connection.AsynchronousSocketChannelStreamFactoryFactory;
 import com.mongodb.connection.Cluster;
 import com.mongodb.connection.DefaultClusterFactory;
-import com.mongodb.connection.SocketSettings;
 import com.mongodb.connection.StreamFactory;
 import com.mongodb.connection.StreamFactoryFactory;
+import com.mongodb.connection.TlsChannelStreamFactoryFactory;
 import com.mongodb.lang.Nullable;
 import org.bson.codecs.configuration.CodecRegistry;
 
@@ -38,7 +40,6 @@
  * @since 3.0
  * @deprecated Prefer the Reactive Streams-based asynchronous driver (mongodb-driver-reactivestreams artifactId)
  */
-@SuppressWarnings("deprecation")
 @Deprecated
 public final class MongoClients {
 
@@ -129,7 +130,6 @@ public static MongoClient create(final MongoClientSettings settings, @Nullable f
      * @throws IllegalArgumentException if the connection string's stream type is not one of "netty" or "nio2"
      * @see MongoClients#create(ConnectionString)
      */
-    @SuppressWarnings("deprecation")
     public static MongoClient create(final ConnectionString connectionString,
                                      @Nullable final MongoDriverInformation mongoDriverInformation) {
 
@@ -158,34 +158,40 @@ public static MongoClient create(final com.mongodb.MongoClientSettings settings)
      * @return the client
      * @since 3.7
      */
-    @SuppressWarnings("deprecation")
     public static MongoClient create(final com.mongodb.MongoClientSettings settings,
                                      @Nullable final MongoDriverInformation mongoDriverInformation) {
         return create(MongoClientSettings.createFromClientSettings(settings), mongoDriverInformation, null);
     }
 
-    @SuppressWarnings("deprecation")
     private static MongoClient create(final MongoClientSettings settings,
                                       @Nullable final MongoDriverInformation mongoDriverInformation,
                                       @Nullable final String requestedStreamType) {
         String streamType = getStreamType(requestedStreamType);
-        if (isNetty(streamType) && settings.getStreamFactoryFactory() == null) {
-            return NettyMongoClients.create(settings, mongoDriverInformation);
+        if (settings.getStreamFactoryFactory() == null) {
+           if (isNetty(streamType)) {
+               return NettyMongoClients.create(settings, mongoDriverInformation);
+           } else if (isNio(streamType)) {
+               if (settings.getSslSettings().isEnabled()) {
+                   return createWithTlsChannel(settings, mongoDriverInformation);
+               } else {
+                   return createWithAsynchronousSocketChannel(settings, mongoDriverInformation);
+               }
+           } else {
+               throw new IllegalArgumentException("Unsupported stream type: " + streamType);
+           }
         } else {
-            return new MongoClientImpl(settings, createCluster(settings, mongoDriverInformation,
-                    getStreamFactory(settings, streamType, false), getStreamFactory(settings, streamType, true)), (Closeable) null);
+            return createMongoClient(settings, mongoDriverInformation, getStreamFactory(settings, false),
+                    getStreamFactory(settings, true), null);
         }
     }
 
-    @SuppressWarnings("deprecation")
     static MongoClient createMongoClient(final MongoClientSettings settings, @Nullable final MongoDriverInformation mongoDriverInformation,
                                          final StreamFactory streamFactory, final StreamFactory heartbeatStreamFactory,
                                          @Nullable final Closeable externalResourceCloser) {
         return new MongoClientImpl(settings, createCluster(settings, mongoDriverInformation, streamFactory, heartbeatStreamFactory),
                 externalResourceCloser);
     }
 
-    @SuppressWarnings("deprecation")
     private static Cluster createCluster(final MongoClientSettings settings, @Nullable final MongoDriverInformation mongoDriverInformation,
                                          final StreamFactory streamFactory, final StreamFactory heartbeatStreamFactory) {
         notNull("settings", settings);
@@ -218,25 +224,57 @@ public static CodecRegistry getDefaultCodecRegistry() {
         return com.mongodb.MongoClientSettings.getDefaultCodecRegistry();
     }
 
-    @SuppressWarnings("deprecation")
-    private static StreamFactory getStreamFactory(final MongoClientSettings settings, final String streamType,
-                                                  final boolean isHeartbeat) {
+
+    private static MongoClient createWithTlsChannel(final MongoClientSettings settings,
+                                                    @Nullable final MongoDriverInformation mongoDriverInformation) {
+        if (!isJava8()) {
+            throw new MongoClientException("TLS is only supported natively with Java 8 and above. Please use Netty instead");
+        }
+        final TlsChannelStreamFactoryFactory streamFactoryFactory = new TlsChannelStreamFactoryFactory();
+        StreamFactory streamFactory = streamFactoryFactory.create(settings.getSocketSettings(), settings.getSslSettings());
+        StreamFactory heartbeatStreamFactory = streamFactoryFactory.create(settings.getHeartbeatSocketSettings(),
+                settings.getSslSettings());
+        return createMongoClient(settings, mongoDriverInformation, streamFactory, heartbeatStreamFactory,
+                new Closeable() {
+                    @Override
+                    public void close() {
+                        streamFactoryFactory.close();
+                    }
+                });
+    }
+
+    private static boolean isJava8() {
+        try {
+            Class.forName("java.time.Instant");
+            return true;
+        } catch (ClassNotFoundException e) {
+            return false;
+        }
+    }
+
+    private static MongoClient createWithAsynchronousSocketChannel(final MongoClientSettings settings,
+                                                                   @Nullable final MongoDriverInformation mongoDriverInformation) {
+        StreamFactoryFactory streamFactoryFactory = new AsynchronousSocketChannelStreamFactoryFactory();
+        StreamFactory streamFactory = streamFactoryFactory.create(settings.getSocketSettings(), settings.getSslSettings());
+        StreamFactory heartbeatStreamFactory = streamFactoryFactory.create(settings.getHeartbeatSocketSettings(),
+                settings.getSslSettings());
+        return createMongoClient(settings, mongoDriverInformation, streamFactory, heartbeatStreamFactory, null);
+    }
+
+    private static StreamFactory getStreamFactory(final MongoClientSettings settings, final boolean isHeartbeat) {
         StreamFactoryFactory streamFactoryFactory = settings.getStreamFactoryFactory();
-        SocketSettings socketSettings = isHeartbeat ? settings.getHeartbeatSocketSettings() : settings.getSocketSettings();
-        if (streamFactoryFactory != null) {
-            return streamFactoryFactory.create(socketSettings, settings.getSslSettings());
-        } else if (isNio2(streamType)) {
-            return new AsynchronousSocketChannelStreamFactory(socketSettings, settings.getSslSettings());
-        } else {
-            throw new IllegalArgumentException("Unsupported stream type: " + streamType);
+        if (streamFactoryFactory == null) {
+            throw new MongoInternalException("should not happen");
         }
+        return streamFactoryFactory.create(isHeartbeat ? settings.getHeartbeatSocketSettings() : settings.getSocketSettings(),
+                settings.getSslSettings());
     }
 
     private static boolean isNetty(final String streamType) {
         return streamType.toLowerCase().equals("netty");
     }
 
-    private static boolean isNio2(final String streamType) {
+    private static boolean isNio(final String streamType) {
         return streamType.toLowerCase().equals("nio2");
     }
 

driver-async/src/test/functional/com/mongodb/async/client/MongoClientsSpecification.groovy

@@ -18,14 +18,19 @@ package com.mongodb.async.client
 
 import com.mongodb.MongoCompressor
 import com.mongodb.MongoCredential
+import com.mongodb.MongoDriverInformation
 import com.mongodb.ReadConcern
 import com.mongodb.ServerAddress
 import com.mongodb.WriteConcern
-import com.mongodb.MongoDriverInformation
+import com.mongodb.async.FutureResultCallback
 import org.bson.Document
 import spock.lang.IgnoreIf
 import spock.lang.Unroll
 
+import static com.mongodb.ClusterFixture.connectionString
+import static com.mongodb.ClusterFixture.getCredentialList
+import static com.mongodb.ClusterFixture.getSslSettings
+import static com.mongodb.ClusterFixture.isNotAtLeastJava8
 import static com.mongodb.ClusterFixture.isStandalone
 import static com.mongodb.ClusterFixture.serverVersionAtLeast
 import static com.mongodb.ReadPreference.primary
@@ -36,6 +41,34 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS
 
 class MongoClientsSpecification extends FunctionalSpecification {
 
+    @IgnoreIf({ isNotAtLeastJava8() })
+    def 'should connect'() {
+        given:
+        def connectionString = 'mongodb://'
+        if (!getCredentialList().isEmpty()) {
+           connectionString += (getCredentialList()[0].getUserName() + ':' + String.valueOf(getCredentialList()[0].getPassword()) + '@')
+        }
+        connectionString += getConnectionString().getHosts()[0] + '/?'
+        connectionString += 'ssl=' + getSslSettings().isEnabled() + '&'
+        connectionString += 'sslInvalidHostNameAllowed=' + getSslSettings().isInvalidHostNameAllowed() + '&'
+        connectionString += 'streamType=' + streamType
+
+        when:
+        def client = MongoClients.create(connectionString)
+        def callback = new FutureResultCallback()
+        client.getDatabase('admin').runCommand(new Document('ping', 1), callback)
+        callback.get()
+
+        then:
+        noExceptionThrown()
+
+        cleanup:
+        client?.close()
+
+        where:
+        streamType << ['netty', 'nio2']
+    }
+
     def 'should apply connection string to cluster settings'() {
         when:
         def client = MongoClients.create('mongodb://localhost,localhost:27018/')