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

Upgrade to SF-5.0 and Reactor-3.0

Browse Source
This commit is contained in:
Artem Bilan
2016-07-28 15:14:10 -04:00
parent 8a577de83f
commit 11151efc8c
21 changed files with 354 additions and 370 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java
View File

@@ -19,15 +19,13 @@ package org.springframework.integration.channel;
import org.reactivestreams.Processor;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
import reactor.core.publisher.BlockingSink;
import reactor.core.publisher.DirectProcessor;
import reactor.core.subscriber.BaseSubscriber;
import reactor.core.subscriber.SubmissionEmitter;
/**
* @author Artem Bilan
@@ -37,7 +35,7 @@ public class ReactiveChannel implements MessageChannel, Publisher<Message<?>> {
private final Processor<Message<?>, Message<?>> processor;
private final SubmissionEmitter<Message<?>> emitter;
private final BlockingSink<Message<?>> sink;
public ReactiveChannel() {
this(DirectProcessor.create());
@@ -46,24 +44,7 @@ public class ReactiveChannel implements MessageChannel, Publisher<Message<?>> {
public ReactiveChannel(Processor<Message<?>, Message<?>> processor) {
Assert.notNull(processor, "'processor' must not be null");
this.processor = processor;
this.emitter = SubmissionEmitter.create(processor);
}
public Subscriber<Message<?>> asSubscriber() {
return new BaseSubscriber<Message<?>>() {
@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) {
send(message);
}
};
this.sink = BlockingSink.create(this.processor);
}
@Override
@@ -73,7 +54,7 @@ public class ReactiveChannel implements MessageChannel, Publisher<Message<?>> {
@Override
public boolean send(Message<?> message, long timeout) {
return this.emitter.submit(message, timeout) > -1;
return this.sink.submit(message, timeout) > -1;
}
@Override

10
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -40,7 +40,7 @@ import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.ReactiveEndpoint;
import org.springframework.integration.endpoint.ReactiveConsumer;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.advice.HandleMessageAdvice;
import org.springframework.integration.scheduling.PollerMetadata;
@@ -55,8 +55,6 @@ import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import reactor.core.subscriber.Subscribers;
/**
* @author Mark Fisher
@@ -290,14 +288,12 @@ public class ConsumerEndpointFactoryBean
this.endpoint = pollingConsumer;
}
else {
Subscriber<Message<?>> subscriber;
if (this.handler instanceof Subscriber) {
subscriber = (Subscriber<Message<?>>) this.handler;
this.endpoint = new ReactiveConsumer(channel, (Subscriber<Message<?>>) this.handler);
}
else {
subscriber = Subscribers.consumer(this.handler::handleMessage);
this.endpoint = new ReactiveConsumer(channel, this.handler::handleMessage);
}
this.endpoint = new ReactiveEndpoint(channel, subscriber);
}
this.endpoint.setBeanName(this.beanName);
this.endpoint.setBeanFactory(this.beanFactory);

10
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -55,7 +55,7 @@ import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.AbstractPollingEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.ReactiveEndpoint;
import org.springframework.integration.endpoint.ReactiveConsumer;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.handler.AbstractMessageProducingHandler;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -80,7 +80,7 @@ import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import reactor.core.subscriber.Subscribers;
import reactor.core.publisher.DirectProcessor;
/**
* Base class for Method-level annotation post-processors.
@@ -326,9 +326,11 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
}
else {
//TODO errorConsumer, completeConsumer
subscriber = Subscribers.consumer(handler::handleMessage);
DirectProcessor<Message<?>> directProcessor = DirectProcessor.create();
directProcessor.doOnNext(handler::handleMessage);
subscriber = directProcessor;
}
endpoint = new ReactiveEndpoint(inputChannel, subscriber);
endpoint = new ReactiveConsumer(inputChannel, subscriber);
}
else {
endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);

70
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java → spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.endpoint;
import java.util.function.Consumer;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
@@ -31,47 +33,58 @@ import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
import reactor.core.publisher.DirectProcessor;
import reactor.core.subscriber.SubscriberBarrier;
import reactor.core.util.Exceptions;
import reactor.core.publisher.Flux;
/**
* @author Artem Bilan
* @since 5.0
*/
public class ReactiveEndpoint extends AbstractEndpoint {
public class ReactiveConsumer extends AbstractEndpoint {
private final Publisher<Message<?>> publisher;
private final Subscriber<Message<?>> subscriber;
private final SubscriberBarrier<Message<?>, Message<?>> subscriber;
private final Consumer<Message<?>> consumer;
private volatile Flux<Message<?>> publisher;
private volatile Subscription subscription;
private ErrorHandler errorHandler;
@SuppressWarnings("unchecked")
public ReactiveEndpoint(MessageChannel inputChannel, Subscriber<Message<?>> subscriber) {
Assert.notNull(inputChannel);
public ReactiveConsumer(MessageChannel inputChannel, Subscriber<Message<?>> subscriber) {
this(inputChannel, subscriber, null);
Assert.notNull(subscriber);
}
public ReactiveConsumer(MessageChannel inputChannel, Consumer<Message<?>> consumer) {
this(inputChannel, null, consumer);
Assert.notNull(consumer);
}
@SuppressWarnings("unchecked")
private ReactiveConsumer(MessageChannel inputChannel, Subscriber<Message<?>> subscriber,
Consumer<Message<?>> consumer) {
Assert.notNull(inputChannel);
Publisher<Message<?>> publisher;
if (inputChannel instanceof Publisher) {
this.publisher = (Publisher<Message<?>>) inputChannel;
publisher = (Publisher<Message<?>>) inputChannel;
}
else {
this.publisher = adaptToPublisher(inputChannel);
publisher = adaptToPublisher(inputChannel);
}
this.subscriber = new SubscriberBarrier<Message<?>, Message<?>>(subscriber) {
@Override
protected void doNext(Message<?> message) {
try {
super.doNext(message);
}
catch (Exception e) {
Exceptions.throwIfFatal(e);
ReactiveEndpoint.this.errorHandler.handleError(e);
}
}
this.publisher = Flux.from(publisher)
.log()
.retry()
.doOnError(t -> this.errorHandler.handleError(t)) // NPE if method reference
.doOnSubscribe(s -> this.subscription = s);
};
this.subscriber = subscriber;
this.consumer = consumer;
}
public void setErrorHandler(ErrorHandler errorHandler) {
@@ -90,12 +103,19 @@ public class ReactiveEndpoint extends AbstractEndpoint {
@Override
protected void doStart() {
this.publisher.subscribe(this.subscriber);
if (this.subscriber != null) {
this.publisher.subscribe(this.subscriber);
}
else {
this.publisher.subscribe(this.consumer);
}
}
@Override
protected void doStop() {
this.subscriber.cancel();
if (this.subscription != null) {
this.subscription.cancel();
}
}
private Publisher<Message<?>> adaptToPublisher(MessageChannel inputChannel) {
@@ -125,7 +145,7 @@ public class ReactiveEndpoint extends AbstractEndpoint {
private final DirectProcessor<Message<?>> delegate = DirectProcessor.create();
private final MessageHandler subscriberAdapter = this.delegate.connectEmitter()::accept;
private final MessageHandler subscriberAdapter = this.delegate.connectSink()::accept;
private final SubscribableChannel channel;

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

@@ -37,7 +37,7 @@ import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
import reactor.core.util.Exceptions;
import reactor.core.Exceptions;
/**
* Base class for MessageHandler implementations that provides basic validation
@@ -168,7 +168,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
@Override
public void onComplete() {
System.out.println("onComplete()");
}
protected abstract void handleMessageInternal(Message<?> message) throws Exception;

2
spring-integration-core/src/main/java/org/springframework/integration/transformer/MapToObjectTransformer.java
View File

@@ -78,7 +78,7 @@ public class MapToObjectTransformer extends AbstractPayloadTransformer<Map<?, ?>
@Override
protected Object transformPayload(Map<?, ?> payload) throws Exception {
Object target = (this.targetClass != null)
? BeanUtils.instantiate(this.targetClass)
? BeanUtils.instantiateClass(this.targetClass)
: this.getBeanFactory().getBean(this.targetBeanName);
DataBinder binder = new DataBinder(target);

79
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveEndpointTests.java → spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.channel.reactive;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
@@ -32,9 +33,11 @@ import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.ReactiveChannel;
import org.springframework.integration.endpoint.ReactiveEndpoint;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.endpoint.ReactiveConsumer;
import org.springframework.integration.handler.MethodInvokingMessageHandler;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
@@ -43,18 +46,17 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.test.TestSubscriber;
import reactor.test.TestSubscriber;
/**
* @author Artem Bilan
* @since 5.0
*/
public class ReactiveEndpointTests {
public class ReactiveConsumerTests {
@Test
public void testReactiveEndpointReactiveChannel() throws InterruptedException {
ReactiveChannel testChannel =
new ReactiveChannel(EmitterProcessor.create(false));
public void testReactiveConsumerReactiveChannel() throws InterruptedException {
ReactiveChannel testChannel = new ReactiveChannel(EmitterProcessor.create(false));
List<Message<?>> result = new LinkedList<>();
CountDownLatch stopLatch = new CountDownLatch(2);
@@ -66,19 +68,19 @@ public class ReactiveEndpointTests {
MethodInvokingMessageHandler testSubscriber = new MethodInvokingMessageHandler(messageHandler, (String) null);
ReactiveEndpoint reactiveEndpoint = new ReactiveEndpoint(testChannel, testSubscriber);
reactiveEndpoint.setBeanFactory(mock(BeanFactory.class));
reactiveEndpoint.afterPropertiesSet();
reactiveEndpoint.start();
ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber);
reactiveConsumer.setBeanFactory(mock(BeanFactory.class));
reactiveConsumer.afterPropertiesSet();
reactiveConsumer.start();
Message<?> testMessage = new GenericMessage<>("test");
testChannel.send(testMessage);
reactiveEndpoint.stop();
reactiveConsumer.stop();
testChannel.send(testMessage);
reactiveEndpoint.start();
reactiveConsumer.start();
Message<?> testMessage2 = new GenericMessage<>("test2");
testChannel.send(testMessage2);
@@ -89,15 +91,15 @@ public class ReactiveEndpointTests {
@Test
public void testReactiveEndpointDirectChannel() {
public void testReactiveConsumerDirectChannel() {
DirectChannel testChannel = new DirectChannel();
TestSubscriber<Message<?>> testSubscriber = new TestSubscriber<>();
TestSubscriber<Message<?>> testSubscriber = TestSubscriber.create();
ReactiveEndpoint reactiveEndpoint = new ReactiveEndpoint(testChannel, testSubscriber);
reactiveEndpoint.setBeanFactory(mock(BeanFactory.class));
reactiveEndpoint.afterPropertiesSet();
reactiveEndpoint.start();
ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber);
reactiveConsumer.setBeanFactory(mock(BeanFactory.class));
reactiveConsumer.afterPropertiesSet();
reactiveConsumer.start();
Message<?> testMessage = new GenericMessage<>("test");
testChannel.send(testMessage);
@@ -108,7 +110,7 @@ public class ReactiveEndpointTests {
testSubscriber.assertValues(testMessage);
reactiveEndpoint.stop();
reactiveConsumer.stop();
try {
testChannel.send(testMessage);
@@ -121,7 +123,7 @@ public class ReactiveEndpointTests {
new DirectFieldAccessor(testSubscriber).setPropertyValue("s", null);
TestUtils.getPropertyValue(testSubscriber, "values", List.class).clear();
reactiveEndpoint.start();
reactiveConsumer.start();
testSubscriber.request(1);
@@ -130,12 +132,45 @@ public class ReactiveEndpointTests {
testChannel.send(testMessage);
testSubscriber.assertValues(testMessage);
}
@Test
public void testReactiveConsumerViaConsumerEndpointFactoryBean() throws Exception {
ReactiveChannel testChannel = new ReactiveChannel();
List<Message<?>> result = new LinkedList<>();
CountDownLatch stopLatch = new CountDownLatch(3);
MessageHandler messageHandler = m -> {
result.add(m);
stopLatch.countDown();
};
ConsumerEndpointFactoryBean endpointFactoryBean = new ConsumerEndpointFactoryBean();
endpointFactoryBean.setBeanFactory(mock(ConfigurableBeanFactory.class));
endpointFactoryBean.setInputChannel(testChannel);
endpointFactoryBean.setHandler(messageHandler);
endpointFactoryBean.setBeanName("reactiveConsumer");
endpointFactoryBean.afterPropertiesSet();
endpointFactoryBean.start();
Message<?> testMessage = new GenericMessage<>("test");
testChannel.send(testMessage);
endpointFactoryBean.stop();
testChannel.send(testMessage);
testSubscriber.assertError(IllegalStateException.class);
testSubscriber.assertErrorMessage("Can't deliver value due to lack of requests");
endpointFactoryBean.start();
Message<?> testMessage2 = new GenericMessage<>("test2");
testChannel.send(testMessage2);
testChannel.send(testMessage2);
assertTrue(stopLatch.await(10, TimeUnit.SECONDS));
assertThat(result.size(), equalTo(3));
assertThat(result, Matchers.<Message<?>>contains(testMessage, testMessage2, testMessage2));
}
}

2
spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java
View File

@@ -175,7 +175,7 @@ public class GatewayParserTests {
this.startResponder(requestChannel, replyChannel);
TestService service = context.getBean("promise", TestService.class);
Mono<Message<?>> result = service.promise("foo");
Message<?> reply = result.get(Duration.ofSeconds(1));
Message<?> reply = result.block(Duration.ofSeconds(1));
assertEquals("foo", reply.getPayload());
assertNotNull(TestUtils.getPropertyValue(context.getBean("&promise"), "asyncExecutor"));
}

5
spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
View File

@@ -128,6 +128,7 @@ import org.springframework.test.context.support.AnnotationConfigContextLoader;
import org.springframework.util.MultiValueMap;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* @author Artem Bilan
@@ -620,7 +621,7 @@ public class EnableIntegrationTests {
Flux.just("1", "2", "3", "4", "5")
.map(Integer::parseInt)
.flatMap(this.testGateway::multiply)
.toList()
.collectList()
.subscribe(integers -> {
ref.set(integers);
consumeLatch.countDown();
@@ -1334,7 +1335,7 @@ public class EnableIntegrationTests {
void sendAsync(String payload);
@Gateway(requestChannel = "promiseChannel")
org.reactivestreams.Publisher<Integer> multiply(Integer value);
Mono<Integer> multiply(Integer value);
}

2
spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java
View File

@@ -391,7 +391,7 @@ public class AsyncMessagingTemplateTests {
@Test(expected = TimeoutException.class)
public void timeoutException() throws Exception {
DirectChannel channel = new DirectChannel();
channel.subscribe(new EchoHandler(200));
channel.subscribe(new EchoHandler(10000));
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
template.setDefaultDestination(channel);
Future<Message<?>> result = template.asyncSendAndReceive(MessageBuilder.withPayload("test").build());

6
spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java
View File

@@ -238,7 +238,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Mono<Message<?>> promise = service.returnMessagePromise("foo");
Object result = promise.get(Duration.ofSeconds(10));
Object result = promise.block(Duration.ofSeconds(10));
assertEquals("foobar", ((Message<?>) result).getPayload());
}
@@ -254,7 +254,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Mono<String> promise = service.returnStringPromise("foo");
Object result = promise.get(Duration.ofSeconds(10));
Object result = promise.block(Duration.ofSeconds(10));
assertEquals("foobar", result);
}
@@ -270,7 +270,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Mono<?> promise = service.returnSomethingPromise("foo");
Object result = promise.get(Duration.ofSeconds(10));
Object result = promise.block(Duration.ofSeconds(10));
assertNotNull(result);
assertEquals("foobar", result);
}