diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java new file mode 100644 index 0000000000..ec84af7adf --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java @@ -0,0 +1,136 @@ +/* + * Copyright 2017 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; + +import java.util.Iterator; + +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; + +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 reactor.core.publisher.Flux; +import reactor.core.publisher.FluxSink; +import reactor.core.publisher.Mono; + +/** + * Utilities for adaptation {@link MessageChannel}s to the {@link Publisher}s. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public final class MessageChannelReactiveUtils { + + private MessageChannelReactiveUtils() { + super(); + } + + @SuppressWarnings("unchecked") + public static Publisher> toPublisher(MessageChannel messageChannel) { + if (messageChannel instanceof Publisher) { + return (Publisher>) messageChannel; + } + else if (messageChannel instanceof SubscribableChannel) { + return adaptSubscribableChannelToPublisher((SubscribableChannel) messageChannel); + } + else if (messageChannel instanceof PollableChannel) { + return adaptPollableChannelToPublisher((PollableChannel) messageChannel); + } + else { + throw new IllegalArgumentException("The 'messageChannel' must be an instance of Publisher, " + + "SubscribableChannel or PollableChannel, not: " + messageChannel); + } + } + + private static Publisher> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) { + return new SubscribableChannelPublisherAdapter<>(inputChannel); + } + + private static Publisher> adaptPollableChannelToPublisher(PollableChannel inputChannel) { + return new PollableChannelPublisherAdapter<>(inputChannel); + } + + + private final static class SubscribableChannelPublisherAdapter implements Publisher> { + + private final SubscribableChannel channel; + + SubscribableChannelPublisherAdapter(SubscribableChannel channel) { + this.channel = channel; + } + + @Override + @SuppressWarnings("unchecked") + public void subscribe(Subscriber> subscriber) { + Flux. + >create(emitter -> { + MessageHandler messageHandler = emitter::next; + this.channel.subscribe(messageHandler); + emitter.setCancellation(() -> this.channel.unsubscribe(messageHandler)); + }, + FluxSink.OverflowStrategy.IGNORE) + .subscribe((Subscriber>) subscriber); + } + + } + + private final static class PollableChannelPublisherAdapter implements Publisher> { + + private final PollableChannel channel; + + PollableChannelPublisherAdapter(final PollableChannel channel) { + this.channel = channel; + } + + @Override + @SuppressWarnings("unchecked") + public void subscribe(Subscriber> subscriber) { + Iterator> messageIterator = new Iterator>() { + + private Message next = null; + + @Override + public Message next() { + Message message = this.next; + this.next = null; + return message; + } + + @Override + public boolean hasNext() { + if (this.next == null) { + this.next = PollableChannelPublisherAdapter.this.channel.receive(0); + } + return this.next != null; + } + + }; + + Mono.>delayMillis(100) + .repeat() + .concatMap(value -> Flux.fromIterable(() -> messageIterator)) + .subscribe((Subscriber>) subscriber); + } + + } + +} 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 1e5f58eab5..5eace718db 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,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -16,27 +16,40 @@ package org.springframework.integration.channel; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + 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.publisher.Operators; /** * @author Artem Bilan + * * @since 5.0 */ -public class ReactiveChannel implements MessageChannel, Publisher> { +public class ReactiveChannel extends AbstractMessageChannel + implements Publisher>, ReactiveSubscribableChannel { + + private final List>> subscribers = new ArrayList<>(); + + private final List>> publishers = new CopyOnWriteArrayList<>(); private final Processor, Message> processor; private final BlockingSink> sink; + private volatile boolean upstreamSubscribed; + public ReactiveChannel() { this(DirectProcessor.create()); } @@ -48,18 +61,55 @@ public class ReactiveChannel implements MessageChannel, Publisher> { } @Override - public boolean send(Message message) { - return send(message, -1); - } - - @Override - public boolean send(Message message, long timeout) { + protected boolean doSend(Message message, long timeout) { return this.sink.submit(message, timeout) > -1; } @Override public void subscribe(Subscriber> subscriber) { - this.processor.subscribe(subscriber); + this.subscribers.add(subscriber); + this.processor.subscribe(new Operators.SubscriberAdapter, Message>(subscriber) { + + @Override + protected void doCancel() { + super.doCancel(); + ReactiveChannel.this.subscribers.remove(subscriber); + } + + }); + + if (!this.upstreamSubscribed) { + this.publishers.forEach(this::doSubscribeTo); + } + } + + @Override + public void subscribeTo(Publisher> publisher) { + this.publishers.add(publisher); + if (!this.subscribers.isEmpty()) { + doSubscribeTo(publisher); + } + } + + private void doSubscribeTo(Publisher> publisher) { + publisher.subscribe(new Operators.SubscriberAdapter, Message>(this.processor) { + + @Override + protected void doOnSubscribe(Subscription subscription) { + super.doOnSubscribe(subscription); + ReactiveChannel.this.upstreamSubscribed = true; + } + + @Override + protected void doComplete() { + super.doComplete(); + ReactiveChannel.this.publishers.remove(publisher); + if (ReactiveChannel.this.publishers.isEmpty()) { + ReactiveChannel.this.upstreamSubscribed = false; + } + } + + }); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveSubscribableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveSubscribableChannel.java new file mode 100644 index 0000000000..36352266b9 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveSubscribableChannel.java @@ -0,0 +1,32 @@ +/* + * Copyright 2017 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; + +import org.reactivestreams.Publisher; + +import org.springframework.messaging.Message; + +/** + * @author Artem Bilan + * + * @since 5.0 + */ +public interface ReactiveSubscribableChannel { + + void subscribeTo(Publisher> publisher); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java index 2d1475ad9e..8a21be4fb9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -19,12 +19,15 @@ package org.springframework.integration.dsl; import java.util.Queue; import java.util.concurrent.Executor; +import org.reactivestreams.Processor; + import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.ExecutorChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; import org.springframework.integration.dsl.channel.PriorityChannelSpec; import org.springframework.integration.dsl.channel.PublishSubscribeChannelSpec; import org.springframework.integration.dsl.channel.QueueChannelSpec; +import org.springframework.integration.dsl.channel.ReactiveChannelSpec; import org.springframework.integration.dsl.channel.RendezvousChannelSpec; import org.springframework.integration.store.ChannelMessageStore; import org.springframework.integration.store.PriorityCapableChannelMessageStore; @@ -128,6 +131,23 @@ public class Channels { return MessageChannels.executor(id, executor); } + + public ReactiveChannelSpec reactive() { + return MessageChannels.reactive(); + } + + public ReactiveChannelSpec reactive(String id) { + return MessageChannels.reactive(id); + } + + public ReactiveChannelSpec reactive(Processor, Message> processor) { + return MessageChannels.reactive(processor); + } + + public ReactiveChannelSpec reactive(String id, Processor, Message> processor) { + return MessageChannels.reactive(id, processor); + } + Channels() { super(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index 7249f10881..5ff401a4a5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -38,6 +38,7 @@ import org.springframework.integration.aggregator.ResequencingMessageHandler; import org.springframework.integration.channel.ChannelInterceptorAware; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.FixedSubscriberChannel; +import org.springframework.integration.channel.MessageChannelReactiveUtils; import org.springframework.integration.channel.ReactiveChannel; import org.springframework.integration.channel.interceptor.WireTap; import org.springframework.integration.config.ConsumerEndpointFactoryBean; @@ -48,7 +49,6 @@ import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.channel.WireTapSpec; import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype; import org.springframework.integration.dsl.support.MessageChannelReference; -import org.springframework.integration.endpoint.ReactiveConsumer; import org.springframework.integration.expression.ControlBusMethodFilter; import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.filter.ExpressionEvaluatingSelector; @@ -2726,8 +2726,7 @@ public abstract class IntegrationFlowDefinition messagePublisher = ReactiveConsumer.adaptToPublisher(channelForPublisher); - publisher = (Publisher>) messagePublisher; + publisher = MessageChannelReactiveUtils.toPublisher(channelForPublisher); } else { MessageChannel reactiveChannel = new ReactiveChannel(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index 2fecbe601d..4cef221276 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -17,10 +17,12 @@ package org.springframework.integration.dsl; import java.util.function.Consumer; -import java.util.function.Function; + +import org.reactivestreams.Publisher; import org.springframework.integration.annotation.MessagingGateway; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.ReactiveChannel; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype; @@ -30,6 +32,7 @@ import org.springframework.integration.endpoint.MethodInvokingMessageSource; import org.springframework.integration.gateway.AnnotationGatewayProxyFactoryBean; import org.springframework.integration.gateway.GatewayProxyFactoryBean; import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; @@ -78,20 +81,6 @@ public final class IntegrationFlows { : from(messageChannelName); } - /** - * Populate the {@link MessageChannel} object to the - * {@link IntegrationFlowBuilder} chain using the fluent API from {@link Channels} factory. - * The {@link org.springframework.integration.dsl.IntegrationFlow} {@code inputChannel}. - * @param channels the {@link Function} to use method chain to configure. - * {@link MessageChannel} via {@link Channels} factory. - * @return new {@link IntegrationFlowBuilder}. - * @see Channels - */ - public static IntegrationFlowBuilder from(Function> channels) { - Assert.notNull(channels); - return from(channels.apply(new Channels())); - } - /** * Populate the {@link MessageChannel} object to the * {@link IntegrationFlowBuilder} chain using the fluent API from {@link MessageChannelSpec}. @@ -309,6 +298,18 @@ public final class IntegrationFlows { .addComponent(gatewayProxyFactoryBean); } + /** + * Populate a {@link ReactiveChannel} to the {@link IntegrationFlowBuilder} chain + * and subscribe it to the provided {@link Publisher}. + * @param publisher the {@link Publisher} to subscribe to. + * @return new {@link IntegrationFlowBuilder}. + */ + public static IntegrationFlowBuilder from(Publisher> publisher) { + ReactiveChannel reactiveChannel = new ReactiveChannel(); + reactiveChannel.subscribeTo(publisher); + return from((MessageChannel) reactiveChannel); + } + private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway, IntegrationFlowBuilder integrationFlowBuilder) { MessageChannel outputChannel = inboundGateway.getRequestChannel(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java index f1cd95ad5a..9fd2c590e0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -19,6 +19,8 @@ package org.springframework.integration.dsl.channel; import java.util.Queue; import java.util.concurrent.Executor; +import org.reactivestreams.Processor; + import org.springframework.integration.store.ChannelMessageStore; import org.springframework.integration.store.PriorityCapableChannelMessageStore; import org.springframework.messaging.Message; @@ -123,6 +125,22 @@ public final class MessageChannels { return MessageChannels.publishSubscribe(executor).id(id); } + public static ReactiveChannelSpec reactive() { + return new ReactiveChannelSpec(); + } + + public static ReactiveChannelSpec reactive(String id) { + return reactive().id(id); + } + + public static ReactiveChannelSpec reactive(Processor, Message> processor) { + return new ReactiveChannelSpec(processor); + } + + public static ReactiveChannelSpec reactive(String id, Processor, Message> processor) { + return reactive(processor).id(id); + } + private MessageChannels() { super(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java new file mode 100644 index 0000000000..ab85627199 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java @@ -0,0 +1,39 @@ +/* + * Copyright 2017 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.dsl.channel; + +import org.reactivestreams.Processor; + +import org.springframework.integration.channel.ReactiveChannel; +import org.springframework.messaging.Message; + +/** + * @author Artem Bilan + * + * @since 5.0 + */ +public class ReactiveChannelSpec extends MessageChannelSpec { + + ReactiveChannelSpec() { + this.channel = new ReactiveChannel(); + } + + ReactiveChannelSpec(Processor, Message> processor) { + this.channel = new ReactiveChannel(processor); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java index 4fc4bf8c61..429819848a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -16,20 +16,17 @@ package org.springframework.integration.endpoint; -import java.util.Iterator; import java.util.function.Consumer; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; +import org.springframework.integration.channel.MessageChannelReactiveUtils; 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; @@ -37,9 +34,6 @@ import reactor.core.Disposable; import reactor.core.Exceptions; import reactor.core.Receiver; import reactor.core.Trackable; -import reactor.core.publisher.Flux; -import reactor.core.publisher.FluxSink; -import reactor.core.publisher.Mono; import reactor.core.publisher.Operators; @@ -65,12 +59,8 @@ public class ReactiveConsumer extends AbstractEndpoint { Assert.notNull(inputChannel); Assert.notNull(subscriber); - if (inputChannel instanceof Publisher) { - this.publisher = (Publisher>) inputChannel; - } - else { - this.publisher = adaptToPublisher(inputChannel); - } + Publisher messagePublisher = MessageChannelReactiveUtils.toPublisher(inputChannel); + this.publisher = (Publisher>) messagePublisher; this.subscriber = new Operators.SubscriberAdapter, Message>(subscriber) { @@ -111,89 +101,6 @@ public class ReactiveConsumer extends AbstractEndpoint { this.subscriber.cancel(); } - public static 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 static Publisher> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) { - return new SubscribableChannelPublisherAdapter(inputChannel); - } - - private static Publisher> adaptPollableChannelToPublisher(PollableChannel inputChannel) { - return new PollableChannelPublisherAdapter(inputChannel); - } - - - private final static class SubscribableChannelPublisherAdapter implements Publisher> { - - private final SubscribableChannel channel; - - SubscribableChannelPublisherAdapter(SubscribableChannel channel) { - this.channel = channel; - } - - @Override - public void subscribe(Subscriber> subscriber) { - Flux. - >create(emitter -> { - MessageHandler messageHandler = emitter::next; - this.channel.subscribe(messageHandler); - emitter.setCancellation(() -> this.channel.unsubscribe(messageHandler)); - }, - FluxSink.OverflowStrategy.IGNORE) - .subscribe(subscriber); - } - - } - - private final static class PollableChannelPublisherAdapter implements Publisher> { - - private final PollableChannel channel; - - - PollableChannelPublisherAdapter(final PollableChannel channel) { - this.channel = channel; - } - - @Override - public void subscribe(Subscriber> subscriber) { - Iterator> messageIterator = new Iterator>() { - - private Message next = null; - - @Override - public Message next() { - Message message = this.next; - this.next = null; - return message; - } - - @Override - public boolean hasNext() { - if (this.next == null) { - this.next = PollableChannelPublisherAdapter.this.channel.receive(0); - } - return this.next != null; - } - - }; - - Mono.>delayMillis(100) - .repeat() - .concatMap(value -> Flux.fromIterable(() -> messageIterator)) - .subscribe(subscriber); - } - - } private static final class ConsumerSubscriber implements Subscriber>, Receiver, Disposable, Trackable { 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 f37c4e17c2..5baa1ff17e 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -16,11 +16,18 @@ package org.springframework.integration.channel.reactive; +import static org.hamcrest.Matchers.contains; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.isOneOf; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.junit.Test; import org.junit.runner.RunWith; @@ -29,16 +36,21 @@ 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.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; +import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; +import reactor.core.publisher.Flux; + /** * @author Artem Bilan * @since 5.0 @@ -50,6 +62,9 @@ public class ReactiveChannelTests { @Autowired private MessageChannel reactiveChannel; + @Autowired + private MessageChannel queueChannel; + @Test @SuppressWarnings("unchecked") public void testReactiveMessageChannel() throws InterruptedException { @@ -60,8 +75,9 @@ public class ReactiveChannelTests { this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build()); } catch (Exception e) { - assertThat(e.getCause(), instanceOf(MessageHandlingException.class)); - assertThat(e.getCause().getCause(), instanceOf(IllegalStateException.class)); + assertThat(e, instanceOf(MessageDeliveryException.class)); + assertThat(e.getCause().getCause(), instanceOf(MessageHandlingException.class)); + assertThat(e.getCause().getCause().getCause(), instanceOf(IllegalStateException.class)); assertThat(e.getMessage(), containsString("intentional")); } } @@ -73,6 +89,24 @@ public class ReactiveChannelTests { } } + @Test + public void testMessageChannelReactiveAdaptation() throws InterruptedException { + CountDownLatch done = new CountDownLatch(2); + List results = new ArrayList<>(); + + Flux.from(MessageChannelReactiveUtils.toPublisher(this.queueChannel)) + .map(Message::getPayload) + .map(String::toUpperCase) + .doOnNext(results::add) + .subscribe(v -> done.countDown()); + + this.queueChannel.send(new GenericMessage<>("foo")); + this.queueChannel.send(new GenericMessage<>("bar")); + + assertTrue(done.await(10, TimeUnit.SECONDS)); + assertThat(results, contains("FOO", "BAR")); + } + @Configuration @EnableIntegration public static class TestConfiguration { @@ -90,6 +124,11 @@ public class ReactiveChannelTests { return "" + payload; } + @Bean + public MessageChannel queueChannel() { + return new QueueChannel(); + } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java index 1def736cba..35e1e2693f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -242,7 +242,7 @@ public class CorrelationHandlerTests { @Bean @DependsOn("barrierFlow") public IntegrationFlow releaseBarrierFlow(MessageTriggerAction barrierTriggerAction) { - return IntegrationFlows.from(c -> c.queue("releaseChannel")) + return IntegrationFlows.from(MessageChannels.queue("releaseChannel")) .trigger(barrierTriggerAction, e -> e.poller(p -> p.fixedDelay(100))) .get(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java index eb5ac55f8b..2fe4896e92 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -33,7 +33,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.log4j.Level; -import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; import org.reactivestreams.Publisher; @@ -44,9 +43,12 @@ import org.springframework.context.Lifecycle; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.context.IntegrationFlowContext; import org.springframework.integration.test.rule.Log4jLevelAdjuster; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -67,7 +69,7 @@ import reactor.core.publisher.Flux; @DirtiesContext public class ReactiveStreamsTests { - @Rule +// @Rule public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.DEBUG, "org.springframework.integration"); @Autowired @@ -86,6 +88,9 @@ public class ReactiveStreamsTests { @Qualifier("inputChannel") private MessageChannel inputChannel; + @Autowired + private IntegrationFlowContext integrationFlowContext; + @Test public void testReactiveFlow() throws Exception { List results = new ArrayList<>(); @@ -99,7 +104,7 @@ public class ReactiveStreamsTests { this.messageSource.start(); assertTrue(latch.await(10, TimeUnit.SECONDS)); String[] strings = results.toArray(new String[results.size()]); - assertArrayEquals(new String[] { "A", "B", "C", "D", "E", "F" }, strings); + assertArrayEquals(new String[] {"A", "B", "C", "D", "E", "F"}, strings); this.messageSource.stop(); } @@ -119,17 +124,17 @@ public class ReactiveStreamsTests { Future> future = Executors.newSingleThreadExecutor().submit(() -> - Flux.just("11,12,13") - .map(v -> v.split(",")) - .flatMapIterable(Arrays::asList) - .map(Integer::parseInt) - .>map(GenericMessage::new) - .concatWith(this.pollablePublisher) - .take(7) - .map(Message::getPayload) - .log("org.springframework.integration.flux") - .collectList() - .block(Duration.ofSeconds(10)) + Flux.just("11,12,13") + .map(v -> v.split(",")) + .flatMapIterable(Arrays::asList) + .map(Integer::parseInt) + .>map(GenericMessage::new) + .concatWith(this.pollablePublisher) + .take(7) + .map(Message::getPayload) + .log("org.springframework.integration.flux") + .collectList() + .block(Duration.ofSeconds(10)) ); this.inputChannel.send(new GenericMessage<>("6,7,8,9,10")); @@ -141,6 +146,33 @@ public class ReactiveStreamsTests { assertEquals(7, integers.size()); } + @Test + public void testFromPublisher() { + Flux> messageFlux = Flux.just("1,2,3,4") + .map(v -> v.split(",")) + .flatMapIterable(Arrays::asList) + .map(Integer::parseInt) + .log("org.springframework.integration.flux") + .map(GenericMessage::new); + + QueueChannel resultChannel = new QueueChannel(); + + IntegrationFlow integrationFlow = + IntegrationFlows.from(messageFlux) + .log("org.springframework.integration.flux2") + .transform(p -> p * 2) + .channel(resultChannel) + .get(); + + this.integrationFlowContext.registration(integrationFlow) + .register(); + + for (int i = 0; i < 4; i++) { + Message receive = resultChannel.receive(10000); + assertNotNull(receive); + assertEquals((i + 1) * 2, receive.getPayload()); + } + } @Configuration @EnableIntegration