Change adaptPollableChannelToPublisher to Mono
We can use a `Mono.fromCallable()` in combination with `repeat()` instead of possible race condition deal in the `Flux.create()`
This commit is contained in:
@@ -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;
|
||||
@@ -27,6 +28,7 @@ import org.springframework.messaging.SubscribableChannel;
|
||||
|
||||
import reactor.core.publisher.EmitterProcessor;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.scheduler.Schedulers;
|
||||
|
||||
/**
|
||||
@@ -69,34 +71,12 @@ public final class MessageChannelReactiveUtils {
|
||||
});
|
||||
}
|
||||
|
||||
@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();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user