Fix Async GridFS Download to Stream small buffer bug

When using a smaller than chunksize buffer, need to check the internal buffer has been read before trying to get the next chunk.

JAVA-2294

Author: rozza

driver-async/src/main/com/mongodb/async/client/gridfs/GridFSDownloadStreamImpl.java

@@ -141,7 +141,7 @@ public void onResult(final Integer result, final Throwable t) {
     private void checkAndFetchResults(final int amountRead, final ByteBuffer dst, final SingleResultCallback<Integer> callback) {
         if (currentPosition == fileInfo.getLength() || dst.remaining() == 0) {
             callback.onResult(amountRead, null);
-        } else if (!resultsQueue.isEmpty()) {
+        } else if (hasResultsToProcess()) {
             processResults(amountRead, dst, callback);
         } else if (cursor == null) {
             chunksCollection.find(new Document("files_id", fileInfo.getId())
@@ -181,23 +181,27 @@ public void onResult(final List<Document> result, final Throwable t) {
     private void processResults(final int previousAmountRead, final ByteBuffer dst, final SingleResultCallback<Integer> callback) {
         try {
             int amountRead = previousAmountRead;
-            while (currentPosition < fileInfo.getLength() && dst.remaining() > 0 && !resultsQueue.isEmpty()) {
-                if (buffer == null || bufferOffset == buffer.length) {
+            int amountToCopy = dst.remaining();
+            while (currentPosition < fileInfo.getLength() && amountToCopy > 0) {
+
+                if (getBufferFromResultsQueue()) {
                     buffer = getBufferFromChunk(resultsQueue.poll(), chunkIndex);
                     bufferOffset = 0;
                     chunkIndex += 1;
                 }
 
-                int amountToCopy = dst.remaining();
                 if (amountToCopy > buffer.length - bufferOffset) {
                     amountToCopy = buffer.length - bufferOffset;
                 }
-                dst.put(buffer, bufferOffset, amountToCopy);
-                bufferOffset += amountToCopy;
-                currentPosition += amountToCopy;
-                amountRead += amountToCopy;
-            }
 
+                if (amountToCopy > 0) {
+                    dst.put(buffer, bufferOffset, amountToCopy);
+                    bufferOffset += amountToCopy;
+                    currentPosition += amountToCopy;
+                    amountRead += amountToCopy;
+                    amountToCopy = dst.remaining();
+                }
+            }
             checkAndFetchResults(amountRead, dst, callback);
         } catch (MongoGridFSException e) {
             callback.onResult(null, e);
@@ -259,6 +263,14 @@ private byte[] getBufferFromChunk(final Document chunk, final int expectedChunkI
         return data;
     }
 
+    private boolean getBufferFromResultsQueue() {
+        return !resultsQueue.isEmpty() && (buffer == null || bufferOffset == buffer.length);
+    }
+
+    private boolean hasResultsToProcess() {
+        return !resultsQueue.isEmpty() || (buffer != null && bufferOffset < buffer.length);
+    }
+
     private <A> boolean tryGetReadingLock(final SingleResultCallback<A> callback) {
         if (checkClosed()) {
             callbackClosedException(callback);

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

@@ -226,7 +226,7 @@ class GridFSBucketSmokeTestSpecification extends FunctionalSpecification {
         given:
         def contentBytes = new byte[9000];
         new SecureRandom().nextBytes(contentBytes);
-        def bufferSize = 3000
+        def bufferSize = 2000
         def options = new GridFSUploadOptions().chunkSizeBytes(4000)
 
         when:
@@ -266,6 +266,24 @@ class GridFSBucketSmokeTestSpecification extends FunctionalSpecification {
         fileBuffer.put(byteBuffer.array())
         totalRead += read
         read = run(downloadStream.&read, byteBuffer.clear())
+        then:
+        read == bufferSize
+
+        when:
+        fileBuffer.put(byteBuffer.array())
+        totalRead += read
+        read = run(downloadStream.&read, byteBuffer.clear())
+
+        then:
+        read == 1000
+
+        when:
+        def remaining = new byte[read]
+        byteBuffer.flip()
+        byteBuffer.get(remaining, 0, 1000)
+        fileBuffer.put(remaining)
+        totalRead += read
+        read = run(downloadStream.&read, byteBuffer.clear())
 
         then:
         read == -1