Implement NettyByteBuf.asReadOnly (#1841)

CompositeByteBuf#duplicate should duplicate underlying buffers

JAVA-5982

---------

Co-authored-by: Valentin Kovalenko <valentin.male.kovalenko@gmail.com>

Author: rozza

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

@@ -49,7 +49,7 @@ class CompositeByteBuf implements ByteBuf {
         limit = components.get(components.size() - 1).endOffset;
     }
 
-    CompositeByteBuf(final CompositeByteBuf from) {
+    private CompositeByteBuf(final CompositeByteBuf from) {
         components = from.components;
         position = from.position();
         limit = from.limit();
@@ -58,7 +58,7 @@ class CompositeByteBuf implements ByteBuf {
     @Override
     public ByteBuf order(final ByteOrder byteOrder) {
         if (byteOrder == ByteOrder.BIG_ENDIAN) {
-            throw new UnsupportedOperationException(format("Only %s is supported", ByteOrder.BIG_ENDIAN));
+            throw new UnsupportedOperationException(format("Only %s is supported", ByteOrder.LITTLE_ENDIAN));
         }
         return this;
     }

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

@@ -251,7 +251,7 @@ public ByteBuf limit(final int newLimit) {
 
     @Override
     public ByteBuf asReadOnly() {
-        return this;  // TODO: do we need this method really?  Netty ByteBuf does not have this concept
+        return new NettyByteBuf(proxied.asReadOnly().retain(), false);
     }
 
     @Override

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
@@ -44,6 +43,7 @@ class CommandHelperSpecification extends Specification {
     InternalConnection connection
 
     def setup() {
+        InternalStreamConnection.setRecordEverything(true)
         connection = new InternalStreamConnectionFactory(ClusterConnectionMode.SINGLE,
                 new NettyStreamFactory(SocketSettings.builder().build(), getSslSettings()),
                 getCredentialWithCache(), CLIENT_METADATA, [], LoggerSettings.builder().build(), null, getServerApi())
@@ -55,7 +55,6 @@ class CommandHelperSpecification extends Specification {
         connection?.close()
     }
 
-    @Ignore("JAVA-5982")
     def 'should execute command asynchronously'() {
         when:
         BsonDocument receivedDocument = null