Add AsyncTransportSettings, ExecutorService

Author: katcharov

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

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 com.mongodb.connection;
+
+import com.mongodb.lang.Nullable;
+
+import java.util.concurrent.ExecutorService;
+
+/**
+ * {@link TransportSettings} for an async transport implementation.
+ *
+ * @since 5.2
+ */
+public class AsyncTransportSettings extends TransportSettings {
+
+    private final ExecutorService executorService;
+
+    public AsyncTransportSettings(final Builder builder) {
+        this.executorService = builder.executorService;
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    /**
+     * A builder for an instance of {@link AsyncTransportSettings}
+     */
+    public static final class Builder {
+
+        private ExecutorService executorService;
+
+        private Builder() {
+        }
+
+        /**
+         * Sets the executor service
+         *
+         * @param executorService the executor service
+         * @return this
+         * @see #getExecutorService()
+         */
+        public Builder executorService(final ExecutorService executorService) {
+            this.executorService = executorService;
+            return this;
+        }
+
+        /**
+         * Build an instance of {@link AsyncTransportSettings}
+         * @return an instance of {@link AsyncTransportSettings}
+         */
+        public AsyncTransportSettings build() {
+            return new AsyncTransportSettings(this);
+        }
+    }
+
+    /**
+     * Gets the executor service
+     *
+     * @return the executor service
+     * @see Builder#executorService(ExecutorService)
+     */
+    @Nullable
+    public ExecutorService getExecutorService() {
+        return executorService;
+    }
+}

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

@@ -137,7 +137,7 @@ public Builder sslContext(final SslContext sslContext) {
         /**
          * Build an instance of {@code NettyTransportSettings}.
          *
-         * @return factory for {@code NettyTransportSettings}
+         * @return an instance of {@code NettyTransportSettings}
          */
         public NettyTransportSettings build() {
             return new NettyTransportSettings(this);

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

@@ -35,4 +35,14 @@ public abstract class TransportSettings {
     public static NettyTransportSettings.Builder nettyBuilder() {
         return NettyTransportSettings.builder();
     }
+
+    /**
+     * A builder for {@link AsyncTransportSettings}.
+     *
+     * @return a builder for {@link AsyncTransportSettings}
+     * @since 5.2
+     */
+    public static AsyncTransportSettings.Builder asyncBuilder() {
+        return AsyncTransportSettings.builder();
+    }
 }

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

@@ -28,10 +28,12 @@
 import java.net.SocketAddress;
 import java.net.StandardSocketOptions;
 import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousChannelGroup;
 import java.nio.channels.AsynchronousSocketChannel;
 import java.nio.channels.CompletionHandler;
 import java.util.LinkedList;
 import java.util.Queue;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -46,13 +48,24 @@ public final class AsynchronousSocketChannelStream extends AsynchronousChannelSt
     private final ServerAddress serverAddress;
     private final InetAddressResolver inetAddressResolver;
     private final SocketSettings settings;
+    @Nullable
+    private final ExecutorService executorService;
 
-    public AsynchronousSocketChannelStream(final ServerAddress serverAddress, final InetAddressResolver inetAddressResolver,
+    public AsynchronousSocketChannelStream(
+            final ServerAddress serverAddress, final InetAddressResolver inetAddressResolver,
             final SocketSettings settings, final PowerOfTwoBufferPool bufferProvider) {
+        this(serverAddress, inetAddressResolver, settings, bufferProvider, null);
+    }
+
+    public AsynchronousSocketChannelStream(
+            final ServerAddress serverAddress, final InetAddressResolver inetAddressResolver,
+            final SocketSettings settings, final PowerOfTwoBufferPool bufferProvider,
+            @Nullable final ExecutorService executorService) {
         super(serverAddress, settings, bufferProvider);
         this.serverAddress = serverAddress;
         this.inetAddressResolver = inetAddressResolver;
         this.settings = settings;
+        this.executorService = executorService;
     }
 
     @Override
@@ -77,7 +90,13 @@ private void initializeSocketChannel(final AsyncCompletionHandler<Void> handler,
             SocketAddress socketAddress = socketAddressQueue.poll();
 
             try {
-                AsynchronousSocketChannel attemptConnectionChannel = AsynchronousSocketChannel.open();
+                AsynchronousSocketChannel attemptConnectionChannel;
+                if (executorService != null) {
+                    AsynchronousChannelGroup group = AsynchronousChannelGroup.withThreadPool(executorService);
+                    attemptConnectionChannel = AsynchronousSocketChannel.open(group);
+                } else {
+                    attemptConnectionChannel = AsynchronousSocketChannel.open();
+                }
                 attemptConnectionChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
                 attemptConnectionChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
                 if (settings.getReceiveBufferSize() > 0) {

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

@@ -19,8 +19,11 @@
 import com.mongodb.ServerAddress;
 import com.mongodb.connection.SocketSettings;
 import com.mongodb.connection.SslSettings;
+import com.mongodb.lang.Nullable;
 import com.mongodb.spi.dns.InetAddressResolver;
 
+import java.util.concurrent.ExecutorService;
+
 import static com.mongodb.assertions.Assertions.assertFalse;
 import static com.mongodb.assertions.Assertions.notNull;
 
@@ -31,23 +34,34 @@ public class AsynchronousSocketChannelStreamFactory implements StreamFactory {
     private final PowerOfTwoBufferPool bufferProvider = PowerOfTwoBufferPool.DEFAULT;
     private final SocketSettings settings;
     private final InetAddressResolver inetAddressResolver;
+    @Nullable
+    private final ExecutorService executorService;
 
     /**
      * Create a new factory with the default {@code BufferProvider} and {@code AsynchronousChannelGroup}.
      *
      * @param settings    the settings for the connection to a MongoDB server
      * @param sslSettings the settings for connecting via SSL
      */
-    public AsynchronousSocketChannelStreamFactory(final InetAddressResolver inetAddressResolver, final SocketSettings settings,
+    public AsynchronousSocketChannelStreamFactory(
+            final InetAddressResolver inetAddressResolver, final SocketSettings settings,
             final SslSettings sslSettings) {
+        this(inetAddressResolver, settings, sslSettings, null);
+    }
+
+    public AsynchronousSocketChannelStreamFactory(
+            final InetAddressResolver inetAddressResolver, final SocketSettings settings,
+            final SslSettings sslSettings, @Nullable final ExecutorService executorService) {
         assertFalse(sslSettings.isEnabled());
         this.inetAddressResolver = inetAddressResolver;
         this.settings = notNull("settings", settings);
+        this.executorService = executorService;
     }
 
     @Override
     public Stream create(final ServerAddress serverAddress) {
-        return new AsynchronousSocketChannelStream(serverAddress, inetAddressResolver, settings, bufferProvider);
+        return new AsynchronousSocketChannelStream(
+                serverAddress, inetAddressResolver, settings, bufferProvider, executorService);
     }
 
 }

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

@@ -18,23 +18,36 @@
 
 import com.mongodb.connection.SocketSettings;
 import com.mongodb.connection.SslSettings;
+import com.mongodb.lang.Nullable;
 import com.mongodb.spi.dns.InetAddressResolver;
 
+import java.util.concurrent.ExecutorService;
+
 /**
  * A {@code StreamFactoryFactory} implementation for AsynchronousSocketChannel-based streams.
  *
  * @see java.nio.channels.AsynchronousSocketChannel
  */
 public final class AsynchronousSocketChannelStreamFactoryFactory implements StreamFactoryFactory {
     private final InetAddressResolver inetAddressResolver;
+    @Nullable
+    private final ExecutorService executorService;
 
     public AsynchronousSocketChannelStreamFactoryFactory(final InetAddressResolver inetAddressResolver) {
+        this(inetAddressResolver, null);
+    }
+
+    public AsynchronousSocketChannelStreamFactoryFactory(
+            final InetAddressResolver inetAddressResolver,
+            @Nullable final ExecutorService executorService) {
         this.inetAddressResolver = inetAddressResolver;
+        this.executorService = executorService;
     }
 
     @Override
     public StreamFactory create(final SocketSettings socketSettings, final SslSettings sslSettings) {
-        return new AsynchronousSocketChannelStreamFactory(inetAddressResolver, socketSettings, sslSettings);
+        return new AsynchronousSocketChannelStreamFactory(
+                inetAddressResolver, socketSettings, sslSettings, executorService);
     }
 
     @Override

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

@@ -17,26 +17,63 @@
 package com.mongodb.internal.connection;
 
 import com.mongodb.MongoClientException;
+import com.mongodb.MongoClientSettings;
+import com.mongodb.connection.AsyncTransportSettings;
 import com.mongodb.connection.NettyTransportSettings;
+import com.mongodb.connection.SocketSettings;
 import com.mongodb.connection.TransportSettings;
 import com.mongodb.internal.connection.netty.NettyStreamFactoryFactory;
 import com.mongodb.spi.dns.InetAddressResolver;
 
+import java.util.concurrent.ExecutorService;
+
 /**
  * <p>This class is not part of the public API and may be removed or changed at any time</p>
  */
 public final class StreamFactoryHelper {
-    public static StreamFactoryFactory getStreamFactoryFactoryFromSettings(final TransportSettings transportSettings,
+
+    public static StreamFactory getSyncStreamFactory(final MongoClientSettings settings,
+            final InetAddressResolver inetAddressResolver, final SocketSettings socketSettings) {
+        TransportSettings transportSettings = settings.getTransportSettings();
+        if (transportSettings == null) {
+            return new SocketStreamFactory(inetAddressResolver, socketSettings, settings.getSslSettings());
+        } else if (transportSettings instanceof AsyncTransportSettings) {
+            throw new MongoClientException("Unsupported async transport settings: " + transportSettings.getClass().getName());
+        } else if (transportSettings instanceof NettyTransportSettings) {
+            return getNettyStreamFactoryFactory(inetAddressResolver, (NettyTransportSettings) transportSettings)
+                    .create(socketSettings, settings.getSslSettings());
+        } else {
+            throw new MongoClientException("Unsupported transport settings: " + transportSettings.getClass().getName());
+        }
+    }
+
+    public static StreamFactoryFactory getAsyncStreamFactoryFactory(final MongoClientSettings settings,
             final InetAddressResolver inetAddressResolver) {
-        if (transportSettings instanceof NettyTransportSettings) {
-            return NettyStreamFactoryFactory.builder().applySettings((NettyTransportSettings) transportSettings)
-                    .inetAddressResolver(inetAddressResolver)
-                    .build();
+        TransportSettings transportSettings = settings.getTransportSettings();
+        if (transportSettings == null || transportSettings instanceof AsyncTransportSettings) {
+            ExecutorService executorService = transportSettings == null
+                    ? null
+                    : ((AsyncTransportSettings) transportSettings).getExecutorService();
+            if (settings.getSslSettings().isEnabled()) {
+                return new TlsChannelStreamFactoryFactory(inetAddressResolver, executorService);
+            } else {
+                return new AsynchronousSocketChannelStreamFactoryFactory(inetAddressResolver, executorService);
+            }
+        } else  if (transportSettings instanceof NettyTransportSettings) {
+            return getNettyStreamFactoryFactory(inetAddressResolver, (NettyTransportSettings) transportSettings);
         } else {
             throw new MongoClientException("Unsupported transport settings: " + transportSettings.getClass().getName());
         }
     }
 
+    private static NettyStreamFactoryFactory getNettyStreamFactoryFactory(final InetAddressResolver inetAddressResolver,
+            final NettyTransportSettings transportSettings) {
+        return NettyStreamFactoryFactory.builder()
+                .applySettings(transportSettings)
+                .inetAddressResolver(inetAddressResolver)
+                .build();
+    }
+
     private StreamFactoryHelper() {
     }
 }

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

@@ -46,6 +46,7 @@
 import java.security.NoSuchAlgorithmException;
 import java.util.Iterator;
 import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
@@ -71,13 +72,18 @@ public class TlsChannelStreamFactoryFactory implements StreamFactoryFactory {
     /**
      * Construct a new instance
      */
-    public TlsChannelStreamFactoryFactory(final InetAddressResolver inetAddressResolver) {
+    public TlsChannelStreamFactoryFactory(final InetAddressResolver inetAddressResolver,
+            @Nullable final ExecutorService executorService) {
         this.inetAddressResolver = inetAddressResolver;
-        this.group = new AsynchronousTlsChannelGroup();
+        this.group = new AsynchronousTlsChannelGroup(executorService);
         selectorMonitor = new SelectorMonitor();
         selectorMonitor.start();
     }
 
+    public TlsChannelStreamFactoryFactory(final InetAddressResolver inetAddressResolver) {
+        this(inetAddressResolver, null);
+    }
+
     @Override
     public StreamFactory create(final SocketSettings socketSettings, final SslSettings sslSettings) {
         assertTrue(sslSettings.isEnabled());