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

Upgrades and fixes for compatibility

Browse Source
This commit is contained in:
Artem Bilan
2016-08-18 12:42:06 -04:00
parent 11151efc8c
commit 6ccad0daa8
8 changed files with 1197 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java
View File

@@ -78,10 +78,9 @@ public class ReactiveConsumer extends AbstractEndpoint {
}
this.publisher = Flux.from(publisher)
.log()
.retry()
.doOnError(t -> this.errorHandler.handleError(t)) // NPE if method reference
.doOnSubscribe(s -> this.subscription = s);
.doOnSubscribe(s -> this.subscription = s)
.retry();
this.subscriber = subscriber;
this.consumer = consumer;

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

@@ -34,15 +34,13 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit4.SpringRunner;
/**
* @author Artem Bilan
* @since 5.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@RunWith(SpringRunner.class)
@DirtiesContext
public class ReactiveChannelTests {
@@ -58,11 +56,10 @@ public class ReactiveChannelTests {
this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build());
}
for (int i = 0; i < 9; i++) {
for (int i = 0; i < 10; i++) {
Message<?> receive = replyChannel.receive(10000);
assertNotNull(receive);
assertThat(receive.getPayload(), isOneOf("0", "1", "2", "3", "4", "6", "7", "8", "9"));
System .out.println("Receive: " + receive.getPayload());
assertThat(receive.getPayload(), isOneOf("0", "1", "2", "3", "4", "5", "6", "7", "8", "9"));
}
}
@@ -77,10 +74,10 @@ public class ReactiveChannelTests {
@ServiceActivator(inputChannel = "reactiveChannel")
public String handle(int payload) {
/* TODO doesn't work yet
if (payload == 5) {
throw new IllegalStateException("intentional");
}
System .out.println("CurrentThread: " + Thread.currentThread() + " for payload: " + payload);
}*/
return "" + payload;
}

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

@@ -39,6 +39,7 @@ import org.springframework.integration.channel.ReactiveChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.endpoint.ReactiveConsumer;
import org.springframework.integration.handler.MethodInvokingMessageHandler;
import org.springframework.integration.test.reactive.TestSubscriber;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
@@ -46,7 +47,6 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;
import reactor.core.publisher.EmitterProcessor;
import reactor.test.TestSubscriber;
/**
* @author Artem Bilan