From 012bebd7bfb680760c29d17951d91d262234ed86 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 13 Jul 2017 17:20:16 -0400 Subject: [PATCH] Impl CoreSubscriber instead of Subscriber 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` --- .../endpoint/ReactiveStreamsConsumer.java | 3 ++- .../handler/AbstractMessageHandler.java | 5 +++-- .../AttributePollingChannelAdapterParserTests.java | 14 ++++++++------ 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java index 1ae2f01007..4e29b0ce87 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java @@ -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>, Disposable, Lifecycle { + implements CoreSubscriber>, Disposable, Lifecycle { private final Consumer> consumer; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index ada1c4ef78..cf577671e7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -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, TrackableComponent, Orderable, - Subscriber> { + CoreSubscriber> { private volatile boolean shouldTrack = false; diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/AttributePollingChannelAdapterParserTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/AttributePollingChannelAdapterParserTests.java index 6939a93c46..6f62fb9321 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/AttributePollingChannelAdapterParserTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/AttributePollingChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2017 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. @@ -37,6 +37,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Mark Fisher * @author Gary Russell + * * @since 2.0 */ @ContextConfiguration @@ -56,21 +57,22 @@ public class AttributePollingChannelAdapterParserTests { @Autowired private MessageChannel autoChannel; - @Autowired @Qualifier("autoChannel.adapter") + @Autowired + @Qualifier("autoChannel.adapter") private SourcePollingChannelAdapter autoChannelAdapter; @Test public void pollForAttribute() throws Exception { - testBean.test("foo"); - adapter.start(); - Message result = channel.receive(1000); + this.testBean.test("foo"); + this.adapter.start(); + Message result = this.channel.receive(10000); assertNotNull(result); assertEquals("foo", result.getPayload()); } @Test public void testAutoChannel() { - assertSame(autoChannel, TestUtils.getPropertyValue(autoChannelAdapter, "outputChannel")); + assertSame(this.autoChannel, TestUtils.getPropertyValue(this.autoChannelAdapter, "outputChannel")); } }