This repository has been archived by the owner on Jan 19, 2022. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 692
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Reactive support for Pub/Sub subscription (#1461)
The polling Pub/Sub flux is demand-sensitive, implementing the pull/push strategy. Unbounded/bounded demand is treated differently: * For unlimited demand, the Pub/Sub subscription will be polled regularly, at intervals determined by `pollingPeriodMs` parameter passed in when creating the `Flux`. * For bounded demand, as many messages as possible (up to the requested number) are delivered immediately, with the remaining messages delivered as they become available.
- Loading branch information
Showing
14 changed files
with
643 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
59 changes: 59 additions & 0 deletions
59
...rg/springframework/cloud/gcp/autoconfigure/pubsub/GcpPubSubReactiveAutoConfiguration.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| /* | ||
| * Copyright 2017-2019 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. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.springframework.cloud.gcp.autoconfigure.pubsub; | ||
|
|
||
| import reactor.core.publisher.Flux; | ||
| import reactor.core.scheduler.Scheduler; | ||
| import reactor.core.scheduler.Schedulers; | ||
|
|
||
| import org.springframework.beans.factory.annotation.Qualifier; | ||
| import org.springframework.boot.autoconfigure.AutoConfigureAfter; | ||
| import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; | ||
| import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; | ||
| import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; | ||
| import org.springframework.cloud.gcp.pubsub.core.subscriber.PubSubSubscriberTemplate; | ||
| import org.springframework.cloud.gcp.pubsub.reactive.PubSubReactiveFactory; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Configuration; | ||
|
|
||
| /** | ||
| * Reactive Pub/Sub support autoconfiguration. | ||
| * | ||
| * @author Elena Felder | ||
| * | ||
| * @since 1.2 | ||
| */ | ||
| @Configuration | ||
| @AutoConfigureAfter(GcpPubSubAutoConfiguration.class) | ||
| @ConditionalOnClass(Flux.class) | ||
| @ConditionalOnProperty(value = "spring.cloud.gcp.pubsub.reactive.enabled", matchIfMissing = true) | ||
| public class GcpPubSubReactiveAutoConfiguration { | ||
|
|
||
| @Bean | ||
| @ConditionalOnMissingBean(name = "pubSubReactiveScheduler") | ||
| Scheduler pubSubReactiveScheduler() { | ||
| return Schedulers.elastic(); | ||
| } | ||
|
|
||
| @Bean | ||
| @ConditionalOnMissingBean | ||
| public PubSubReactiveFactory pubSubReactiveFactory( | ||
| PubSubSubscriberTemplate subscriberTemplate, @Qualifier("pubSubReactiveScheduler") Scheduler scheduler) { | ||
| return new PubSubReactiveFactory(subscriberTemplate, scheduler); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
179 changes: 179 additions & 0 deletions
179
...ub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,179 @@ | ||
| /* | ||
| * Copyright 2017-2019 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. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.springframework.cloud.gcp.pubsub.reactive; | ||
|
|
||
| import java.util.List; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| import com.google.api.gax.rpc.DeadlineExceededException; | ||
| import org.apache.commons.logging.Log; | ||
| import org.apache.commons.logging.LogFactory; | ||
| import reactor.core.publisher.Flux; | ||
| import reactor.core.publisher.FluxSink; | ||
| import reactor.core.scheduler.Scheduler; | ||
|
|
||
| import org.springframework.cloud.gcp.pubsub.core.subscriber.PubSubSubscriberOperations; | ||
| import org.springframework.cloud.gcp.pubsub.support.AcknowledgeablePubsubMessage; | ||
| import org.springframework.util.Assert; | ||
|
|
||
| /** | ||
| * A factory for procuring {@link Flux} instances backed by GCP Pub/Sub Subscriptions. | ||
| * | ||
| * @author Elena Felder | ||
| * | ||
| * @since 1.2 | ||
| */ | ||
| public final class PubSubReactiveFactory { | ||
|
|
||
| private static final Log LOGGER = LogFactory.getLog(PubSubReactiveFactory.class); | ||
|
|
||
| private final PubSubSubscriberOperations subscriberOperations; | ||
|
|
||
| private final Scheduler scheduler; | ||
|
|
||
| /** | ||
| * Instantiate `PubSubReactiveFactory` capable of generating subscription-based streams. | ||
| * @param subscriberOperations template for interacting with GCP Pub/Sub subscriber operations. | ||
| * @param scheduler scheduler to use for asynchronously retrieving Pub/Sub messages. | ||
| */ | ||
| public PubSubReactiveFactory(PubSubSubscriberOperations subscriberOperations, Scheduler scheduler) { | ||
| Assert.notNull(subscriberOperations, "subscriberOperations cannot be null."); | ||
| Assert.notNull(scheduler, "scheduler cannot be null."); | ||
| this.subscriberOperations = subscriberOperations; | ||
| this.scheduler = scheduler; | ||
| } | ||
|
|
||
| /** | ||
| * Create an infinite stream {@link Flux} of {@link AcknowledgeablePubsubMessage} objects. | ||
| * <p>The {@link Flux} respects backpressure by using of Pub/Sub Synchronous Pull to retrieve | ||
| * batches of up to the requested number of messages until the full demand is fulfilled | ||
| * or subscription terminated. | ||
| * <p>For unlimited demand, the underlying subscription will be polled at a regular interval, | ||
| * requesting up to {@code Integer.MAX_VALUE} messages at each poll. | ||
| * <p>For specific demand, as many messages as are available will be returned immediately, | ||
| * with remaining demand being fulfilled in the future. | ||
| * Pub/Sub timeout will cause a retry with the same demand. | ||
| * @param subscriptionName subscription from which to retrieve messages. | ||
| * @param pollingPeriodMs how frequently to poll the source subscription in case of unlimited demand, in milliseconds. | ||
| * @return infinite stream of {@link AcknowledgeablePubsubMessage} objects. | ||
| */ | ||
| public Flux<AcknowledgeablePubsubMessage> poll(String subscriptionName, long pollingPeriodMs) { | ||
|
|
||
| return Flux.create(sink -> { | ||
|
|
||
| Scheduler.Worker subscriptionWorker = this.scheduler.createWorker(); | ||
|
|
||
| sink.onRequest((numRequested) -> { | ||
| if (numRequested == Long.MAX_VALUE) { | ||
| // unlimited demand | ||
| subscriptionWorker.schedulePeriodically( | ||
| new NonBlockingUnlimitedDemandPullTask(subscriptionName, sink), 0, pollingPeriodMs, TimeUnit.MILLISECONDS); | ||
| } | ||
| else { | ||
| subscriptionWorker.schedule(new BlockingLimitedDemandPullTask(subscriptionName, numRequested, sink)); | ||
| } | ||
| }); | ||
|
|
||
| sink.onCancel(subscriptionWorker); | ||
|
|
||
| }); | ||
| } | ||
|
|
||
| private abstract class PubSubPullTask implements Runnable { | ||
|
|
||
| protected final String subscriptionName; | ||
|
|
||
| protected final FluxSink<AcknowledgeablePubsubMessage> sink; | ||
|
|
||
| PubSubPullTask(String subscriptionName, FluxSink<AcknowledgeablePubsubMessage> sink) { | ||
| this.subscriptionName = subscriptionName; | ||
| this.sink = sink; | ||
| } | ||
|
|
||
| /** | ||
| * Retrieve up to a specified number of messages, sending them to the subscription. | ||
| * @param demand maximum number of messages to retrieve | ||
| * @param block whether to wait for the first message to become available | ||
| * @return number of messages retrieved | ||
| */ | ||
| protected int pullToSink(int demand, boolean block) { | ||
|
|
||
| List<AcknowledgeablePubsubMessage> messages = | ||
| PubSubReactiveFactory.this.subscriberOperations.pull(this.subscriptionName, demand, !block); | ||
|
|
||
| if (!this.sink.isCancelled()) { | ||
| messages.forEach(sink::next); | ||
| } | ||
|
|
||
| return messages.size(); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * Runnable task issuing blocking Pub/Sub Pull requests until the specified number of | ||
| * messages has been retrieved. | ||
| */ | ||
| private class BlockingLimitedDemandPullTask extends PubSubPullTask { | ||
|
|
||
| private final long initialDemand; | ||
|
|
||
| BlockingLimitedDemandPullTask(String subscriptionName, long initialDemand, FluxSink<AcknowledgeablePubsubMessage> sink) { | ||
| super(subscriptionName, sink); | ||
| this.initialDemand = initialDemand; | ||
| } | ||
|
|
||
| @Override | ||
| public void run() { | ||
| long demand = this.initialDemand; | ||
| List<AcknowledgeablePubsubMessage> messages; | ||
|
|
||
| while (demand > 0 && !this.sink.isCancelled()) { | ||
| try { | ||
| int intDemand = demand > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) demand; | ||
| demand -= pullToSink(intDemand, true); | ||
| } | ||
| catch (DeadlineExceededException e) { | ||
| if (LOGGER.isTraceEnabled()) { | ||
| LOGGER.trace("Blocking pull timed out due to empty subscription " | ||
| + this.subscriptionName | ||
| + "; retrying."); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * Runnable task issuing a single Pub/Sub Pull request for all available messages. | ||
| * Terminates immediately if no messages are available. | ||
| */ | ||
| private class NonBlockingUnlimitedDemandPullTask extends PubSubPullTask { | ||
|
|
||
| NonBlockingUnlimitedDemandPullTask(String subscriptionName, FluxSink<AcknowledgeablePubsubMessage> sink) { | ||
| super(subscriptionName, sink); | ||
| } | ||
|
|
||
| @Override | ||
| public void run() { | ||
| pullToSink(Integer.MAX_VALUE, false); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } |
Oops, something went wrong.