Skip to content

Commit

Permalink
Change adaptPollableChannelToPublisher to Mono
Browse files Browse the repository at this point in the history
We can use a `Mono.fromCallable()` in combination with `repeat()`
instead of possible race condition deal in the `Flux.create()`
  • Loading branch information
artembilan committed Feb 12, 2020
1 parent 1ff69d4 commit a0cdfd9
Showing 1 changed file with 8 additions and 28 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,9 @@

package org.springframework.integration.channel;

import java.time.Duration;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
Expand All @@ -27,6 +28,7 @@

import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

/**
Expand Down Expand Up @@ -69,34 +71,12 @@ private static <T> Publisher<Message<T>> adaptSubscribableChannelToPublisher(Sub
});
}

@SuppressWarnings("unchecked")
private static <T> Publisher<Message<T>> adaptPollableChannelToPublisher(PollableChannel inputChannel) {
return new PollableChannelPublisherAdapter<>(inputChannel);
}

private static final class PollableChannelPublisherAdapter<T> implements Publisher<Message<T>> {

private final PollableChannel channel;

PollableChannelPublisherAdapter(final PollableChannel channel) {
this.channel = channel;
}

@Override
@SuppressWarnings("unchecked")
public void subscribe(Subscriber<? super Message<T>> subscriber) {
Flux
.<Message<T>>create(sink ->
sink.onRequest(n -> {
Message<?> m;
while (!sink.isCancelled() && n-- > 0
&& (m = this.channel.receive()) != null) { // NOSONAR
sink.next((Message<T>) m);
}
}))
.subscribeOn(Schedulers.elastic())
.subscribe(subscriber);
}

return Mono.fromCallable(() -> (Message<T>) inputChannel.receive())
.subscribeOn(Schedulers.boundedElastic())
.repeatWhenEmpty(it -> it.delayElements(Duration.ofMillis(10)))
.repeat();
}

}

0 comments on commit a0cdfd9

Please sign in to comment.