Netty fixes

Update Netty to 4.1.5.final
Fix leak in NettyStream
Made connection string streamType the authorative setting for the streamFactoryFactory
Added deprecation warning in NettyByteBuf

JAVA-2302

Author: rozza

build.gradle

@@ -18,7 +18,7 @@ apply plugin: 'eclipse'
 apply plugin: 'idea'
 
 def configDir = new File(rootDir, 'config')
-ext.nettyVersion = '4.0.26.Final'
+ext.nettyVersion = '4.1.5.Final'
 
 buildscript {
     repositories {

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

@@ -20,6 +20,7 @@
 import com.mongodb.client.MongoDriverInformation;
 import com.mongodb.client.gridfs.codecs.GridFSFileCodecProvider;
 import com.mongodb.client.model.geojson.codecs.GeoJsonCodecProvider;
+import com.mongodb.connection.AsynchronousSocketChannelStreamFactoryFactory;
 import com.mongodb.connection.Cluster;
 import com.mongodb.connection.ClusterSettings;
 import com.mongodb.connection.ConnectionPoolSettings;
@@ -155,6 +156,8 @@ public static MongoClient create(final ConnectionString connectionString, final
         if (connectionString.getStreamType() != null) {
             if (connectionString.getStreamType().toLowerCase().equals("netty")) {
                 builder.streamFactoryFactory(NettyStreamFactoryFactory.builder().build());
+            } else if (connectionString.getStreamType().toLowerCase().equals("nio2")) {
+                builder.streamFactoryFactory(new AsynchronousSocketChannelStreamFactoryFactory());
             } else if (!connectionString.getStreamType().toLowerCase().equals("nio2")) {
                 throw new IllegalArgumentException(format("Unsupported stream type %s", connectionString.getStreamType()));
             }

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

@@ -198,6 +198,30 @@ class MongoClientsSpecification extends FunctionalSpecification {
         'mongodb://localhost/?appname=app1' | 'app1'
     }
 
+    @Unroll
+    def 'should respect the streamType over the system properties'() {
+        given:
+        def asyncType = System.getProperty('org.mongodb.async.type', null)
+        System.setProperty('org.mongodb.async.type', systemType)
+
+        when:
+        def client = MongoClients.create(uri)
+
+        then:
+        client.settings.getStreamFactoryFactory().getClass() == streamFactoryFactoryClass
+
+        cleanup:
+        client?.close()
+        if (asyncType != null) {
+            System.setProperty('org.mongodb.async.type', asyncType)
+        }
+
+        where:
+        uri                                     |  systemType | streamFactoryFactoryClass
+        'mongodb://localhost/?streamType=nio2'  | 'netty'     | AsynchronousSocketChannelStreamFactoryFactory
+        'mongodb://localhost/?streamType=netty' | 'nio2'      | NettyStreamFactoryFactory
+    }
+
     @IgnoreIf({ !serverVersionAtLeast([3, 3, 9]) || !isStandalone() })
     def 'application name should appear in the system.profile collection'() {
         given:

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

@@ -113,6 +113,7 @@ public ByteBuf clear() {
     }
 
     @Override
+    @SuppressWarnings("deprecation")
     public ByteBuf order(final ByteOrder byteOrder) {
         proxied = proxied.order(byteOrder);
         return this;

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

@@ -171,9 +171,7 @@ public ByteBuf read(final int numBytes) throws IOException {
     public void writeAsync(final List<ByteBuf> buffers, final AsyncCompletionHandler<Void> handler) {
         CompositeByteBuf composite = PooledByteBufAllocator.DEFAULT.compositeBuffer();
         for (ByteBuf cur : buffers) {
-            io.netty.buffer.ByteBuf byteBuf = ((NettyByteBuf) cur).asByteBuf();
-            composite.addComponent(byteBuf.retain());
-            composite.writerIndex(composite.writerIndex() + byteBuf.writerIndex());
+            composite.addComponent(true, ((NettyByteBuf) cur).asByteBuf());
         }
 
         channel.writeAndFlush(composite).addListener(new ChannelFutureListener() {