GridFS fix source onComplete logic

Ensure that onComplete of the source publisher can trigger the completion
steps of the upload publisher

JAVA-3587

Author: rozza

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/GridFSUploadPublisherImpl.java

@@ -79,6 +79,9 @@ public void onSubscribe(final Subscription s) {
 
             @Override
             public void onNext(final ByteBuffer byteBuffer) {
+                synchronized (GridFSUploadSubscription.this) {
+                    currentAction = Action.IN_PROGRESS;
+                }
                 Publishers.publish((Block<SingleResultCallback<Integer>>) callback -> gridFSUploadStream.write(byteBuffer, callback))
                         .subscribe(new GridFSUploadStreamSubscriber(byteBuffer));
             }
@@ -95,8 +98,9 @@ public void onError(final Throwable t) {
             public void onComplete() {
                 synchronized (GridFSUploadSubscription.this) {
                     hasCompleted = true;
-                    if (currentAction == Action.WAITING) {
+                    if (currentAction == Action.REQUESTING_MORE) {
                         currentAction = Action.COMPLETE;
+                        tryProcess();
                     }
                 }
             }
@@ -180,9 +184,9 @@ private void tryProcess() {
                         if (sourceSubscription == null) {
                             nextStep = NextStep.SUBSCRIBE;
                         } else {
-                            nextStep = NextStep.WRITE;
+                            nextStep = NextStep.REQUEST_MORE;
                         }
-                        currentAction = Action.IN_PROGRESS;
+                        currentAction = Action.REQUESTING_MORE;
                         break;
                     case COMPLETE:
                         nextStep = NextStep.COMPLETE;
@@ -192,6 +196,7 @@ private void tryProcess() {
                         nextStep = NextStep.TERMINATE;
                         currentAction = Action.FINISHED;
                         break;
+                    case REQUESTING_MORE:
                     case IN_PROGRESS:
                     case FINISHED:
                     default:
@@ -204,7 +209,7 @@ private void tryProcess() {
                 case SUBSCRIBE:
                     source.subscribe(sourceSubscriber);
                     break;
-                case WRITE:
+                case REQUEST_MORE:
                     synchronized (this) {
                         sourceSubscription.request(1);
                     }
@@ -309,6 +314,7 @@ public void onComplete() {
 
     enum Action {
         WAITING,
+        REQUESTING_MORE,
         IN_PROGRESS,
         TERMINATE,
         COMPLETE,
@@ -317,7 +323,7 @@ enum Action {
 
     enum NextStep {
         SUBSCRIBE,
-        WRITE,
+        REQUEST_MORE,
         COMPLETE,
         TERMINATE,
         DO_NOTHING

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/gridfs/GridFSPublisherSpecification.groovy

@@ -119,8 +119,8 @@ class GridFSPublisherSpecification extends FunctionalSpecification {
         given:
         def contentSize = 1024 * 500
         def chunkSize = 10
-        def contentBytes = new byte[contentSize];
-        new SecureRandom().nextBytes(contentBytes);
+        def contentBytes = new byte[contentSize]
+        new SecureRandom().nextBytes(contentBytes)
         def options = new GridFSUploadOptions().chunkSizeBytes(chunkSize)
 
         when:
@@ -167,6 +167,32 @@ class GridFSPublisherSpecification extends FunctionalSpecification {
                                            ByteBuffer.wrap(contentBytes)])
     }
 
+    def 'should upload from the source publisher when it contains multiple parts and the total size is smaller than chunksize'() {
+        given:
+        def contentBytes = singleChunkString.getBytes()
+
+        when:
+        def publisher = gridFSBucket.uploadFromPublisher('myFile',
+                createPublisher(ByteBuffer.wrap(contentBytes), ByteBuffer.wrap(contentBytes)))
+
+        def subscriber = new ObservableSubscriber()
+        publisher.subscribe(subscriber)
+        def fileId = subscriber.await(1, 60, SECONDS).getReceived().get(0)
+
+        then:
+        run(filesCollection.&countDocuments) == 1
+        run(chunksCollection.&countDocuments) == 1
+
+        when:
+        subscriber = new ObservableSubscriber()
+        publisher = gridFSBucket.downloadToPublisher(fileId as ObjectId)
+        publisher.subscribe(subscriber)
+        def data = subscriber.await(1, 30, SECONDS).getReceived().get(0)
+
+        then:
+        data.array() == concatByteBuffers([ByteBuffer.wrap(contentBytes), ByteBuffer.wrap(contentBytes)])
+    }
+
     def 'should round trip with data larger than the internal bufferSize'() {
         given:
         def contentSize = 1024 * 1024 * 5