JAVA-2149: add ability to configure channel type of NettyStream in async driver

Author: Sergey Polovko

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

@@ -305,7 +305,7 @@ private static StreamFactoryFactory createDefaultStreamFactoryFactory() {
             String streamType = System.getProperty("org.mongodb.async.type", "nio2");
 
             if (streamType.equals("netty")) {
-                return new NettyStreamFactoryFactory();
+                return NettyStreamFactoryFactory.builder().build();
             } else if (streamType.equals("nio2")) {
                 return new AsynchronousSocketChannelStreamFactoryFactory();
             } else {

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2016 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.async.client
+
+import com.mongodb.async.FutureResultCallback
+import com.mongodb.connection.netty.NettyStreamFactoryFactory
+import io.netty.channel.oio.OioEventLoopGroup
+import io.netty.channel.socket.oio.OioSocketChannel
+import org.bson.Document
+
+import static com.mongodb.async.client.Fixture.getMongoClientBuilderFromConnectionString
+import static java.util.concurrent.TimeUnit.SECONDS
+
+class NettyStreamFactoryFactorySmokeTestSpecification extends FunctionalSpecification {
+
+    private MongoClient mongoClient
+
+    def 'should allow a custom Event Loop Group and Socket Channel'() {
+        given:
+        def eventLoopGroup = new OioEventLoopGroup()
+        def streamFactoryFactory = NettyStreamFactoryFactory.builder()
+                .eventLoopGroup(eventLoopGroup).socketChannelClass(OioSocketChannel).build()
+        MongoClientSettings.Builder builder = getMongoClientBuilderFromConnectionString().streamFactoryFactory(streamFactoryFactory)
+        def document = new Document('a', 1)
+
+        when:
+        mongoClient = MongoClients.create(builder.build())
+        def collection = mongoClient.getDatabase(databaseName).getCollection(collectionName)
+
+
+        then:
+        run(collection.&insertOne, document) == null
+
+        then: 'The count is one'
+        run(collection.&count) == 1
+
+        cleanup:
+        mongoClient?.close()
+
+    }
+
+    def run(operation, ... args) {
+        def futureResultCallback = new FutureResultCallback()
+        def opArgs = (args != null) ? args : []
+        operation.call(*opArgs + futureResultCallback)
+        futureResultCallback.get(60, SECONDS)
+    }
+}

driver-async/src/test/unit/com/mongodb/async/client/MongoClientSettingsSpecification.groovy

@@ -124,7 +124,7 @@ class MongoClientSettingsSpecification extends Specification {
 
     def 'should build with set options'() {
         given:
-        def streamFactoryFactory = new NettyStreamFactoryFactory()
+        def streamFactoryFactory = NettyStreamFactoryFactory.builder().build()
         def sslSettings = Stub(SslSettings)
         def socketSettings = Stub(SocketSettings)
         def serverSettings = Stub(ServerSettings)

driver-core/src/main/com/mongodb/connection/netty/NettyStream.java

@@ -40,7 +40,6 @@
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.timeout.ReadTimeoutException;
 import io.netty.util.concurrent.EventExecutor;
@@ -66,6 +65,7 @@ final class NettyStream implements Stream {
     private final SocketSettings settings;
     private final SslSettings sslSettings;
     private final EventLoopGroup workerGroup;
+    private final Class<? extends SocketChannel> socketChannelClass;
     private final ByteBufAllocator allocator;
 
     private volatile boolean isClosed;
@@ -76,11 +76,13 @@ final class NettyStream implements Stream {
     private volatile Throwable pendingException;
 
     public NettyStream(final ServerAddress address, final SocketSettings settings, final SslSettings sslSettings,
-                       final EventLoopGroup workerGroup, final ByteBufAllocator allocator) {
+                       final EventLoopGroup workerGroup, final Class<? extends SocketChannel> socketChannelClass,
+                       final ByteBufAllocator allocator) {
         this.address = address;
         this.settings = settings;
         this.sslSettings = sslSettings;
         this.workerGroup = workerGroup;
+        this.socketChannelClass = socketChannelClass;
         this.allocator = allocator;
     }
 
@@ -100,7 +102,7 @@ public void open() throws IOException {
     public void openAsync(final AsyncCompletionHandler<Void> handler) {
         Bootstrap bootstrap = new Bootstrap();
         bootstrap.group(workerGroup);
-        bootstrap.channel(NioSocketChannel.class);
+        bootstrap.channel(socketChannelClass);
 
         bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, settings.getConnectTimeout(MILLISECONDS));
         bootstrap.option(ChannelOption.TCP_NODELAY, true);
@@ -278,6 +280,26 @@ public boolean isClosed() {
         return isClosed;
     }
 
+    public SocketSettings getSettings() {
+        return settings;
+    }
+
+    public SslSettings getSslSettings() {
+        return sslSettings;
+    }
+
+    public EventLoopGroup getWorkerGroup() {
+        return workerGroup;
+    }
+
+    public Class<? extends SocketChannel> getSocketChannelClass() {
+        return socketChannelClass;
+    }
+
+    public ByteBufAllocator getAllocator() {
+        return allocator;
+    }
+
     private class InboundBufferHandler extends SimpleChannelInboundHandler<io.netty.buffer.ByteBuf> {
         @Override
         protected void channelRead0(final ChannelHandlerContext ctx, final io.netty.buffer.ByteBuf buffer) throws Exception {
@@ -344,7 +366,6 @@ public T get() throws IOException {
         }
     }
 
-
     private void scheduleReadTimeout() {
         adjustTimeout(false);
     }
@@ -384,6 +405,5 @@ public void run() {
                     }
                 }
             }
-
     }
 }

driver-core/src/main/com/mongodb/connection/netty/NettyStreamFactory.java

@@ -25,6 +25,8 @@
 import io.netty.buffer.PooledByteBufAllocator;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
 
 import static com.mongodb.assertions.Assertions.notNull;
 
@@ -37,6 +39,7 @@ public class NettyStreamFactory implements StreamFactory {
     private final SocketSettings settings;
     private final SslSettings sslSettings;
     private final EventLoopGroup eventLoopGroup;
+    private final Class<? extends SocketChannel> socketChannelClass;
     private final ByteBufAllocator allocator;
 
     /**
@@ -45,18 +48,36 @@ public class NettyStreamFactory implements StreamFactory {
      * @param settings the socket settings
      * @param sslSettings the SSL settings
      * @param eventLoopGroup the event loop group that all channels created by this factory will be a part of
+     * @param socketChannelClass the socket channel class
      * @param allocator the allocator to use for ByteBuf instances
+     *
+     * @since 3.3
      */
-    public NettyStreamFactory(final SocketSettings settings, final SslSettings sslSettings, final EventLoopGroup eventLoopGroup,
+    public NettyStreamFactory(final SocketSettings settings, final SslSettings sslSettings,
+                              final EventLoopGroup eventLoopGroup, final Class<? extends SocketChannel> socketChannelClass,
                               final ByteBufAllocator allocator) {
         this.settings = notNull("settings", settings);
         this.sslSettings = notNull("sslSettings", sslSettings);
         this.eventLoopGroup = notNull("eventLoopGroup", eventLoopGroup);
+        this.socketChannelClass = notNull("socketChannelClass", socketChannelClass);
         this.allocator = notNull("allocator", allocator);
     }
 
     /**
-     * Construct a new instance of the factory with a default allocator and event loop group.
+     * Construct a new instance of the factory.
+     *
+     * @param settings the socket settings
+     * @param sslSettings the SSL settings
+     * @param eventLoopGroup the event loop group that all channels created by this factory will be a part of
+     * @param allocator the allocator to use for ByteBuf instances
+     */
+    public NettyStreamFactory(final SocketSettings settings, final SslSettings sslSettings, final EventLoopGroup eventLoopGroup,
+                              final ByteBufAllocator allocator) {
+        this(settings, sslSettings, eventLoopGroup, NioSocketChannel.class, allocator);
+    }
+
+    /**
+     * Construct a new instance of the factory with a default allocator, nio event loop group and nio socket channel.
      *
      * @param settings the socket settings
      * @param sslSettings the SSL settings
@@ -67,7 +88,7 @@ public NettyStreamFactory(final SocketSettings settings, final SslSettings sslSe
 
     @Override
     public Stream create(final ServerAddress serverAddress) {
-        return new NettyStream(serverAddress, settings, sslSettings, eventLoopGroup, allocator);
+        return new NettyStream(serverAddress, settings, sslSettings, eventLoopGroup, socketChannelClass, allocator);
     }
 
 }

driver-core/src/main/com/mongodb/connection/netty/NettyStreamFactoryFactory.java

@@ -23,6 +23,8 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
 
 import static com.mongodb.assertions.Assertions.notNull;
 
@@ -34,28 +36,118 @@
 public class NettyStreamFactoryFactory implements StreamFactoryFactory {
 
     private final EventLoopGroup eventLoopGroup;
+    private final Class<? extends SocketChannel> socketChannelClass;
     private final ByteBufAllocator allocator;
 
+    /**
+     * Construct an instance with the default {@code EventLoopGroup} and {@code ByteBufAllocator}.
+     *
+     * @deprecated Use {@link NettyStreamFactoryFactory#builder()} instead to construct the {@code  NettyStreamFactoryFactory}.
+     */
+    @Deprecated
+    public NettyStreamFactoryFactory() {
+        this(builder());
+    }
+
     /**
      * Construct an instance with the given {@code EventLoopGroup} and {@code ByteBufAllocator}.
      *
      * @param eventLoopGroup the non-null event loop group
      * @param allocator the non-null byte buf allocator
+     * @deprecated Use {@link NettyStreamFactoryFactory#builder()} instead to construct the {@code  NettyStreamFactoryFactory}.
      */
+    @Deprecated
     public NettyStreamFactoryFactory(final EventLoopGroup eventLoopGroup, final ByteBufAllocator allocator) {
-        this.eventLoopGroup = notNull("eventLoopGroup", eventLoopGroup);
-        this.allocator = notNull("allocator", allocator);
+        this(builder().eventLoopGroup(eventLoopGroup).allocator(allocator));
     }
 
     /**
-     * Construct an instance with the default {@code EventLoopGroup} and {@code ByteBufAllocator}.
+     * Gets a builder for an instance of {@code NettyStreamFactoryFactory}.
+     * @return the builder
+     * @since 3.3
      */
-    public NettyStreamFactoryFactory() {
-        this(new NioEventLoopGroup(), ByteBufAllocator.DEFAULT);
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    /**
+     * A builder for an instance of {@code NettyStreamFactoryFactory}.
+     *
+     * @since 3.3
+     */
+    public static final class Builder {
+        private ByteBufAllocator allocator;
+        private Class<? extends SocketChannel> socketChannelClass;
+        private EventLoopGroup eventLoopGroup;
+
+        private Builder() {
+            allocator(ByteBufAllocator.DEFAULT);
+            socketChannelClass(NioSocketChannel.class);
+        }
+
+        /**
+         * Sets the allocator.
+         *
+         * @param allocator the allocator to use for ByteBuf instances
+         * @return this
+         */
+        public Builder allocator(final ByteBufAllocator allocator) {
+            this.allocator = notNull("allocator", allocator);
+            return this;
+        }
+
+        /**
+         * Sets the socket channel class
+         *
+         * @param socketChannelClass the socket channel class
+         * @return this
+         */
+        public Builder socketChannelClass(final Class<? extends SocketChannel> socketChannelClass) {
+            this.socketChannelClass = notNull("socketChannelClass", socketChannelClass);
+            return this;
+        }
+
+        /**
+         * Sets the event loop group.
+         *
+         * @param eventLoopGroup the event loop group that all channels created by this factory will be a part of
+         * @return this
+         */
+        public Builder eventLoopGroup(final EventLoopGroup eventLoopGroup) {
+            this.eventLoopGroup = notNull("eventLoopGroup", eventLoopGroup);
+            return this;
+        }
+
+        /**
+         * Build an instance of {@code NettyStreamFactoryFactory}.
+         * @return factory of the netty stream factory
+         */
+        public NettyStreamFactoryFactory build() {
+            return new NettyStreamFactoryFactory(this);
+        }
     }
 
     @Override
     public StreamFactory create(final SocketSettings socketSettings, final SslSettings sslSettings) {
-        return new NettyStreamFactory(socketSettings, sslSettings, eventLoopGroup, allocator);
+        return new NettyStreamFactory(socketSettings, sslSettings, eventLoopGroup, socketChannelClass, allocator);
+    }
+
+    @Override
+    public String toString() {
+        return "NettyStreamFactoryFactory{"
+                + "eventLoopGroup=" + eventLoopGroup
+                + ", socketChannelClass=" + socketChannelClass
+                + ", allocator=" + allocator
+                + '}';
+    }
+
+    private NettyStreamFactoryFactory(final Builder builder) {
+        allocator = builder.allocator;
+        socketChannelClass = builder.socketChannelClass;
+        if (builder.eventLoopGroup != null) {
+            eventLoopGroup = builder.eventLoopGroup;
+        } else {
+            eventLoopGroup = new NioEventLoopGroup();
+        }
     }
 }