diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index 2a5bb80..96ea395 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -57,6 +57,8 @@ dependencies { testCompile "org.mockito:mockito-all:$mockitoVersion" testCompile "org.springframework:spring-test:$springVersion" + testCompile "org.springframework.integration:spring-integration-event:$springIntegrationVersion" + jacoco group: "org.jacoco", name: "org.jacoco.agent", version: "0.5.6.201201232323", classifier: "runtime" } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java index 2bb37d1..c7137da 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java @@ -23,9 +23,9 @@ import org.springframework.messaging.MessageHandler; * @author Artem Bilan * @since 4.0 */ -public final class GenericEndpointSpec extends ConsumerEndpointSpec, H> { +public final class GenericEndpointSpec extends ConsumerEndpointSpec { - GenericEndpointSpec(H messageHandler) { + GenericEndpointSpec(MessageHandler messageHandler) { super(messageHandler); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index aad3672..7a1a436 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -16,7 +16,7 @@ package org.springframework.integration.dsl; -import org.springframework.expression.Expression; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; @@ -29,12 +29,14 @@ import org.springframework.integration.dsl.support.EndpointConfigurer; import org.springframework.integration.filter.ExpressionEvaluatingSelector; import org.springframework.integration.filter.MessageFilter; import org.springframework.integration.filter.MethodInvokingSelector; +import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.transformer.ExpressionEvaluatingTransformer; import org.springframework.integration.transformer.GenericTransformer; import org.springframework.integration.transformer.MessageTransformingHandler; import org.springframework.integration.transformer.MethodInvokingTransformer; import org.springframework.integration.transformer.Transformer; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; /** @@ -65,6 +67,11 @@ public final class IntegrationFlowBuilder { public IntegrationFlowBuilder channel(MessageChannel messageChannel) { Assert.notNull(messageChannel); + if (this.currentMessageChannel != null) { + GenericEndpointSpec endpointSpec = new GenericEndpointSpec(new BridgeHandler()); + endpointSpec.get().getT1().setInputChannel(this.currentMessageChannel); + this.addComponent(endpointSpec).currentComponent(endpointSpec.get().getT2()); + } this.currentMessageChannel = messageChannel; return this.addComponent(this.currentMessageChannel).registerOutputChannelIfCan(this.currentMessageChannel); } @@ -75,11 +82,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder transform(String expression) { - return this.transform(PARSER.parseExpression(expression)); - } - - public IntegrationFlowBuilder transform(Expression expression) { - return this.transform(new ExpressionEvaluatingTransformer(expression)); + return this.transform(new ExpressionEvaluatingTransformer(PARSER.parseExpression(expression))); } public IntegrationFlowBuilder transform(GenericTransformer genericTransformer) { @@ -87,18 +90,14 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder transform(GenericTransformer genericTransformer, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer endpointConfigurer) { Transformer transformer = genericTransformer instanceof Transformer ? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer); - return this.register(new GenericEndpointSpec(new MessageTransformingHandler(transformer)), endpointConfigurer); + return this.handle(new MessageTransformingHandler(transformer), endpointConfigurer); } public IntegrationFlowBuilder filter(String expression) { - return this.filter(PARSER.parseExpression(expression)); - } - - public IntegrationFlowBuilder filter(Expression expression) { - return this.filter(new ExpressionEvaluatingSelector(expression)); + return this.filter(new ExpressionEvaluatingSelector(PARSER.parseExpression(expression))); } public IntegrationFlowBuilder filter(GenericSelector genericSelector) { @@ -111,6 +110,18 @@ public final class IntegrationFlowBuilder { return this.register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer); } + public IntegrationFlowBuilder handle(MessageHandler messageHandler) { + return this.handle(messageHandler, null); + } + + public IntegrationFlowBuilder handle(MessageHandler messageHandler, EndpointConfigurer endpointConfigurer) { + return this.register(new GenericEndpointSpec(messageHandler), endpointConfigurer); + } + + public IntegrationFlowBuilder bridge(EndpointConfigurer endpointConfigurer) { + return this.register(new GenericEndpointSpec(new BridgeHandler()), endpointConfigurer); + } + private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) { this.flow.addComponent(outputChannel); if (this.currentComponent != null) { @@ -120,6 +131,10 @@ public final class IntegrationFlowBuilder { else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) { ((SourcePollingChannelAdapterFactoryBean) this.currentComponent).setOutputChannel(outputChannel); } + else { + throw new BeanCreationException("The 'currentComponent' (" + this.currentComponent + ") is a one-way 'MessageHandler'" + + "and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow."); + } this.currentComponent = null; } return this; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/ConsumerEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/ConsumerEndpointSpec.java index 87c43e8..4782ad7 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/ConsumerEndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/ConsumerEndpointSpec.java @@ -67,4 +67,12 @@ public abstract class ConsumerEndpointSpec, return _this(); } + public S sendTimeout(long sendTimeout) { + H handler = this.target.getT2(); + if (handler instanceof AbstractReplyProducingMessageHandler) { + ((AbstractReplyProducingMessageHandler) handler).setSendTimeout(sendTimeout); + } + return _this(); + } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/EndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/EndpointSpec.java index be8cd2c..033f511 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/EndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/EndpointSpec.java @@ -1,7 +1,7 @@ package org.springframework.integration.dsl.core; import org.springframework.beans.factory.BeanNameAware; -import org.springframework.core.ResolvableType; +import org.springframework.core.GenericTypeResolver; import org.springframework.integration.dsl.support.PollerSpec; import org.springframework.integration.dsl.tuple.Tuple; import org.springframework.integration.dsl.tuple.Tuple2; @@ -16,7 +16,7 @@ public abstract class EndpointSpec, F extends Be @SuppressWarnings("unchecked") protected EndpointSpec(H handler) { try { - Class fClass = ResolvableType.forClass(this.getClass()).as(EndpointSpec.class).resolveGenerics()[1]; + Class fClass = GenericTypeResolver.resolveTypeArguments(this.getClass(), EndpointSpec.class)[1]; F endpointFactoryBean = (F) fClass.newInstance(); this.target = Tuple.of(endpointFactoryBean, handler); } diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index dbc9d07..481e8a8 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java @@ -18,21 +18,27 @@ package org.springframework.integration.dsl.test; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.aopalliance.aop.Advice; import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationListener; import org.springframework.context.Lifecycle; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; @@ -51,6 +57,8 @@ import org.springframework.integration.dsl.channel.QueueChannelSpec; import org.springframework.integration.dsl.support.PollerSpec; import org.springframework.integration.dsl.support.Pollers; import org.springframework.integration.endpoint.MethodInvokingMessageSource; +import org.springframework.integration.event.core.MessagingEvent; +import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler; import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.store.SimpleMessageStore; @@ -60,6 +68,7 @@ import org.springframework.integration.transformer.PayloadSerializingTransformer import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.support.AnnotationConfigContextLoader; @@ -71,18 +80,43 @@ import org.springframework.test.context.support.AnnotationConfigContextLoader; @RunWith(SpringJUnit4ClassRunner.class) public class IntegrationFlowTests { + @Autowired + private BeanFactory beanFactory; + @Autowired @Qualifier("flow1QueueChannel") private PollableChannel outputChannel; @Autowired + @Qualifier("inputChannel") private DirectChannel inputChannel; @Autowired + @Qualifier("successChannel") private PollableChannel successChannel; @Autowired - private BeanFactory beanFactory; + @Qualifier("flow3Input") + private DirectChannel flow3Input; + + @Autowired + private AtomicReference eventHolder; + + @Autowired + @Qualifier("bridgeFlowInput") + private PollableChannel bridgeFlowInput; + + @Autowired + @Qualifier("bridgeFlowOutput") + private PollableChannel bridgeFlowOutput; + + @Autowired + @Qualifier("bridgeFlow2Input") + private DirectChannel bridgeFlow2Input; + + @Autowired + @Qualifier("bridgeFlow2Output") + private PollableChannel bridgeFlow2Output; @Test public void testPollingFlow() { @@ -119,6 +153,50 @@ public class IntegrationFlowTests { assertEquals(100, successMessage.getPayload()); } + @Test + public void testHandle() { + assertNull(this.eventHolder.get()); + this.flow3Input.send(new GenericMessage<>("foo")); + assertNotNull(this.eventHolder.get()); + assertEquals("foo", this.eventHolder.get()); + } + + @Test + public void testBridge() { + GenericMessage message = new GenericMessage<>("test"); + this.bridgeFlowInput.send(message); + Message reply = this.bridgeFlowOutput.receive(5000); + assertNotNull(reply); + assertEquals("test", reply.getPayload()); + + try { + this.bridgeFlow2Input.send(message); + fail("Expected MessageDispatchingException"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(MessageDeliveryException.class)); + assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class)); + assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers")); + } + this.beanFactory.getBean("bridge", Lifecycle.class).start(); + this.bridgeFlow2Input.send(message); + reply = this.bridgeFlow2Output.receive(5000); + assertNotNull(reply); + assertEquals("test", reply.getPayload()); + } + + @Test + public void testWrongLastComponent() { + try { + new AnnotationConfigApplicationContext(InvalidLastComponentFlowContext.class); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(BeanCreationException.class)); + assertThat(e.getMessage(), Matchers.containsString("is a one-way 'MessageHandler'")); + } + } + + @Configuration @EnableIntegration @@ -182,6 +260,7 @@ public class IntegrationFlowTests { .transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice())) .get(); } + } @MessageEndpoint @@ -193,5 +272,59 @@ public class IntegrationFlowTests { } } + @Configuration + public static class ContextConfiguration3 { + + @Bean + public AtomicReference eventHolder() { + return new AtomicReference<>(); + } + + @Bean + public ApplicationListener eventListener() { + return new ApplicationListener() { + + @Override + public void onApplicationEvent(MessagingEvent event) { + eventHolder().set(event.getMessage().getPayload()); + } + }; + } + + @Bean + public IntegrationFlow flow3() { + return IntegrationFlows.from(MessageChannels.direct("flow3Input")) + .handle(new ApplicationEventPublishingMessageHandler()) + .get(); + } + + @Bean + public IntegrationFlow bridgeFlow() { + return IntegrationFlows.from(MessageChannels.queue("bridgeFlowInput")) + .channel(MessageChannels.queue("bridgeFlowOutput")) + .get(); + } + + @Bean + public IntegrationFlow bridgeFlow2() { + return IntegrationFlows.from(MessageChannels.direct("bridgeFlow2Input")) + .bridge(c -> c.autoStartup(false).id("bridge")) + .channel(MessageChannels.queue("bridgeFlow2Output")) + .get(); + } + + } + + private static class InvalidLastComponentFlowContext { + + @Bean + public IntegrationFlow wrongLastComponent() { + return IntegrationFlows.from(MessageChannels.direct()) + .handle(Object::toString) + .channel(MessageChannels.direct()) + .get(); + } + + } }