propagate cancellation

Author: filippor

spring-cloud-gcp-pubsub/src/main/java/com/google/cloud/spring/pubsub/reactive/PubSubReactiveFactory.java

@@ -129,6 +129,7 @@ private Flux<List<AcknowledgeablePubsubMessage>> backpressurePull(
     int intDemand = numRequested > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) numRequested;
     var future = this.subscriberOperations.pullAsync(subscriptionName, intDemand, false);
     return Mono.fromFuture(future)
+          .publishOn(scheduler)
           .flatMapMany(messages -> {
             long numToPull = numRequested - messages.size();
             if (numToPull > 0) {
@@ -159,6 +160,7 @@ private void transferMessages(Flux<List<AcknowledgeablePubsubMessage>> source, F
           }
           if (destination.isCancelled()) {
             messages.forEach(msg -> msg.modifyAckDeadline(0));
+            source.cancelOn(scheduler);
           }
         }).doOnError(destination::error).subscribe());
   }