Reactive support for Pub/Sub subscription #1461

elefeint · 2019-02-08T19:15:42Z

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.

Fixes #186.

artembilan

Some review.
Good stuff, but more questions, than answers 😄

artembilan · 2019-02-08T19:43:40Z

...es/spring-cloud-gcp-pubsub-reactive-sample/src/main/java/com/example/ReactiveController.java

+	@GetMapping(value = "/getmessages", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+	public Flux<? super String> getMessages() {
+		Flux<ConvertedAcknowledgeablePubsubMessage<String>> flux
+				= FluxSubscriber.createPolledFlux(this.template, String.class);


Doesn't look like conversion logic propagation is a good idea in terms of Reactive Streams.
You just need to request data and then downstream in the .map() decide to what to convert it into.

artembilan · 2019-02-08T19:46:06Z

...d-gcp-pubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/FluxSubscriber.java

+ * @author Elena Felder
+ * @since 1.2
+ */
+public class FluxSubscriber {


The name of the class is misleading. There is already a Subscriber abstraction in the Reactive Streams and Flux just simple deal with them as is.

Essentially what you did here is a factory. so, please, reconsider the class name and I believe something like PubSub must be included in its name.

Something like PubSubSubscriberFactory?
Then we can move the .subscribe() methods into this, as we discussed before for the refactoring.

On the other hand, we already have a SubscriberFactory, so I could just add the Flux creation logic into the existing DefaultSubscriberFactory.

You can if reactor-core is hard dependency, not optional.

Hmm, good point -- it should probably be optional. so it should not sit in a core class. Reactive is not the most common usecase.

artembilan · 2019-02-08T19:46:55Z

...d-gcp-pubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/FluxSubscriber.java

+
+				LOGGER.info("  Current remaining demand: " + remainingDemand);
+				List<ConvertedAcknowledgeablePubsubMessage<T>> messages
+						= this.subscriberOperations.pullAndConvert("reactiveSubscription",


M-m-m. I think this is not intentional.
Why do you use an explicit, non-configurable subscription name ?

Definitely not intentional; I'll move it into a parameter.

artembilan · 2019-02-08T19:48:18Z

...d-gcp-pubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/FluxSubscriber.java

+		return Flux.create(sink -> {
+			sink.onRequest((numRequested) -> {
+				// request must be non-blocking.
+				executorService.submit(new PubSubPullTask<T>(numRequested, sink, subscriberOperations, targetType));


I think that not your responsibility to do this.
What you, essentially, have is simply covered by the subscribeOn().

Although since underlying API is really not reactive (PullResponse):

this.receivedMessages_.add(input.readMessage(ReceivedMessage.parser(), extensionRegistry));

Does it really make sense to promote this from the Framework perspective?..

The problem is that I am making a blocking pull call here, so if I request on the same thread, the code will violate the reactive streams rule that request should be non-obstructing.

There might be a way to do the pull asynchronously without your own ExecutorService.
See here: https://github.com/googleapis/google-cloud-java/blob/master/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriberSnippets.java#L213

subscriber.pullCallable().futureCall()

It's hard for me to object, but I think your thread spawning per subscriber is still bad idea.
Consider to use somehow Schedulers.elastic().

I will try to pull some one from our Reactor team to see what that think on the matter...

Oh! futureCall(). that's an answer! Right., we need to add it into the PubSubSubscriberOperations API and use it from this Flux trick.
There is simply need to be have a flatMap(e -> Mono.fromFuture()) and so on with flatMapIterable().

futureCall() sounded good in theory, but now we have potentially multiple outstanding calls to PubSub with additional callbacks being spawned.
Flux making an additional request when the data is 75% or so processed, combined with PubSub synchronous pull timing out after 10 minutes makes for messy logic.

I'd almost prefer to take a "poll interval" parameter, accumulate total outstanding demand in the custom flux, and then make "return_immediately=true" requests at an interval.

artembilan · 2019-02-08T19:59:57Z

...d-gcp-pubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/FluxSubscriber.java

+				int numReceived = messages.size();
+				messages.forEach(m -> {
+					LOGGER.info("Sending message to subscriber: " + m.getPayload());
+					sink.next(m);


I would send the whole list as a single event, then would have have a flatMapIterable() before returning from the createPolledFlux()

…coded subscription name

codecov-io · 2019-02-15T20:05:33Z

Codecov Report

Merging #1461 into master will decrease coverage by 10.13%.
The diff coverage is 89.13%.

@@             Coverage Diff              @@
##             master   #1461       +/-   ##
============================================
- Coverage     78.94%   68.8%   -10.14%     
+ Complexity     1989    1692      -297     
============================================
  Files           244     246        +2     
  Lines          6526    6572       +46     
  Branches        662     667        +5     
============================================
- Hits           5152    4522      -630     
- Misses         1094    1755      +661     
- Partials        280     295       +15

Flag	Coverage Δ	Complexity Δ
#integration	`?`	`?`
#unittests	`68.8% <89.13%> (+0.14%)`	`1692 <9> (+9)`	⬆️

Impacted Files	Coverage Δ	Complexity Δ
...toconfigure/pubsub/GcpPubSubAutoConfiguration.java	`90.47% <ø> (ø)`	`38 <0> (ø)`	⬇️
...ure/pubsub/GcpPubSubReactiveAutoConfiguration.java	`100% <100%> (ø)`	`3 <3> (?)`
...oud/gcp/pubsub/reactive/PubSubReactiveFactory.java	`88.37% <88.37%> (ø)`	`6 <6> (?)`
...main/kotlin/com/example/app/PubSubConfiguration.kt	`0% <0%> (-100%)`	`0% <0%> (-5%)`
...ample/src/main/java/com/example/WebController.java	`0% <0%> (-100%)`	`0% <0%> (-3%)`
...a/spanner/repository/query/SpannerQueryMethod.java	`0% <0%> (-100%)`	`0% <0%> (-5%)`
...ple/src/main/java/com/example/MyAppProperties.java	`0% <0%> (-100%)`	`0% <0%> (-5%)`
...ple/src/main/java/com/example/HelloController.java	`0% <0%> (-100%)`	`0% <0%> (-2%)`
...ample/src/main/java/com/example/BluegrassBand.java	`0% <0%> (-100%)`	`0% <0%> (-2%)`
...figure/config/GcpConfigBootstrapConfiguration.java	`0% <0%> (-100%)`	`0% <0%> (-2%)`
... and 74 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 72d7464...3ba8683. Read the comment docs.

elefeint · 2019-02-21T20:52:13Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+
+					if (remainingDemand > 0) {
+						// Ensure requested messages get fulfilled eventually.
+						produceMessages(subscriptionName, remainingDemand, sink, subscriberOperations);


Between the need to keep onRequest() non-blocking and attempting to fulfill outstanding demand, the code relying on gRPC's futureCall() became messy and with potentially too many simultaneous calls to Pub/Sub. And adding callbacks to ApiFuture requires a separate executor, anyway (I reused the former ackExecutor in the latest implementation).

I would prefer to go back to maintaining a thread pool (but a shared one, not one per flux) in PubSubReactiveFactory, and using non-blocking polling at a customizable interval to satisfy remaining demand.

@artembilan @meltsufin What do you think?

Yeah, the call stack can get huge with this recursive call. So, it might make sense to go back to the synchronous pull with a pooled executor. It doesn't look like we're able to get away from providing our own executor anyway. WDYT @artembilan?

I would say that's not our problem to deal with remained here and die in the stack overflow.

Please, see Subscription.request() JavaDocs:

* A {@link Publisher} can send less than is requested if the stream ends but * then must emit either {@link Subscriber#onError(Throwable)} or {@link Subscriber#onComplete()}.

so, our goal here is to produce exactly requested amount or less and then onComplete(): FluxSink.complete().

Not sure why have you gone a bit different way...

Also the rule number 1 of Publisher specification: https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2/README.md#specification

The total number of onNext´s signalled by a Publisher to a Subscriber MUST be less than or equal to the total number of elements requested by that Subscriber´s Subscription at all times.

I think you are missing the fact that our createPolledPublisher() is just finite and that's already out of our scope to keep it hot and constantly updating.

Is a cold publisher useful in Pub/Sub context? If the requirement is "get up to N messages and complete", then the client code can just use generate:

Flux.<List<AcknowledgeablePubsubMessage>>generate( () -> subscriberOperations.pull(subscriptionName, demand, true) ).flatMapIterable(Function.identity());

It seems to me based on discussion in reactor/reactor-core#314 that onRequest() was added to support our exact usecase of an unlimited flux that still responds to backpressure.

Off-topic, if I start a metal band, "We Die in the Stack Overflow" will be its name.

Hm... Then I have no answer or my Reactive knowledge is limited...

I wonder if we can bring someone who is an expert in the area to give us some feedback...

Well, everything I learned about Reactive I learned in the last month, so I am hardly an expert. Could we maybe set up a meeting with Reactive folks?

Thinking about this a bit more, I'm not sure what's wrong with just returning whatever we get from Pub/Sub with the return-immediately=true, and not trying to fulfill the full demand. If Pub/Sub returns less than the demand, we just call onComplete assuming there are no more messages for the time being. The client can always resume its requests.
What's our outstanding question for the experts? I think the specification is fairly precise.

Ok, just talked to @elefeint. I see why it's bad to terminate the stream when the demand is unmet by Pub/Sub. It doesn't work for infinite streams. So, it looks like it's better to keep the subscription open knowing that Pub/Sub can eventually provide more messages. We just need to poll it on a regular basis. Ideally, Pub/Sub would support true reactive streaming itself, but for the time being we need to simulate it by polling.

Hey that'd be great to meet if you folks want to introduce us to this in a zoom or somewhere and we can pair-review together !

artembilan · 2019-02-21T22:16:15Z

...main/java/org/springframework/cloud/gcp/pubsub/core/subscriber/PubSubSubscriberTemplate.java

-	public void setAckExecutor(Executor ackExecutor) {
-		Assert.notNull(ackExecutor, "ackExecutor can't be null.");
-		this.ackExecutor = ackExecutor;
+	public void setCallbackExecutor(Executor callbackExecutor) {


Do you realize that this is a breaking change? We just can't rename API just because.
Please, consider to deprecate exiting method in favor of a new one.
And let's keep it here at least for the next version!

Point well taken. If this is the direction in which this code will go, I'll add the proper deprecation (I hope it does not go there, though -- per discussion below, I would prefer to revert the changes to PubSubSubscriberTemplate and manage multi threading in PubSubReactiveFactory instead).

artembilan · 2019-02-21T22:24:35Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+
+					if (remainingDemand > 0) {
+						// Ensure requested messages get fulfilled eventually.
+						produceMessages(subscriptionName, remainingDemand, sink, subscriberOperations);


I would say that's not our problem to deal with remained here and die in the stack overflow.

Please, see Subscription.request() JavaDocs:

* A {@link Publisher} can send less than is requested if the stream ends but * then must emit either {@link Subscriber#onError(Throwable)} or {@link Subscriber#onComplete()}.

so, our goal here is to produce exactly requested amount or less and then onComplete(): FluxSink.complete().

Not sure why have you gone a bit different way...

Also the rule number 1 of Publisher specification: https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2/README.md#specification

The total number of onNext´s signalled by a Publisher to a Subscriber MUST be less than or equal to the total number of elements requested by that Subscriber´s Subscription at all times.

I think you are missing the fact that our createPolledPublisher() is just finite and that's already out of our scope to keep it hot and constantly updating.

elefeint

From in-person review with @smaldini @artembilan @meltsufin

elefeint · 2019-02-25T20:12:40Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+
+		PubSubReactiveSubscription reactiveSubscription = new PubSubReactiveSubscription(subscriptionName);
+
+		return Flux.interval(Duration.ofMillis(delayInMilliseconds))


onBackpressureDrop()

elefeint · 2019-02-25T20:13:18Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+
+		return Flux.interval(Duration.ofMillis(delayInMilliseconds))
+				.doOnRequest(reactiveSubscription::addDemand)
+				.map(t -> reactiveSubscription.pullMessages())


collapse map and flatmap into one operation

elefeint · 2019-02-25T20:34:11Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+	public Publisher<AcknowledgeablePubsubMessage> createPolledPublisher(
+			String subscriptionName, int delayInMilliseconds) {
+
+		PubSubReactiveSubscription reactiveSubscription = new PubSubReactiveSubscription(subscriptionName);


avoid creating subscription object outside of flux creation.
bottom line: use Flux.create for more control over the process

…to make it global

…spring-integration

elefeint · 2019-03-06T21:34:00Z

@smaldini @artembilan @meltsufin I've changed the Pub/Sub flux creation implementation as we discussed last Monday (with Flux.create and special treatment of unlimited demand). Could you take another look?

I still owe tests.

artembilan

Nothing critical.
The solution looks great! 👍

...main/java/org/springframework/cloud/gcp/autoconfigure/pubsub/GcpPubSubAutoConfiguration.java

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

artembilan · 2019-03-07T15:35:51Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+	}
+
+	/**
+	 * Creates an infinite stream {@link Publisher} of {@link AcknowledgeablePubsubMessage} objects.


Method JavaDocs must be in the imperative style: https://github.com/spring-projects/spring-framework/wiki/Code-Style#javadoc-formatting

artembilan · 2019-03-07T15:37:34Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+	 * @param pollingPeriod how frequently to poll the source subscription in case of unlimited demand.
+	 * @return infinite stream of {@link AcknowledgeablePubsubMessage} objects.
+	 */
+	public Publisher<AcknowledgeablePubsubMessage> createPolledPublisher(String subscriptionName, long pollingPeriod) {


I think our (Spring) programming module to expect Reactive Streams interfaces as method arguments, but return just an explicit Reacor types: Flux and Mono.
No reason to narrow it if you deal internally with only Reactor.

artembilan · 2019-03-07T15:41:46Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+				messages.forEach(sink::next);
+			}
+
+			return messages;


I think we may end up with the contract just really return an amount we have just pulled, not the whole list of messages.

artembilan · 2019-03-07T15:42:50Z

spring-cloud-gcp-samples/spring-cloud-gcp-pubsub-reactive-sample/pom.xml

+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+			<optional>true</optional>


I believe it can be non-optional here already.
This is end-user code, so that is fully OK to have it as a hard dependency.

artembilan · 2019-03-07T15:44:29Z

...es/spring-cloud-gcp-pubsub-reactive-sample/src/main/java/com/example/ReactiveController.java

+		return Flux.from(flux)
+				.doOnNext(message -> {
+					System.out.println("Received a message: " + message.getPubsubMessage().getMessageId());
+					message.ack();


I think we need to document this manual ack'ing or maybe consider an auto-ack mode as well...

Customizable ack mode is a good idea. Would it make sense to move (properly, with deprecation) AckMode from o.s.c.gcp.pubsub.integration up a level, or to simply use a boolean autoAck parameter when creating the Flux?

I think boolean autoAck is fully enough, but let's really consider this improvement when we are already done with basics in this PR!

I created #1517 to track; it will be a small follow-up PR.

…he messages

docs/src/main/asciidoc/pubsub.adoc

meltsufin · 2019-03-08T16:49:38Z

docs/src/main/asciidoc/pubsub.adoc

+
+The `Flux` then represents an infinite stream of GCP Pub/Sub messages coming in through the specified subscription.
+The full range of Project Reactor operations can be applied to the stream.
+For example, the infinite stream can become a finite one with `limitRequest` operation:


I think we should also use simpler language, like "If you only want to fetch 5 messages from Pub/Sub, you can use a finite stream...".

meltsufin · 2019-03-08T16:52:41Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+	 * @param pollingPeriod how frequently to poll the source subscription in case of unlimited demand.
+	 * @return infinite stream of {@link AcknowledgeablePubsubMessage} objects.
+	 */
+	public Flux<AcknowledgeablePubsubMessage> createPolledFlux(String subscriptionName, long pollingPeriod) {


Units for pollingPeriod should be clarified. I would name the parameter pollingPeriodMs and clarify units in the javadoc.

meltsufin · 2019-03-08T16:58:53Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+		 * @return number of messages retrieved
+		 */
+		protected int pullToSink(long demand, boolean block) {
+			int numMessagesToPull = demand > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) demand;


Why not just only allow int demand as parameter and let the caller deal with conversion?

The demand comes from FluxSink.onRequest() parameter, which is a Long. Ultimately, there is tension between Reactor (which operates on Long demand) and PubSub (which operates on int demand), so it has to get resolved somewhere in my code.

Or did you mean to simply move the conversion from PubSubBlockingLimitedDemandPullTask.run() up to PubSubReactiveFactory.createPolledFlux.createPolledFlux()?

Right, hand the responsibility to the caller.

I looked at the code again -- the responsibility for converting from long to int belongs with the polling task, as close to the Pub/Sub polling operation is possible. Consider a request of Integer.MAX_VALUE + 1,000. If we convert to int before creating PubSubBlockingLimitedDemandPullTask, then only a demand of Integer.MAX_VALUE will be fulfilled, leaving the extra demand of 1,000 unmet.

Leaving the conversion responsibility with PubSubBlockingLimitedDemandPullTask allows it to think in terms of long internally, subtracting up to Integer.MAX_VALUE messages at each poll.

I think the conversion belongs in createPolledFlux. How would the demand of Integer.MAX_VALUE + 1,000 be fulfilled, if you cap it at Integer.MAX_VALUE here? Automatically changing demand > Integer.MAX_VALUE to Integer.MAX_VALUE seems a bug waiting to happen. If the flux ever requests a demand greater than Integer.MAX_VALUE and it's not the special Long.MAX_VALUE we should probably just throw an unsupported exception in createPolledFlux.

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

artembilan · 2019-03-11T14:58:54Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+
+	private static final Log LOGGER = LogFactory.getLog(PubSubReactiveFactory.class);
+
+	private PubSubSubscriberOperations subscriberOperations;


artembilan · 2019-03-11T15:03:36Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+					this.subscriptionName, numMessagesToPull, !block);
+
+			if (!this.sink.isCancelled()) {
+				messages.forEach(sink::next);


I wonder what is wrong with sending the whole List<AcknowledgeablePubsubMessage> batch to the sink as a single event and then use a flatMapIterable(Function.identity()) before returning the Flux to end-user...

flatMapIterable changes demand, which could look unintuitive to the consumer (if subscription requests 3 messages, flatMapIterable() will cause 256 to be requested instead).

Also (this is not a good reason, just relevant) when I was testing, I ran into an issue with additional demand not being propagated from HTTP request to the Pub/Sub flux even though the initial request was fulfilled. I'd like to reproduce a minimal case to understand whether this behaves as intended or is a bug.

OK. Then let's deffer it to @smaldini .

I'm not sure how to be with this situation then 😄

Hi! Great work @elefeint! We were also thinking of implementing our own Reactor Pubsub client due to improper back-pressure semantics which is giving us quite a lot of headache and thankfully @bsideup pointed this ticket to us. About this specific messages.forEach(sink::next) line, I have a question: If the demand is a big number, wouldn't you be fetching a gigantic batch of messages from Pubsub and choking the consumer via over-production? I would rather cap it by Math.min(demand, maxDemand) (e.g., maxDemand=100 by default) for each request(demand) and wait for any follow up request()s that the consumer will signal at its convenience. Otherwise my impression is that we will be periodically bombarding the consumer with messages regardless of its availability. Is there a detail I am missing?

If the consumer is specifying the large demand, isn't it a signal that it can handle that number of messages? Certainly throttling can be implemented, but it may be unintuitive to consumers that do want a larger batch.

Most of the time consumers do not specify anything and simple subscribe() just triggers a request(Long.MAX_VALUE). Though in the case of a message queue poller, "Apparently the consumer is asking for a cargo ship load of messages." does not sound like a practical assumption to me. Consider the following case:

There are 1e6 messages in the queue.

request(unbounded) triggers a continuous messages.forEach(sink::next) sequence.

Scheduler threads start working on the initial tasks and the rest quickly get piled up in the work queue of the ScheduledExecutorService.

After a while, majority of the messages timeout, work queue starts choking memory, and every task threads pick up turn out to be obsolete.

This is the problem I am dealing for the last couple of months at work. Let me know if there is a way I can work around this.

@vy I understand now. As a practical matter, would adding limitRequest() or limitRate() operators to the invocation chain work?

@smaldini The conversation about default behavior of subscribe() is an interesting one.
Would it make sense to open an issue on reactor-core? Unbounded demand is not the most common usecase, so perhaps a .subscribeWithInitialDemand() operator makes sense?

@elefeint I suppose a combination of limit{Request,Rate}() would do the trick, though it still expects an explicit action from the consumer side. I would still favor shipping a solution with decent defaults minimizing any effort from user side.

I think subscribeWithInitialDemand() is sort of against the Reactor's fluent API encouraging composition. Isn't it just a shortcut for .limitRequest(n).subscribe()?

Nevertheless, @smaldini, we would really appreciate your insights on the general back-pressure issue. (@simonbasle, is this something you can help us out?)

FYI limitRate only, not limitRequest. limitRequest will cut the requests after the threshold specified as an argument

@vy Would you mind creating an issue in this project for adding the maxMessagesPerPoll property? I've been thinking about this some more, and for unbounded demand scenario it makes sense to provide a configurable limit, especially if it's combined with a configurable project-wide default for pollingPeriodMs.
The firehose of messages would likely still be the default, but smoothing message arrival by getting N messages per poll seems valuable.

Co-Authored-By: elefeint <41136058+elefeint@users.noreply.github.com>

elefeint

from in-person review

elefeint · 2019-03-20T17:11:01Z

...main/java/org/springframework/cloud/gcp/autoconfigure/pubsub/GcpPubSubAutoConfiguration.java

+	protected static class ReactivePubSubAutoconfiguration {
+
+		@Bean
+		@ConditionalOnMissingBean


Should there be a property preventing this object from instantiating?

For what reason?
Doesn't look like this light object is some kind of overhead for the target application.

Right, it's not necessary to keep it off by default because it's just a wrapper around a SubscriberTemplate.
But I am also going to add a default scheduler bean into autoconfig and extract reactive configuration into a separate file, at which point it will make sense to add an overridable matchIfMissing=true property spring.cloud.gcp.pubsub.reactive.enabled.

elefeint · 2019-03-20T17:12:23Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+					// unlimited demand
+					Disposable task = Schedulers.single().schedulePeriodically(
+							new PubSubNonBlockingUnlimitedDemandPullTask(subscriptionName, sink), 0, pollingPeriodMs, TimeUnit.MILLISECONDS);
+					sink.onCancel(task::dispose);


pass task directly

elefeint · 2019-03-20T17:15:21Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+	 * @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> createPolledFlux(String subscriptionName, long pollingPeriodMs) {


rename to poll(subscriptionName, pollingPeriodMs)

elefeint · 2019-03-20T17:17:49Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+				if (numRequested == Long.MAX_VALUE) {
+					// unlimited demand
+					Disposable task = Schedulers.single().schedulePeriodically(
+							new PubSubNonBlockingUnlimitedDemandPullTask(subscriptionName, sink), 0, pollingPeriodMs, TimeUnit.MILLISECONDS);


remove PubSub prefix

elefeint · 2019-03-20T17:21:52Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+					demand -= pullToSink(intDemand, true);
+				}
+				catch (DeadlineExceededException e) {
+					LOGGER.trace("Blocking pull timed out due to empty subscription " + this.subscriptionName + "; retrying.");


add isTraceEnabled()

No reason: this is in the catch() block, so only in case of exception.
It's not going to be a big deal to build a logging message in case of this exception.

This exception is actually expected. So, it may happen quiet often.

Every 10 minutes by default, which is negligible in computer speed. On the other hand, we might as well add isTraceEnabled() for consistency.

elefeint · 2019-03-20T17:31:23Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+					sink.onCancel(task::dispose);
+				}
+				else {
+					Schedulers.single().schedule(new PubSubBlockingLimitedDemandPullTask(subscriptionName, numRequested, sink));


use elastic(), create single worker (don't call scheduler() directly)
create worker before request

also, kill task when sink is cancelled -- 10 minutes is too long

reuse the worker for both, unlimited and limited demand

autoconfigure the scheduler

...es/spring-cloud-gcp-pubsub-reactive-sample/src/main/java/com/example/ReactiveController.java

artembilan · 2019-03-20T18:22:27Z

...main/java/org/springframework/cloud/gcp/autoconfigure/pubsub/GcpPubSubAutoConfiguration.java

+	protected static class ReactivePubSubAutoconfiguration {
+
+		@Bean
+		@ConditionalOnMissingBean


For what reason?
Doesn't look like this light object is some kind of overhead for the target application.

artembilan · 2019-03-20T18:24:23Z

...ubsub/src/main/java/org/springframework/cloud/gcp/pubsub/reactive/PubSubReactiveFactory.java

+					demand -= pullToSink(intDemand, true);
+				}
+				catch (DeadlineExceededException e) {
+					LOGGER.trace("Blocking pull timed out due to empty subscription " + this.subscriptionName + "; retrying.");


No reason: this is in the catch() block, so only in case of exception.
It's not going to be a big deal to build a logging message in case of this exception.

meltsufin

Changes look good!
Would it be possible to add a integration test for the sample, like we have for other samples?

...a/org/springframework/cloud/gcp/autoconfigure/pubsub/GcpPubSubReactiveAutoConfiguration.java

Co-Authored-By: elefeint <41136058+elefeint@users.noreply.github.com>

elefeint · 2019-03-21T17:33:24Z

I'll add an integration test in a follow-up PR -- I also want to change the sample, so users can send messages through a web UI and not have to have gcloud installed or multi-task with a third command-line window.

This commit updates URLs to prefer the https protocol.

smaldini

Nice job @elefeint and team!

wip reactive polling pubsub

d3f0dd3

elefeint requested review from artembilan and meltsufin February 8, 2019 19:15

artembilan suggested changes Feb 8, 2019

View reviewed changes

elefeint added 2 commits February 14, 2019 14:11

rename flux factory; make return type generic Publisher; extract hard…

cf7a419

…coded subscription name

rely on async implementation of sync pull to deliver remaining demand

78ff7e4

remove unused parameter

6f88ee5

elefeint commented Feb 21, 2019

View reviewed changes

artembilan suggested changes Feb 21, 2019

View reviewed changes

elefeint added 4 commits February 22, 2019 13:57

replace flux implementation with polling; clean up api

83284ff

remove no longer needed PubSubTemplate pullAsync changes

22688cb

remove pubsub ack executor renaming in autoconfig

cbdb23f

remove logging; style cleanup

f04839b

meltsufin added the pubsub GCP PubSub label Feb 22, 2019

replace custom executor with Flux.interval()

620c2b4

elefeint commented Feb 25, 2019

View reviewed changes

elefeint added 7 commits March 4, 2019 17:59

wip do not review; initial refactoring to address feedback

001bb1e

move polling period customizable per flux again; it was a bad change …

5b542d2

…to make it global

Merge branch 'master' into pubsub-reactive

961717e

autoconfigure PubSubReactiveFactory if Reactor is on classpath

ec192d5

add explicit reactor dependencies, so it's not just a side effect of …

072efef

…spring-integration

add docs

64be90e

remove unnecessary logging

b0822fd

artembilan suggested changes Mar 7, 2019

View reviewed changes

elefeint added 3 commits March 7, 2019 17:30

address feedback and fix style

e912a6f

document manual acking, as a placeholder

3ed80d1

prevent subscription cancelled while blocked on pull from receiving t…

5ca2446

…he messages

meltsufin suggested changes Mar 8, 2019

View reviewed changes

add tests

abe1cc0

artembilan suggested changes Mar 11, 2019

View reviewed changes

meltsufin and others added 2 commits March 11, 2019 11:19

Apply suggestions from code review

5e59792

Co-Authored-By: elefeint <41136058+elefeint@users.noreply.github.com>

address feedback

332f889

elefeint changed the title ~~wip reactive polling pubsub~~ Reactive support for Pub/Sub subscription Mar 11, 2019

move long-to-int conversion to class that keeps track of long demand

af1a4eb

This was referenced Mar 11, 2019

Add autoacking to Pub/Sub reactive stream #1517

Closed

Add reactive binder for PubSub #186

Closed

Spring Operator and others added 2 commits March 18, 2019 11:29

Merge branch 'master' into pubsub-reactive

4ec6365

elefeint commented Mar 20, 2019

View reviewed changes

artembilan suggested changes Mar 20, 2019

View reviewed changes

address feedback

e96adcd

meltsufin reviewed Mar 20, 2019

View reviewed changes

...a/org/springframework/cloud/gcp/autoconfigure/pubsub/GcpPubSubReactiveAutoConfiguration.java Outdated Show resolved Hide resolved

...a/org/springframework/cloud/gcp/autoconfigure/pubsub/GcpPubSubReactiveAutoConfiguration.java Show resolved Hide resolved

Apply suggestions from code review

e33a3e4

Co-Authored-By: elefeint <41136058+elefeint@users.noreply.github.com>

artembilan previously approved these changes Mar 21, 2019

View reviewed changes

URL Cleanup (#1561)

651c134

This commit updates URLs to prefer the https protocol.

dzou dismissed artembilan’s stale review via 651c134 March 21, 2019 18:28

Merge branch 'master' into pubsub-reactive

3ba8683

meltsufin approved these changes Mar 21, 2019

View reviewed changes

smaldini approved these changes Mar 22, 2019

View reviewed changes

elefeint merged commit 8df2a6b into master Mar 22, 2019

elefeint deleted the pubsub-reactive branch March 22, 2019 18:10


		PubSubReactiveSubscription reactiveSubscription = new PubSubReactiveSubscription(subscriptionName);

		return Flux.interval(Duration.ofMillis(delayInMilliseconds))


		private static final Log LOGGER = LogFactory.getLog(PubSubReactiveFactory.class);

		private PubSubSubscriberOperations subscriberOperations;

Reactive support for Pub/Sub subscription #1461

Reactive support for Pub/Sub subscription #1461

Conversation

elefeint commented Feb 8, 2019 • edited Loading

artembilan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-io commented Feb 15, 2019 • edited by codecov bot Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

elefeint left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

elefeint commented Mar 6, 2019

artembilan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

elefeint Apr 11, 2019 • edited Loading

Choose a reason for hiding this comment

elefeint left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

elefeint commented Feb 8, 2019 •

edited

Loading

codecov-io commented Feb 15, 2019 •

edited by codecov bot

Loading

elefeint Apr 11, 2019 •

edited

Loading