Account for ByteBuffer.arrayOffset on ASCII fast-path and add UTF-8 encoding tests.

- Adjust logic to handle non-zero ByteBuffer.arrayOffset, as some Netty Pooled ByteBuffer implementations return an offset != 0.
- Add unit tests for UTF-8 encoding across buffer boundaries and for malformed surrogate pairs.
- Fix issue with a leacked reference count on ByteBufs in the pipe() method (2 non-released reference counts were retained).

JAVA-5816

Author: vbabanin

bson/src/main/org/bson/ByteBuf.java

@@ -136,8 +136,26 @@ public interface ByteBuf  {
      */
     byte[] array();
 
+    /**
+     * <p>States whether this buffer is backed by an accessible byte array.</p>
+     *
+     * <p>If this method returns {@code true} then the {@link #array()} and {@link #arrayOffset()} methods may safely be invoked.</p>
+     *
+     * @return {@code true} if, and only if, this buffer is backed by an array and is not read-only
+     * @since 5.5
+     */
     boolean hasArray();
 
+    /**
+     * Returns the offset of the first byte within the backing byte array of
+     * this buffer.
+     *
+     * @throws java.nio.ReadOnlyBufferException If this buffer is backed by an array but is read-only
+     * @throws UnsupportedOperationException if this buffer is not backed by an accessible array
+     * @since 5.5
+     */
+    int arrayOffset();
+
     /**
      * Returns this buffer's limit.
      *

bson/src/main/org/bson/ByteBufNIO.java

@@ -113,6 +113,11 @@ public boolean hasArray() {
         return buf.hasArray();
     }
 
+    @Override
+    public int arrayOffset() {
+        return buf.arrayOffset();
+    }
+
     @Override
     public int limit() {
         return buf.limit();

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

@@ -110,9 +110,10 @@ private ByteBuf getNextByteBuffer() {
 
     private ByteBuf getByteBufferAtIndex(final int index) {
         if (bufferList.size() < index + 1) {
-            bufferList.add(bufferProvider.getBuffer(index >= (MAX_SHIFT - INITIAL_SHIFT)
-                                                            ? MAX_BUFFER_SIZE
-                                                            : Math.min(INITIAL_BUFFER_SIZE << index, MAX_BUFFER_SIZE)));
+            ByteBuf buffer = bufferProvider.getBuffer(index >= (MAX_SHIFT - INITIAL_SHIFT)
+                    ? MAX_BUFFER_SIZE
+                    : Math.min(INITIAL_BUFFER_SIZE << index, MAX_BUFFER_SIZE));
+            bufferList.add(buffer);
         }
         return bufferList.get(index);
     }
@@ -155,6 +156,16 @@ public List<ByteBuf> getByteBuffers() {
         return buffers;
     }
 
+    public List<ByteBuf> getDuplicateByteBuffers() {
+        ensureOpen();
+
+        List<ByteBuf> buffers = new ArrayList<>(bufferList.size());
+        for (final ByteBuf cur : bufferList) {
+            buffers.add(cur.duplicate().order(ByteOrder.LITTLE_ENDIAN));
+        }
+        return buffers;
+    }
+
 
     @Override
     public int pipe(final OutputStream out) throws IOException {
@@ -163,14 +174,18 @@ public int pipe(final OutputStream out) throws IOException {
         byte[] tmp = new byte[INITIAL_BUFFER_SIZE];
 
         int total = 0;
-        for (final ByteBuf cur : getByteBuffers()) {
-            ByteBuf dup = cur.duplicate();
-            while (dup.hasRemaining()) {
-                int numBytesToCopy = Math.min(dup.remaining(), tmp.length);
-                dup.get(tmp, 0, numBytesToCopy);
-                out.write(tmp, 0, numBytesToCopy);
+        List<ByteBuf> byteBuffers = getByteBuffers();
+        try {
+            for (final ByteBuf cur : byteBuffers) {
+                while (cur.hasRemaining()) {
+                    int numBytesToCopy = Math.min(cur.remaining(), tmp.length);
+                    cur.get(tmp, 0, numBytesToCopy);
+                    out.write(tmp, 0, numBytesToCopy);
+                }
+                total += cur.limit();
             }
-            total += dup.limit();
+        } finally {
+            byteBuffers.forEach(ByteBuf::release);
         }
         return total;
     }
@@ -301,13 +316,14 @@ protected int writeCharacters(final String str, final boolean checkNullTerminati
         int limit = buf.limit();
         int remaining = limit - currBufferPos;
 
-        if (buf instanceof PowerOfTwoBufferPool.PooledByteBufNIO && buf.hasArray()) {
+        if (buf.hasArray()) {
             byte[] dst = buf.array();
+            int arrayOffset = buf.arrayOffset();
             if (remaining >= str.length() + 1) {
-                sp = writeOnArrayAscii(str, dst, currBufferPos, checkNullTermination);
+                sp = writeOnArrayAscii(str, dst, arrayOffset + currBufferPos, checkNullTermination);
                 currBufferPos += sp;
                 if (sp == len) {
-                    dst[currBufferPos++] = 0;
+                    dst[arrayOffset + currBufferPos++] = 0;
                     position += sp + 1;
                     buf.position(currBufferPos);
                     return sp + 1;
@@ -397,9 +413,9 @@ protected int writeCharacters(final String str, final boolean checkNullTerminati
 
     private static int writeOnArrayAscii(final String str,
                                          final byte[] dst,
-                                         final int currentPos,
+                                         final int arrayPosition,
                                          final boolean checkNullTermination) {
-        int pos = currentPos;
+        int pos = arrayPosition;
         int sp = 0;
         for (; sp < str.length(); sp++, pos++) {
             char c = str.charAt(sp);

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

@@ -218,6 +218,11 @@ public boolean hasArray() {
         return false;
     }
 
+    @Override
+    public int arrayOffset() {
+        throw new UnsupportedOperationException("Not implemented yet!");
+    }
+
     @Override
     public ByteBuf limit(final int newLimit) {
         if (newLimit < 0 || newLimit > capacity()) {

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

@@ -156,7 +156,7 @@ static int roundUpToNextHighestPowerOfTwo(final int size) {
         return v;
     }
 
-    public class PooledByteBufNIO extends ByteBufNIO {
+    private class PooledByteBufNIO extends ByteBufNIO {
 
         PooledByteBufNIO(final ByteBuffer buf) {
             super(buf);

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

@@ -105,6 +105,11 @@ public boolean hasArray() {
         return proxied.hasArray();
     }
 
+    @Override
+    public int arrayOffset() {
+        return proxied.arrayOffset();
+    }
+
     @Override
     public int limit() {
         if (isWriting) {

driver-core/src/test/unit/com/mongodb/internal/connection/ByteBufSpecification.groovy

@@ -249,11 +249,7 @@ class ByteBufSpecification extends Specification {
         @Override
         ByteBuf getBuffer(final int size) {
             io.netty.buffer.ByteBuf buffer = allocator.directBuffer(size, size)
-            try {
-                new NettyByteBuf(buffer.retain())
-            } finally {
-                buffer.release();
-            }
+            new NettyByteBuf(buffer)
         }
     }
 }