Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Impl CoreSubscriber instead of Subscriber

Browse Source 
According Reactor recommendation it would be better to `CoreSubscriber`
from Reactor instead of Reactive Streams `Subscriber` for chasing
some optimisation in case of Reactor flows

Increase `receive()` timeout in the `AttributePollingChannelAdapterParserTests`
This commit is contained in:
Artem Bilan
2017-07-13 17:20:16 -04:00
parent 6829da995f
commit 012bebd7bf
3 changed files with 13 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
View File

@@ -34,6 +34,7 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
import reactor.core.CoreSubscriber;
import reactor.core.Disposable;
import reactor.core.publisher.BaseSubscriber;
@@ -165,7 +166,7 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
private static final class MessageHandlerSubscriber
implements Subscriber<Message<?>>, Disposable, Lifecycle {
implements CoreSubscriber<Message<?>>, Disposable, Lifecycle {
private final Consumer<Message<?>> consumer;

5
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java
View File

@@ -16,7 +16,6 @@
package org.springframework.integration.handler;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.core.Ordered;
@@ -37,6 +36,8 @@ import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
import reactor.core.CoreSubscriber;
/**
* Base class for MessageHandler implementations that provides basic validation
* and error handling capabilities. Asserts that the incoming Message is not
@@ -50,7 +51,7 @@ import org.springframework.util.Assert;
@IntegrationManagedResource
public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler,
MessageHandlerMetrics, ConfigurableMetricsAware<AbstractMessageHandlerMetrics>, TrackableComponent, Orderable,
Subscriber<Message<?>> {
CoreSubscriber<Message<?>> {
private volatile boolean shouldTrack = false;