Make NettyByteBuf share parent reference count (#1897)

JAVA-6107

Author: vbabanin

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

@@ -256,7 +256,7 @@ public ByteBuf asReadOnly() {
 
     @Override
     public ByteBuf duplicate() {
-        return new NettyByteBuf(proxied.retainedDuplicate(), isWriting);
+        return new NettyByteBuf(proxied.duplicate().retain(), isWriting);
     }
 
     @Override

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

@@ -307,7 +307,8 @@ private void readAsync(final int numBytes, final AsyncCompletionHandler<ByteBuf>
                             composite.addComponent(next);
                             iter.remove();
                         } else {
-                            composite.addComponent(next.readRetainedSlice(bytesNeededFromCurrentBuffer));
+                            next.retain();
+                            composite.addComponent(next.readSlice(bytesNeededFromCurrentBuffer));
                         }
                         composite.writerIndex(composite.writerIndex() + bytesNeededFromCurrentBuffer);
                         bytesNeeded -= bytesNeededFromCurrentBuffer;

driver-core/src/test/functional/com/mongodb/internal/connection/CommandHelperSpecification.groovy

@@ -25,7 +25,6 @@ import com.mongodb.connection.SocketSettings
 import com.mongodb.internal.connection.netty.NettyStreamFactory
 import org.bson.BsonDocument
 import org.bson.BsonInt32
-import spock.lang.Ignore
 import spock.lang.Specification
 
 import java.util.concurrent.CountDownLatch
@@ -57,7 +56,6 @@ class CommandHelperSpecification extends Specification {
         connection?.close()
     }
 
-    @Ignore("JAVA-5982")
     def 'should execute command asynchronously'() {
         when:
         BsonDocument receivedDocument = null
@@ -84,5 +82,4 @@ class CommandHelperSpecification extends Specification {
         !receivedDocument
         receivedException instanceof MongoCommandException
     }
-
 }