diff --git a/build.gradle b/build.gradle index 40088343c9..02d7546f73 100644 --- a/build.gradle +++ b/build.gradle @@ -34,6 +34,7 @@ allprojects { maven { url 'https://repo.spring.io/libs-snapshot' } } maven { url 'https://repo.spring.io/libs-milestone' } + maven { url 'https://repo.spring.io/libs-staging-local' } mavenCentral() } @@ -114,7 +115,7 @@ subprojects { subproject -> mockitoVersion = '1.10.19' mysqlVersion = '5.1.34' nettyVersion = '4.0.27.Final' - openJpaVersion = '2.4.0' + openJpaVersion = '2.4.1' pahoMqttClientVersion = '1.0.2' postgresVersion = '9.1-901-1.jdbc4' reactorVersion = '2.5.0.BUILD-SNAPSHOT' diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml index f8c7e18ee1..7a665877e3 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml @@ -3,18 +3,14 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:si-amqp="http://www.springframework.org/schema/integration/amqp" xmlns:int="http://www.springframework.org/schema/integration" - xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/integration/amqp http://www.springframework.org/schema/integration/amqp/spring-integration-amqp.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd - http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> - - + connection-factory="rabbitConnectionFactory" message-converter="testConverter"/> diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java index d38a438f4b..67efd16887 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java @@ -1,11 +1,11 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2002-2016 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. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -25,10 +25,9 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; -import reactor.core.publisher.EmitterProcessor; +import reactor.core.publisher.DirectProcessor; import reactor.core.subscriber.BaseSubscriber; -import reactor.core.subscriber.SignalEmitter; -import reactor.core.util.PlatformDependent; +import reactor.core.subscriber.SubmissionEmitter; /** * @author Artem Bilan @@ -38,19 +37,19 @@ public class ReactiveChannel implements MessageChannel, Publisher> { private final Processor, Message> processor; - private final SignalEmitter> emitter; + private final SubmissionEmitter> emitter; public ReactiveChannel() { - this(EmitterProcessor.create(PlatformDependent.SMALL_BUFFER_SIZE, Integer.MAX_VALUE, false)); + this(DirectProcessor.create()); } public ReactiveChannel(Processor, Message> processor) { Assert.notNull(processor, "'processor' must not be null"); this.processor = processor; - this.emitter = SignalEmitter.create(processor); + this.emitter = SubmissionEmitter.create(processor); } - Subscriber> asSubscriber() { + public Subscriber> asSubscriber() { return new BaseSubscriber>() { @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 06bbbb20b0..583f02f21c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -295,7 +295,6 @@ public class ConsumerEndpointFactoryBean subscriber = (Subscriber>) this.handler; } else { - //TODO errorConsumer, completeConsumer subscriber = Subscribers.consumer(this.handler::handleMessage); } this.endpoint = new ReactiveEndpoint(channel, subscriber); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index 4e5c7d9900..423791eeef 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -307,7 +307,8 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations) { + protected AbstractEndpoint doCreateEndpoint(MessageHandler handler, MessageChannel inputChannel, + List annotations) { AbstractEndpoint endpoint; if (inputChannel instanceof PollableChannel) { PollingConsumer pollingConsumer = new PollingConsumer((PollableChannel) inputChannel, handler); @@ -422,7 +423,7 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations, AbstractReplyProducingMessageHandler handler) { + protected void setOutputChannelIfPresent(List annotations, AbstractReplyProducingMessageHandler handler) { String outputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, "outputChannel", String.class); if (StringUtils.hasText(outputChannelName)) { handler.setOutputChannelName(outputChannelName); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java index ad2e863263..b57ba75620 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java @@ -18,12 +18,21 @@ package org.springframework.integration.endpoint; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import org.springframework.integration.channel.MessagePublishingErrorHandler; +import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; +import org.springframework.util.ErrorHandler; -import reactor.core.subscriber.SignalEmitter; +import reactor.core.publisher.DirectProcessor; +import reactor.core.subscriber.SubscriberBarrier; +import reactor.core.util.Exceptions; /** @@ -32,35 +41,142 @@ import reactor.core.subscriber.SignalEmitter; */ public class ReactiveEndpoint extends AbstractEndpoint { - private final Publisher> inputChannel; + private final Publisher> publisher; - private final Subscriber> subscriber; + private final SubscriberBarrier, Message> subscriber; + + private ErrorHandler errorHandler; - private SignalEmitter> emitter; @SuppressWarnings("unchecked") public ReactiveEndpoint(MessageChannel inputChannel, Subscriber> subscriber) { Assert.notNull(inputChannel); Assert.notNull(subscriber); if (inputChannel instanceof Publisher) { - this.inputChannel = (Publisher>) inputChannel; + this.publisher = (Publisher>) inputChannel; } else { - //TODO: Wrap all other channels to the Publisher - this.inputChannel = null; + this.publisher = adaptToPublisher(inputChannel); + } + this.subscriber = new SubscriberBarrier, Message>(subscriber) { + + @Override + protected void doNext(Message message) { + try { + super.doNext(message); + } + catch (Exception e) { + Exceptions.throwIfFatal(e); + ReactiveEndpoint.this.errorHandler.handleError(e); + } + } + + }; + } + + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + if (this.errorHandler == null) { + Assert.notNull(getBeanFactory(), "BeanFactory is required"); + this.errorHandler = new MessagePublishingErrorHandler( + new BeanFactoryChannelResolver(getBeanFactory())); } - this.subscriber = subscriber; } @Override protected void doStart() { - this.emitter = SignalEmitter.create(this.subscriber); - this.inputChannel.subscribe(this.emitter); + this.publisher.subscribe(this.subscriber); } @Override protected void doStop() { - this.emitter.finish(); + this.subscriber.cancel(); + } + + private Publisher> adaptToPublisher(MessageChannel inputChannel) { + if (inputChannel instanceof SubscribableChannel) { + return adaptSubscribableChannelToPublisher((SubscribableChannel) inputChannel); + } + else if (inputChannel instanceof PollableChannel) { + return adaptPollableChannelToPublisher((PollableChannel) inputChannel); + } + else { + throw new IllegalArgumentException("The 'inputChannel' must be an instance of SubscribableChannel or " + + "PollableChannel, not: " + inputChannel); + } + } + + private Publisher> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) { + return new SubscribableChannelPublisherAdapter(inputChannel); + } + + private Publisher> adaptPollableChannelToPublisher(PollableChannel inputChannel) { + return null; + } + + + private static class SubscribableChannelPublisherAdapter + implements Publisher>, Subscriber>, Subscription { + + private final DirectProcessor> delegate = DirectProcessor.create(); + + private final MessageHandler subscriberAdapter = this.delegate.connectEmitter()::accept; + + private final SubscribableChannel channel; + + private Subscriber> actualSubscriber; + + private Subscription actualSubscription; + + + private SubscribableChannelPublisherAdapter(SubscribableChannel channel) { + this.channel = channel; + } + + @Override + public void subscribe(Subscriber> subscriber) { + this.actualSubscriber = subscriber; + this.delegate.subscribe(this); + this.channel.subscribe(this.subscriberAdapter); + } + + @Override + public void onSubscribe(Subscription subscription) { + this.actualSubscription = subscription; + this.actualSubscriber.onSubscribe(this); + } + + @Override + public void onNext(Message message) { + this.actualSubscriber.onNext(message); + } + + @Override + public void onError(Throwable t) { + this.actualSubscriber.onError(t); + } + + @Override + public void onComplete() { + this.actualSubscriber.onComplete(); + } + + @Override + public void request(long n) { + this.actualSubscription.request(n); + } + + @Override + public void cancel() { + this.channel.unsubscribe(this.subscriberAdapter); + this.actualSubscription.cancel(); + } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 606b33940a..1de9b8e0e4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -29,7 +29,6 @@ import java.util.concurrent.Future; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; -import org.reactivestreams.Publisher; import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.AopUtils; @@ -46,7 +45,6 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.support.TaskExecutorAdapter; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; -import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.annotation.Gateway; import org.springframework.integration.annotation.GatewayHeader; import org.springframework.integration.context.IntegrationProperties; @@ -85,11 +83,6 @@ import reactor.core.publisher.Mono; public class GatewayProxyFactoryBean extends AbstractEndpoint implements TrackableComponent, FactoryBean, MethodInterceptor, BeanClassLoaderAware { - private static final SpelExpressionParser PARSER = new SpelExpressionParser(); - - private static final boolean reactorPresent = ClassUtils.isPresent("reactor.rx.Promise", - GatewayProxyFactoryBean.class.getClassLoader()); - private volatile Class serviceInterface; private volatile MessageChannel defaultRequestChannel; @@ -400,7 +393,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } } } - if (reactorPresent && Mono.class.isAssignableFrom(returnType)) { + if (Mono.class.isAssignableFrom(returnType)) { return Mono.fromCallable(new AsyncInvocationTask(invocation)); } return this.doInvoke(invocation, true); @@ -530,7 +523,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } headerExpressions.put(name, hasValue ? new LiteralExpression(value) - : PARSER.parseExpression(expression)); + : EXPRESSION_PARSER.parseExpression(expression)); } } @@ -639,7 +632,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint if (Future.class.isAssignableFrom(expectedReturnType)) { return (T) source; } - if (reactorPresent && Publisher.class.isAssignableFrom(expectedReturnType)) { + if (Mono.class.isAssignableFrom(expectedReturnType)) { return (T) source; } if (this.getConversionService() != null) { @@ -653,7 +646,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private static boolean hasReturnParameterizedWithMessage(Method method, boolean runningOnCallerThread) { if (!runningOnCallerThread && (Future.class.isAssignableFrom(method.getReturnType()) - || (reactorPresent && Publisher.class.isAssignableFrom(method.getReturnType())))) { + || Mono.class.isAssignableFrom(method.getReturnType()))) { Type returnType = method.getGenericReturnType(); if (returnType instanceof ParameterizedType) { Type[] typeArgs = ((ParameterizedType) returnType).getActualTypeArguments(); 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 f428f6a88a..5fdc5c3cae 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,6 +16,9 @@ package org.springframework.integration.handler; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; + import org.springframework.core.Ordered; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.context.Orderable; @@ -34,6 +37,8 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; +import reactor.core.util.Exceptions; + /** * Base class for MessageHandler implementations that provides basic validation * and error handling capabilities. Asserts that the incoming Message is not @@ -46,7 +51,8 @@ import org.springframework.util.Assert; */ @IntegrationManagedResource public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler, - MessageHandlerMetrics, ConfigurableMetricsAware, TrackableComponent, Orderable { + MessageHandlerMetrics, ConfigurableMetricsAware, TrackableComponent, Orderable, + Subscriber> { private volatile boolean shouldTrack = false; @@ -118,7 +124,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im boolean countsEnabled = this.countsEnabled; AbstractMessageHandlerMetrics handlerMetrics = this.handlerMetrics; try { - if (message != null && this.shouldTrack) { + if (this.shouldTrack) { message = MessageHistory.write(message, this, this.getMessageBuilderFactory()); } if (countsEnabled) { @@ -140,6 +146,31 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im } } + @Override + public void onSubscribe(Subscription subscription) { + Assert.notNull(subscription, "'subscription' must not be null"); + subscription.request(Long.MAX_VALUE); + } + + @Override + public void onNext(Message message) { + handleMessage(message); + } + + @Override + public void onError(Throwable throwable) { + Exceptions.throwIfFatal(throwable); + if (throwable instanceof MessagingException) { + throw (MessagingException) throwable; + } + throw new MessagingException("Error occurred in message handler [" + this + "]", throwable); + } + + @Override + public void onComplete() { + + } + protected abstract void handleMessageInternal(Message message) throws Exception; @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java index 39f01fb587..63c6e92581 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -16,7 +16,9 @@ package org.springframework.integration.channel.reactive; +import static org.hamcrest.Matchers.isOneOf; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; import org.junit.Test; import org.junit.runner.RunWith; @@ -56,10 +58,11 @@ public class ReactiveChannelTests { this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build()); } - for (int i = 0; i < 10; i++) { + for (int i = 0; i < 9; i++) { Message receive = replyChannel.receive(10000); assertNotNull(receive); - System.out.println("Receive: " + receive.getPayload()); + assertThat(receive.getPayload(), isOneOf("0", "1", "2", "3", "4", "6", "7", "8", "9")); + System .out.println("Receive: " + receive.getPayload()); } } @@ -74,7 +77,10 @@ public class ReactiveChannelTests { @ServiceActivator(inputChannel = "reactiveChannel") public String handle(int payload) { - System.out.println("CurrentThread: " + Thread.currentThread() + " for payload: " + payload); + if (payload == 5) { + throw new IllegalStateException("intentional"); + } + System .out.println("CurrentThread: " + Thread.currentThread() + " for payload: " + payload); return "" + payload; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveEndpointTests.java new file mode 100644 index 0000000000..58f69c0bda --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveEndpointTests.java @@ -0,0 +1,141 @@ +/* + * Copyright 2016 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.channel.reactive; + +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; + +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.hamcrest.Matchers; +import org.junit.Test; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.ReactiveChannel; +import org.springframework.integration.endpoint.ReactiveEndpoint; +import org.springframework.integration.handler.MethodInvokingMessageHandler; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.support.GenericMessage; + +import reactor.core.publisher.EmitterProcessor; +import reactor.core.test.TestSubscriber; + +/** + * @author Artem Bilan + * @since 5.0 + */ +public class ReactiveEndpointTests { + + @Test + public void testReactiveEndpointReactiveChannel() throws InterruptedException { + ReactiveChannel testChannel = + new ReactiveChannel(EmitterProcessor.create(false)); + + List> result = new LinkedList<>(); + CountDownLatch stopLatch = new CountDownLatch(2); + + MessageHandler messageHandler = m -> { + result.add(m); + stopLatch.countDown(); + }; + + MethodInvokingMessageHandler testSubscriber = new MethodInvokingMessageHandler(messageHandler, (String) null); + + ReactiveEndpoint reactiveEndpoint = new ReactiveEndpoint(testChannel, testSubscriber); + reactiveEndpoint.setBeanFactory(mock(BeanFactory.class)); + reactiveEndpoint.afterPropertiesSet(); + reactiveEndpoint.start(); + + Message testMessage = new GenericMessage<>("test"); + testChannel.send(testMessage); + + reactiveEndpoint.stop(); + + testChannel.send(testMessage); + + reactiveEndpoint.start(); + + Message testMessage2 = new GenericMessage<>("test2"); + testChannel.send(testMessage2); + + assertTrue(stopLatch.await(10, TimeUnit.SECONDS)); + assertThat(result, Matchers.>contains(testMessage, testMessage2)); + } + + + @Test + public void testReactiveEndpointDirectChannel() { + DirectChannel testChannel = new DirectChannel(); + + TestSubscriber> testSubscriber = new TestSubscriber<>(); + + ReactiveEndpoint reactiveEndpoint = new ReactiveEndpoint(testChannel, testSubscriber); + reactiveEndpoint.setBeanFactory(mock(BeanFactory.class)); + reactiveEndpoint.afterPropertiesSet(); + reactiveEndpoint.start(); + + Message testMessage = new GenericMessage<>("test"); + testChannel.send(testMessage); + + testSubscriber.assertSubscribed(); + testSubscriber.assertNoError(); + testSubscriber.assertNotComplete(); + + testSubscriber.assertValues(testMessage); + + reactiveEndpoint.stop(); + + try { + testChannel.send(testMessage); + fail("MessageDeliveryException"); + } + catch (Exception e) { + assertThat(e, instanceOf(MessageDeliveryException.class)); + } + + new DirectFieldAccessor(testSubscriber).setPropertyValue("s", null); + TestUtils.getPropertyValue(testSubscriber, "values", List.class).clear(); + + reactiveEndpoint.start(); + + testSubscriber.request(1); + + testMessage = new GenericMessage<>("test2"); + + testChannel.send(testMessage); + + testSubscriber.assertValues(testMessage); + + testChannel.send(testMessage); + + testSubscriber.assertError(IllegalStateException.class); + testSubscriber.assertErrorMessage("Can't deliver value due to lack of requests"); + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index cc25bc58a1..8e6d96cef4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -57,10 +57,6 @@ import reactor.core.publisher.Mono; */ public class AsyncGatewayTests { - - // TODO: changed from 0 because of recurrent failure: is this right? - private final long safety = 100; - @Test public void futureWithMessageReturned() throws Exception { QueueChannel requestChannel = new QueueChannel(); @@ -73,10 +69,7 @@ public class AsyncGatewayTests { proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); Future> f = service.returnMessage("foo"); - long start = System.currentTimeMillis(); Object result = f.get(10000, TimeUnit.MILLISECONDS); - long elapsed = System.currentTimeMillis() - start; - assertTrue(elapsed >= 200); assertNotNull(result); assertEquals("foobar", ((Message) result).getPayload()); } @@ -210,10 +203,7 @@ public class AsyncGatewayTests { proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); Future f = service.returnString("foo"); - long start = System.currentTimeMillis(); Object result = f.get(10000, TimeUnit.MILLISECONDS); - long elapsed = System.currentTimeMillis() - start; - assertTrue(elapsed >= 200 - safety); assertNotNull(result); assertEquals("foobar", result); } @@ -230,10 +220,7 @@ public class AsyncGatewayTests { proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); Future f = service.returnSomething("foo"); - long start = System.currentTimeMillis(); Object result = f.get(10000, TimeUnit.MILLISECONDS); - long elapsed = System.currentTimeMillis() - start; - assertTrue(elapsed >= 200 - safety); assertTrue(result instanceof String); assertEquals("foobar", result); } @@ -344,7 +331,7 @@ public class AsyncGatewayTests { } - interface TestEchoService { + private interface TestEchoService { Future returnString(String s);