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

Rename ReactiveChannel

Browse Source 
- as discussed last week

TODO:
- should we take the channel outside of the `AbstractMessageChannel` hierarchy?
  - avoid blocking interceptors
  - we would lose channel metrics though
- rename `ReactiveConsumer` ?

Polishing some missed renaming
This commit is contained in:
Gary Russell
2017-04-26 11:38:13 -04:00
committed by Artem Bilan
parent cd4964eb7f
commit ca231763a3
11 changed files with 69 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -36,9 +36,9 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.MessageChannelReactiveUtils;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.ReactiveChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -57,10 +57,10 @@ import reactor.core.publisher.Flux;
*/
@RunWith(SpringRunner.class)
@DirtiesContext
public class ReactiveChannelTests {
public class FluxMessageChannelTests {
@Autowired
private MessageChannel reactiveChannel;
private MessageChannel fluxMessageChannel;
@Autowired
private MessageChannel queueChannel;
@@ -69,11 +69,11 @@ public class ReactiveChannelTests {
private PollableChannel errorChannel;
@Test
public void testReactiveMessageChannel() throws InterruptedException {
public void testFluxMessageChannel() throws InterruptedException {
QueueChannel replyChannel = new QueueChannel();
for (int i = 0; i < 10; i++) {
this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build());
this.fluxMessageChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build());
}
for (int i = 0; i < 9; i++) {
@@ -116,11 +116,11 @@ public class ReactiveChannelTests {
}
@Bean
public MessageChannel reactiveChannel() {
return new ReactiveChannel();
public MessageChannel fluxMessageChannel() {
return new FluxMessageChannel();
}
@ServiceActivator(inputChannel = "reactiveChannel")
@ServiceActivator(inputChannel = "fluxMessageChannel")
public String handle(int payload) {
if (payload == 5) {
throw new IllegalStateException("intentional");

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

@@ -47,7 +47,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.ReactiveChannel;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.endpoint.ReactiveConsumer;
import org.springframework.integration.handler.MethodInvokingMessageHandler;
@@ -67,7 +67,7 @@ public class ReactiveConsumerTests {
@Test
public void testReactiveConsumerReactiveChannel() throws InterruptedException {
ReactiveChannel testChannel = new ReactiveChannel(EmitterProcessor.create(false));
FluxMessageChannel testChannel = new FluxMessageChannel(EmitterProcessor.create(false));
List<Message<?>> result = new LinkedList<>();
CountDownLatch stopLatch = new CountDownLatch(2);
@@ -224,7 +224,7 @@ public class ReactiveConsumerTests {
@Test
public void testReactiveConsumerViaConsumerEndpointFactoryBean() throws Exception {
ReactiveChannel testChannel = new ReactiveChannel();
FluxMessageChannel testChannel = new FluxMessageChannel();
List<Message<?>> result = new LinkedList<>();
CountDownLatch stopLatch = new CountDownLatch(3);