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

Upgrade to Rector-2.5

Browse Source
This commit is contained in:
Artem Bilan
2016-01-21 17:01:43 -05:00
parent 3eae0239bf
commit e6af887972
12 changed files with 55 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
build.gradle
View File

@@ -293,10 +293,7 @@ project('spring-integration-core') {
exclude group: 'org.springframework', module: 'spring-core'
}
// compile ("org.springframework.cloud:spring-cloud-cluster-core:$springCloudClusterVersion", optional)
compile ("io.projectreactor:reactor-stream:$reactorVersion") {
optional it
exclude group: 'org.slf4j', module: 'slf4j-api'
}
compile "io.projectreactor:reactor-core:$reactorVersion"
compile("com.fasterxml.jackson.core:jackson-databind:$jackson2Version", optional)
compile("com.jayway.jsonpath:json-path:$jsonpathVersion") {
optional it
@@ -309,8 +306,7 @@ project('spring-integration-core') {
// testCompile ("net.openhft:chronicle:$chronicleVersion")
// testCompile ("io.projectreactor:reactor-chronicle:$reactorVersion")
testCompile "org.testng:testng:6.8.21"
testCompile "org.reactivestreams:reactive-streams-tck:1.0.0"
testCompile "io.projectreactor:reactor-stream:$reactorVersion"
}
}

4
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml
View File

@@ -12,7 +12,9 @@
<int:channel id="requests"/>
<si-amqp:inbound-gateway id="gateway" request-channel="requests" queue-names="test" reply-timeout="1234"
connection-factory="rabbitConnectionFactory" message-converter="testConverter"/>
connection-factory="rabbitConnectionFactory" message-converter="testConverter">
<reactive/>
</si-amqp:inbound-gateway>
<bean id="rabbitConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>

12
spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveMessageChannel.java → spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java
View File

@@ -23,24 +23,24 @@ import org.reactivestreams.Subscriber;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import reactor.core.subscription.ReactiveSession;
import reactor.rx.broadcast.Broadcaster;
import reactor.core.publisher.ProcessorGroup;
import reactor.core.subscriber.ReactiveSession;
/**
* @author Artem Bilan
* @since 5.0
*/
public class ReactiveMessageChannel implements MessageChannel, Publisher<Message<?>> {
public class ReactiveChannel implements MessageChannel, Publisher<Message<?>> {
private final Processor<Message<?>, Message<?>> processor;
private final ReactiveSession<Message<?>> reactiveSession;
public ReactiveMessageChannel() {
this(Broadcaster.passthrough());
public ReactiveChannel() {
this(ProcessorGroup.<Message<?>>sync().get());
}
public ReactiveMessageChannel(Processor<Message<?>, Message<?>> processor) {
public ReactiveChannel(Processor<Message<?>, Message<?>> processor) {
this.processor = processor;
this.reactiveSession = ReactiveSession.create(processor);
}

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

@@ -21,7 +21,6 @@ import java.util.List;
import org.aopalliance.aop.Advice;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.springframework.aop.framework.Advised;
@@ -56,7 +55,8 @@ import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import reactor.core.subscriber.SubscriberFactory;
import reactor.core.subscriber.Subscribers;
/**
* @author Mark Fisher
@@ -289,20 +289,16 @@ public class ConsumerEndpointFactoryBean
pollingConsumer.setBeanFactory(this.beanFactory);
this.endpoint = pollingConsumer;
}
else if (channel instanceof Publisher) {
Publisher<Message<?>> publisher = (Publisher<Message<?>>) channel;
else {
Subscriber<Message<?>> subscriber;
if (this.handler instanceof Subscriber) {
subscriber = (Subscriber<Message<?>>) this.handler;
}
else {
//TODO errorConsumer, completeConsumer
subscriber = SubscriberFactory.consumer(this.handler::handleMessage);
subscriber = Subscribers.consumer(this.handler::handleMessage);
}
this.endpoint = new ReactiveEndpoint(publisher, subscriber);
}
else {
throw new IllegalArgumentException("unsupported channel type: [" + channel.getClass() + "]");
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

@@ -80,7 +80,7 @@ import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import reactor.core.subscriber.SubscriberFactory;
import reactor.core.subscriber.Subscribers;
/**
* Base class for Method-level annotation post-processors.
@@ -306,7 +306,8 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
return endpoint;
}
protected AbstractEndpoint doCreateEndpoint(MessageHandler handler, MessageChannel inputChannel, List<Annotation> annotations) {
@SuppressWarnings("unchecked")
protected AbstractEndpoint doCreateEndpoint(MessageHandler handler, MessageChannel inputChannel,List<Annotation> annotations) {
AbstractEndpoint endpoint;
if (inputChannel instanceof PollableChannel) {
PollingConsumer pollingConsumer = new PollingConsumer((PollableChannel) inputChannel, handler);
@@ -318,16 +319,15 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
Assert.state(ObjectUtils.isEmpty(pollers), "A '@Poller' should not be specified for Annotation-based " +
"endpoint, since '" + inputChannel + "' is a SubscribableChannel (not pollable).");
if (inputChannel instanceof Publisher) {
Publisher<Message<?>> publisher = (Publisher<Message<?>>) inputChannel;
Subscriber<Message<?>> subscriber;
if (handler instanceof Subscriber) {
subscriber = (Subscriber<Message<?>>) handler;
}
else {
//TODO errorConsumer, completeConsumer
subscriber = SubscriberFactory.consumer(handler::handleMessage);
subscriber = Subscribers.consumer(handler::handleMessage);
}
endpoint = new ReactiveEndpoint(publisher, subscriber);
endpoint = new ReactiveEndpoint(inputChannel, subscriber);
}
else {
endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);

18
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java
View File

@@ -20,9 +20,11 @@ import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
import reactor.core.subscription.ReactiveSession;
import reactor.core.subscriber.ReactiveSession;
/**
* @author Artem Bilan
@@ -36,11 +38,17 @@ public class ReactiveEndpoint extends AbstractEndpoint {
private ReactiveSession<Message<?>> reactiveSession;
public ReactiveEndpoint(Publisher<Message<?>> inputChannel, Subscriber<Message<?>> subscriber) {
Assert.isInstanceOf(Publisher.class, inputChannel,
"The 'inputChannel', must implement org.reactivestreams.Publisher.");
@SuppressWarnings("unchecked")
public ReactiveEndpoint(MessageChannel inputChannel, Subscriber<Message<?>> subscriber) {
Assert.notNull(inputChannel);
Assert.notNull(subscriber);
this.inputChannel = inputChannel;
if (inputChannel instanceof Publisher) {
this.inputChannel = (Publisher<Message<?>>) inputChannel;
}
else {
//TODO: Wrap all other channels to the Publisher<?>
this.inputChannel = null;
}
this.subscriber = subscriber;
}

5
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -66,7 +66,8 @@ import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
import reactor.rx.Promises;
import reactor.core.publisher.Mono;
/**
* Generates a proxy for the provided service interface to enable interaction
@@ -400,7 +401,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
}
}
if (reactorPresent && Publisher.class.isAssignableFrom(returnType)) {
return Promises.<Object>task(() -> new AsyncInvocationTask(invocation));
return Mono.fromCallable(new AsyncInvocationTask(invocation));
}
return this.doInvoke(invocation, true);
}

8
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveMessageChannelTests.java → spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java
View File

@@ -26,7 +26,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.ReactiveMessageChannel;
import org.springframework.integration.channel.ReactiveChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -35,7 +35,7 @@ import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import reactor.Processors;
import reactor.core.publisher.Processors;
/**
* @author Artem Bilan
@@ -44,7 +44,7 @@ import reactor.Processors;
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class ReactiveMessageChannelTests {
public class ReactiveChannelTests {
@Autowired
private MessageChannel reactiveChannel;
@@ -71,7 +71,7 @@ public class ReactiveMessageChannelTests {
@Bean
public MessageChannel reactiveChannel() {
return new ReactiveMessageChannel(Processors.queue());
return new ReactiveChannel(Processors.queue());
}
@ServiceActivator(inputChannel = "reactiveChannel")

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

@@ -65,7 +65,7 @@ import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import reactor.rx.Promises;
import reactor.rx.Promise;
/**
* @author Mark Fisher
@@ -175,7 +175,7 @@ public class GatewayParserTests {
this.startResponder(requestChannel, replyChannel);
TestService service = context.getBean("promise", TestService.class);
Publisher<Message<?>> result = service.promise("foo");
Message<?> reply = Promises.from(result).await(1, TimeUnit.SECONDS);
Message<?> reply = Promise.from(result).await(1, TimeUnit.SECONDS);
assertEquals("foo", reply.getPayload());
assertNotNull(TestUtils.getPropertyValue(context.getBean("&promise"), "asyncExecutor"));
}

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

@@ -127,7 +127,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.support.AnnotationConfigContextLoader;
import org.springframework.util.MultiValueMap;
import reactor.rx.Streams;
import reactor.rx.Stream;
/**
* @author Artem Bilan
@@ -617,11 +617,11 @@ public class EnableIntegrationTests {
final AtomicReference<List<Integer>> ref = new AtomicReference<List<Integer>>();
final CountDownLatch consumeLatch = new CountDownLatch(1);
Streams.just("1", "2", "3", "4", "5")
Stream.just("1", "2", "3", "4", "5")
.map(Integer::parseInt)
.flatMap(this.testGateway::multiply)
.toList()
.onSuccess(integers -> {
.doOnSuccess(integers -> {
ref.set(integers);
consumeLatch.countDown();
});

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

@@ -47,7 +47,7 @@ import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import reactor.fn.Consumer;
import reactor.rx.Promises;
import reactor.rx.Promise;
/**
* @author Mark Fisher
@@ -252,7 +252,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Publisher<Message<?>> promise = service.returnMessagePromise("foo");
Object result = Promises.from(promise).await(10, TimeUnit.SECONDS);
Object result = Promise.from(promise).await(10, TimeUnit.SECONDS);
assertEquals("foobar", ((Message<?>) result).getPayload());
}
@@ -268,7 +268,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Publisher<String> promise = service.returnStringPromise("foo");
Object result = Promises.from(promise).await(10, TimeUnit.SECONDS);
Object result = Promise.from(promise).await(10, TimeUnit.SECONDS);
assertEquals("foobar", result);
}
@@ -284,7 +284,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Publisher<?> promise = service.returnSomethingPromise("foo");
Object result = Promises.from(promise).await(10, TimeUnit.SECONDS);
Object result = Promise.from(promise).await(10, TimeUnit.SECONDS);
assertNotNull(result);
assertEquals("foobar", result);
}
@@ -305,7 +305,7 @@ public class AsyncGatewayTests {
final AtomicReference<String> result = new AtomicReference<String>();
final CountDownLatch latch = new CountDownLatch(1);
Promises.from(promise).onSuccess(new Consumer<String>() {
Promise.from(promise).doOnSuccess(new Consumer<String>() {
@Override
public void accept(String s) {
result.set(s);

8
spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2015 the original author or authors.
* Copyright 2014-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.
@@ -60,7 +60,7 @@ import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.support.GenericXmlContextLoader;
import reactor.rx.Streams;
import reactor.rx.Stream;
/**
* @author Artem Bilan
@@ -189,11 +189,11 @@ public class RoutingSlipTests {
public RoutingSlipRouteStrategy routeStrategy() {
return (requestMessage, reply) -> requestMessage.getPayload() instanceof String
? new FixedSubscriberChannel(m ->
Streams.just((String) m.getPayload())
Stream.just((String) m.getPayload())
.map(String::toUpperCase)
.consume(v -> messagingTemplate().convertAndSend(resultsChannel(), v)))
: new FixedSubscriberChannel(m ->
Streams.just((Integer) m.getPayload())
Stream.just((Integer) m.getPayload())
.map(v -> v * 2)
.consume(v -> messagingTemplate().convertAndSend(resultsChannel(), v)));
}