Making Pub/Sub reactive pull non-blocking

spring-cloud-gcp-autoconfigure/src/main/java/org/springframework/cloud/gcp/autoconfigure/pubsub/GcpPubSubAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -72,6 +72,7 @@
  * @author Mike Eltsufin
  * @author Chengyuan Zhao
  * @author Daniel Zou
+ * @author Maurice Zeijen
  */
 @Configuration
 @AutoConfigureAfter(GcpContextAutoConfiguration.class)
@@ -167,10 +168,12 @@ public Executor pubSubAcknowledgementExecutor() {
 	@ConditionalOnMissingBean
 	public PubSubSubscriberTemplate pubSubSubscriberTemplate(SubscriberFactory subscriberFactory,
 			ObjectProvider<PubSubMessageConverter> pubSubMessageConverter,
-			@Qualifier("pubSubAcknowledgementExecutor") Executor executor) {
+			@Qualifier("pubSubAsynchronousPullExecutor") ObjectProvider<Executor> asyncPullExecutor,
+			@Qualifier("pubSubAcknowledgementExecutor") Executor ackExecutor) {
 		PubSubSubscriberTemplate pubSubSubscriberTemplate = new PubSubSubscriberTemplate(subscriberFactory);
 		pubSubMessageConverter.ifUnique(pubSubSubscriberTemplate::setMessageConverter);
-		pubSubSubscriberTemplate.setAckExecutor(executor);
+		pubSubSubscriberTemplate.setAckExecutor(ackExecutor);
+		asyncPullExecutor.ifAvailable(pubSubSubscriberTemplate::setAsyncPullExecutor);
 		return pubSubSubscriberTemplate;
 	}
 

spring-cloud-gcp-autoconfigure/src/main/java/org/springframework/cloud/gcp/autoconfigure/pubsub/GcpPubSubReactiveAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,8 +18,6 @@
 
 import java.util.Optional;
 
-import javax.annotation.PreDestroy;
-
 import reactor.core.publisher.Flux;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
@@ -38,6 +36,7 @@
  * Reactive Pub/Sub support autoconfiguration.
  *
  * @author Elena Felder
+ * @author Maurice Zeijen
  *
  * @since 1.2
  */
@@ -49,30 +48,14 @@
 		matchIfMissing = true)
 public class GcpPubSubReactiveAutoConfiguration {
 
-	private Scheduler defaultPubSubReactiveScheduler;
-
 	@Bean
 	@ConditionalOnMissingBean
 	public PubSubReactiveFactory pubSubReactiveFactory(
 			PubSubSubscriberTemplate subscriberTemplate,
 			@Qualifier("pubSubReactiveScheduler") Optional<Scheduler> userProvidedScheduler) {
 
-		Scheduler scheduler = null;
-		if (userProvidedScheduler.isPresent()) {
-			scheduler = userProvidedScheduler.get();
-		}
-		else {
-			this.defaultPubSubReactiveScheduler = Schedulers.newElastic("pubSubReactiveScheduler");
-			scheduler = this.defaultPubSubReactiveScheduler;
-		}
+		Scheduler scheduler = userProvidedScheduler.orElseGet(() -> Schedulers.parallel());
 		return new PubSubReactiveFactory(subscriberTemplate, scheduler);
 	}
 
-	@PreDestroy
-	public void closeScheduler() {
-		if (this.defaultPubSubReactiveScheduler != null) {
-			this.defaultPubSubReactiveScheduler.dispose();
-		}
-	}
-
 }

spring-cloud-gcp-autoconfigure/src/test/java/org/springframework/cloud/gcp/autoconfigure/pubsub/GcpPubSubReactiveAutoConfigurationTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -42,13 +42,15 @@
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.AsyncResult;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 /**
- *
+ * @author Elena Felder
+ * @author Maurice Zeijen
  */
 @RunWith(MockitoJUnitRunner.class)
 public class GcpPubSubReactiveAutoConfigurationTest {
@@ -105,17 +107,13 @@ public void reactiveConfigDisabledWhenReactivePubSubDisabled() {
 
 	@Test
 	public void defaultSchedulerUsedWhenNoneProvided() {
-
-		setUpThreadPrefixVerification("pubSubReactiveScheduler");
+		setUpThreadPrefixVerification("parallel");
 
 		ApplicationContextRunner contextRunner = new ApplicationContextRunner()
 				.withBean(PubSubSubscriberOperations.class, () -> mockSubscriberTemplate)
 				.withConfiguration(AutoConfigurations.of(GcpPubSubReactiveAutoConfiguration.class));
 
-		contextRunner.run(ctx -> {
-			pollAndVerify(ctx);
-		});
-
+		contextRunner.run(this::pollAndVerify);
 	}
 
 	@Test
@@ -128,28 +126,26 @@ public void customSchedulerUsedWhenAvailable() {
 				.withConfiguration(AutoConfigurations.of(GcpPubSubReactiveAutoConfiguration.class))
 				.withUserConfiguration(TestConfigWithOverriddenScheduler.class);
 
-		contextRunner.run(ctx -> {
-			pollAndVerify(ctx);
-		});
+		contextRunner.run(this::pollAndVerify);
 	}
 
 	private void setUpThreadPrefixVerification(String threadPrefix) {
-		when(mockSubscriberTemplate.pull("testSubscription", 3, false))
+		when(mockSubscriberTemplate.pullAsync("testSubscription", Integer.MAX_VALUE, true))
 				.then(arg -> {
 					assertThat(Thread.currentThread().getName()).startsWith(threadPrefix);
 
-					return Arrays.asList(mockMessage, mockMessage, mockMessage);
+					return AsyncResult.forValue(Arrays.asList(mockMessage, mockMessage, mockMessage));
 				});
 	}
 
 	private void pollAndVerify(ApplicationContext ctx) {
 		PubSubReactiveFactory reactiveFactory = ctx.getBean(PubSubReactiveFactory.class);
 
 		StepVerifier.create(
-				reactiveFactory.poll("testSubscription", 2)
-						.limitRequest(3))
+				reactiveFactory.poll("testSubscription", 2))
 				.expectNext(mockMessage, mockMessage, mockMessage)
-				.verifyComplete();
+				.thenCancel()
+				.verify();
 	}
 
 

spring-cloud-gcp-pubsub/src/main/java/org/springframework/cloud/gcp/pubsub/core/PubSubTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2018 the original author or authors.
+ * Copyright 2017-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -51,6 +51,7 @@
  * @author Mike Eltsufin
  * @author Chengyuan Zhao
  * @author Doug Hoard
+ * @author Maurice Zeijen
  */
 public class PubSubTemplate implements PubSubOperations, InitializingBean {
 
@@ -158,23 +159,45 @@ public List<AcknowledgeablePubsubMessage> pull(String subscription, Integer maxM
 		return this.pubSubSubscriberTemplate.pull(subscription, maxMessages, returnImmediately);
 	}
 
+	@Override
+	public ListenableFuture<List<AcknowledgeablePubsubMessage>> pullAsync(String subscription, Integer maxMessages,
+																			Boolean returnImmediately) {
+		return this.pubSubSubscriberTemplate.pullAsync(subscription, maxMessages, returnImmediately);
+	}
+
 	@Override
 	public <T> List<ConvertedAcknowledgeablePubsubMessage<T>> pullAndConvert(String subscription, Integer maxMessages,
 			Boolean returnImmediately, Class<T> payloadType) {
 		return this.pubSubSubscriberTemplate.pullAndConvert(subscription, maxMessages, returnImmediately, payloadType);
 	}
 
+	@Override
+	public <T> ListenableFuture<List<ConvertedAcknowledgeablePubsubMessage<T>>> pullAndConvertAsync(String subscription,
+			Integer maxMessages, Boolean returnImmediately, Class<T> payloadType) {
+		return this.pubSubSubscriberTemplate.pullAndConvertAsync(subscription, maxMessages, returnImmediately, payloadType);
+	}
+
 	@Override
 	public List<PubsubMessage> pullAndAck(String subscription, Integer maxMessages,
 			Boolean returnImmediately) {
 		return this.pubSubSubscriberTemplate.pullAndAck(subscription, maxMessages, returnImmediately);
 	}
 
+	@Override
+	public ListenableFuture<List<PubsubMessage>> pullAndAckAsync(String subscription, Integer maxMessages, Boolean returnImmediately) {
+		return this.pubSubSubscriberTemplate.pullAndAckAsync(subscription, maxMessages, returnImmediately);
+	}
+
 	@Override
 	public PubsubMessage pullNext(String subscription) {
 		return this.pubSubSubscriberTemplate.pullNext(subscription);
 	}
 
+	@Override
+	public ListenableFuture<PubsubMessage> pullNextAsync(String subscription) {
+		return this.pubSubSubscriberTemplate.pullNextAsync(subscription);
+	}
+
 	@Override
 	public void afterPropertiesSet() throws Exception {
 	}

spring-cloud-gcp-pubsub/src/main/java/org/springframework/cloud/gcp/pubsub/core/subscriber/PubSubSubscriberOperations.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2018 the original author or authors.
+ * Copyright 2017-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -38,6 +38,7 @@
  * @author Mike Eltsufin
  * @author Chengyuan Zhao
  * @author Doug Hoard
+ * @author Maurice Zeijen
  *
  * @since 1.1
  */
@@ -92,6 +93,19 @@ <T> Subscriber subscribeAndConvert(String subscription,
 	 */
 	List<PubsubMessage> pullAndAck(String subscription, Integer maxMessages, Boolean returnImmediately);
 
+	/**
+	 * Asynchronously pull and auto-acknowledge a number of messages from a Google Cloud Pub/Sub subscription.
+	 * @param subscription canonical subscription name, e.g., "subscriptionName", or the fully-qualified
+	 * subscription name in the {@code projects/<project_name>/subscriptions/<subscription_name>} format
+	 * @param maxMessages the maximum number of pulled messages
+	 * @param returnImmediately returns immediately even if subscription doesn't contain enough
+	 * messages to satisfy {@code maxMessages}
+	 * @return the ListenableFuture for the asynchronous execution, returning the list of
+	 * received acknowledgeable messages
+	 * @since 1.2.3
+	 */
+	ListenableFuture<List<PubsubMessage>> pullAndAckAsync(String subscription, Integer maxMessages, Boolean returnImmediately);
+
 	/**
 	 * Pull a number of messages from a Google Cloud Pub/Sub subscription.
 	 * @param subscription canonical subscription name, e.g., "subscriptionName", or the fully-qualified
@@ -103,6 +117,19 @@ <T> Subscriber subscribeAndConvert(String subscription,
 	 */
 	List<AcknowledgeablePubsubMessage> pull(String subscription, Integer maxMessages, Boolean returnImmediately);
 
+	/**
+	 * Asynchronously pull a number of messages from a Google Cloud Pub/Sub subscription.
+	 * @param subscription canonical subscription name, e.g., "subscriptionName", or the fully-qualified
+	 * subscription name in the {@code projects/<project_name>/subscriptions/<subscription_name>} format
+	 * @param maxMessages the maximum number of pulled messages
+	 * @param returnImmediately returns immediately even if subscription doesn't contain enough
+	 * messages to satisfy {@code maxMessages}
+	 * @return the ListenableFuture for the asynchronous execution, returning the list of
+	 * received acknowledgeable messages
+	 * @since 1.2.3
+	 */
+	ListenableFuture<List<AcknowledgeablePubsubMessage>> pullAsync(String subscription, Integer maxMessages, Boolean returnImmediately);
+
 	/**
 	 * Pull a number of messages from a Google Cloud Pub/Sub subscription and convert them to Spring messages with
 	 * the desired payload type.
@@ -119,6 +146,22 @@ <T> Subscriber subscribeAndConvert(String subscription,
 	<T> List<ConvertedAcknowledgeablePubsubMessage<T>> pullAndConvert(String subscription, Integer maxMessages,
 			Boolean returnImmediately, Class<T> payloadType);
 
+	/**
+	 * Asynchronously pull a number of messages from a Google Cloud Pub/Sub subscription and convert them to Spring messages with
+	 * the desired payload type.
+	 * @param subscription canonical subscription name, e.g., "subscriptionName", or the fully-qualified
+	 * subscription name in the {@code projects/<project_name>/subscriptions/<subscription_name>} format
+	 * @param maxMessages the maximum number of pulled messages
+	 * @param returnImmediately returns immediately even if subscription doesn't contain enough
+	 * messages to satisfy {@code maxMessages}
+	 * @param payloadType the type to which the payload of the Pub/Sub messages should be converted
+	 * @param <T> the type of the payload
+	 * @return the ListenableFuture for the asynchronous execution, returning the list of
+	 *  received acknowledgeable messages
+	 * @since 1.2.3
+	 */
+	<T> ListenableFuture<List<ConvertedAcknowledgeablePubsubMessage<T>>> pullAndConvertAsync(String subscription,
+			Integer maxMessages, Boolean returnImmediately, Class<T> payloadType);
 
 	/**
 	 * Pull and auto-acknowledge a message from a Google Cloud Pub/Sub subscription.
@@ -128,6 +171,16 @@ <T> List<ConvertedAcknowledgeablePubsubMessage<T>> pullAndConvert(String subscri
 	 */
 	PubsubMessage pullNext(String subscription);
 
+	/**
+	 * Asynchronously pull and auto-acknowledge a message from a Google Cloud Pub/Sub subscription.
+	 * @param subscription canonical subscription name, e.g., "subscriptionName", or the fully-qualified
+	 * subscription name in the {@code projects/<project_name>/subscriptions/<subscription_name>} format
+	 * @return the ListenableFuture for the asynchronous execution, returning a received message,
+	 * or {@code null} if none exists in the subscription
+	 * @since 1.2.3
+	 */
+	ListenableFuture<PubsubMessage> pullNextAsync(String subscription);
+
 	/**
 	 * Acknowledge a batch of messages. The messages must have the same project id.
 	 * @param acknowledgeablePubsubMessages messages to be acknowledged