diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index 96ea395..e499f9f 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -58,6 +58,7 @@ dependencies { testCompile "org.springframework:spring-test:$springVersion" testCompile "org.springframework.integration:spring-integration-event:$springIntegrationVersion" + testCompile "org.springframework.integration:spring-integration-file:$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/FilterEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java index b1a0607..d22f6df 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java @@ -22,7 +22,7 @@ import org.springframework.messaging.MessageChannel; /** * @author Artem Bilan - * @since 4.0 + */ public final class FilterEndpointSpec extends ConsumerEndpointSpec { 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 c7137da..fd04107 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 @@ -21,11 +21,11 @@ import org.springframework.messaging.MessageHandler; /** * @author Artem Bilan - * @since 4.0 - */ -public final class GenericEndpointSpec extends ConsumerEndpointSpec { - GenericEndpointSpec(MessageHandler messageHandler) { + */ +public final class GenericEndpointSpec extends ConsumerEndpointSpec, H> { + + GenericEndpointSpec(H 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 7a1a436..f0004d7 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 @@ -25,6 +25,7 @@ import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessageSelector; import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.core.ConsumerEndpointSpec; +import org.springframework.integration.dsl.support.BeanNameMethodInvokingMessageHandler; import org.springframework.integration.dsl.support.EndpointConfigurer; import org.springframework.integration.filter.ExpressionEvaluatingSelector; import org.springframework.integration.filter.MessageFilter; @@ -68,7 +69,7 @@ public final class IntegrationFlowBuilder { public IntegrationFlowBuilder channel(MessageChannel messageChannel) { Assert.notNull(messageChannel); if (this.currentMessageChannel != null) { - GenericEndpointSpec endpointSpec = new GenericEndpointSpec(new BridgeHandler()); + GenericEndpointSpec endpointSpec = new GenericEndpointSpec(new BridgeHandler()); endpointSpec.get().getT1().setInputChannel(this.currentMessageChannel); this.addComponent(endpointSpec).currentComponent(endpointSpec.get().getT2()); } @@ -90,7 +91,7 @@ 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.handle(new MessageTransformingHandler(transformer), endpointConfigurer); @@ -114,12 +115,20 @@ public final class IntegrationFlowBuilder { return this.handle(messageHandler, null); } - public IntegrationFlowBuilder handle(MessageHandler messageHandler, EndpointConfigurer endpointConfigurer) { - return this.register(new GenericEndpointSpec(messageHandler), endpointConfigurer); + public IntegrationFlowBuilder handle(String target, String methodName) { + return this.handle(target, methodName, null); } - public IntegrationFlowBuilder bridge(EndpointConfigurer endpointConfigurer) { - return this.register(new GenericEndpointSpec(new BridgeHandler()), endpointConfigurer); + public IntegrationFlowBuilder handle(String beanName, String methodName, EndpointConfigurer> endpointConfigurer) { + return this.handle(new BeanNameMethodInvokingMessageHandler(beanName, methodName) , endpointConfigurer); + } + + public IntegrationFlowBuilder handle(H 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) { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java index 4504410..25217e4 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java @@ -7,7 +7,7 @@ import org.springframework.integration.scheduling.PollerMetadata; /** * @author Artem Bilan - * @since 4.0 + */ public final class SourcePollingChannelAdapterSpec extends EndpointSpec> { 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 4782ad7..0c3d76f 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 @@ -29,7 +29,7 @@ import org.springframework.messaging.MessageHandler; /** * @author Artem Bilan - * @since 4.0 + */ public abstract class ConsumerEndpointSpec, H extends MessageHandler> extends EndpointSpec { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java index 250f975..7f505e6 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java @@ -47,7 +47,8 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig public void initialize(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException { Assert.isInstanceOf(BeanDefinitionRegistry.class, configurableListableBeanFactory, "To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." + - "Consider using 'GenericApplicationContext' implementation."); + "Consider using 'GenericApplicationContext' implementation." + ); this.initializeIntegrationFlows(configurableListableBeanFactory); this.populateBeansFromSpecs(configurableListableBeanFactory); } @@ -78,10 +79,15 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig ConsumerEndpointFactoryBean endpoint = endpointSpec.get().getT1(); String id = endpointSpec.getId(); - String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler); - String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null; - BeanComponentDefinition definitionHolder = new BeanComponentDefinition(new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias); - BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry); + Collection messageHandlers = beanFactory.getBeansOfType(messageHandler.getClass(), false, false).values(); + + if (!messageHandlers.contains(messageHandler)) { + String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler); + String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null; + BeanComponentDefinition definitionHolder = new BeanComponentDefinition(new InstanceBeanDefinition(messageHandler), + handlerBeanName, handlerAlias); + BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry); + } String endpointBeanName = id; if (endpointBeanName == null) { @@ -111,9 +117,7 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig for (Map.Entry specEntry : specs.entrySet()) { String id = specEntry.getKey(); IntegrationComponentSpec spec = (IntegrationComponentSpec) specEntry.getValue(); - registry.removeBeanDefinition(id); - beanFactory.registerSingleton(id, spec.get()); - beanFactory.initializeBean(spec.get(), id); + registry.registerBeanDefinition(id, new InstanceBeanDefinition(spec.get())); } } 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 033f511..2d23795 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.GenericTypeResolver; +import org.springframework.core.ResolvableType; import org.springframework.integration.dsl.support.PollerSpec; import org.springframework.integration.dsl.tuple.Tuple; import org.springframework.integration.dsl.tuple.Tuple2; @@ -9,14 +9,14 @@ import org.springframework.integration.scheduling.PollerMetadata; /** * @author Artem Bilan - * @since 4.0 + */ public abstract class EndpointSpec, F extends BeanNameAware, H> extends IntegrationComponentSpec> { @SuppressWarnings("unchecked") protected EndpointSpec(H handler) { try { - Class fClass = GenericTypeResolver.resolveTypeArguments(this.getClass(), EndpointSpec.class)[1]; + Class fClass = ResolvableType.forClass(this.getClass()).as(EndpointSpec.class).resolveGenerics()[1]; F endpointFactoryBean = (F) fClass.newInstance(); this.target = Tuple.of(endpointFactoryBean, handler); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationComponentSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationComponentSpec.java index bb135a8..361bbef 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationComponentSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationComponentSpec.java @@ -18,7 +18,7 @@ package org.springframework.integration.dsl.core; /** * @author Artem Bilan - * @since 4.0 + */ public abstract class IntegrationComponentSpec, T> { @@ -31,7 +31,7 @@ public abstract class IntegrationComponentSpec processor; + + public BeanNameMethodInvokingMessageHandler(String object, String methodName) { + this.object = object; + this.methodName = methodName; + } + + @Override + protected void doInit() { + Object target = this.getBeanFactory().getBean(object); + this.processor = new MethodInvokingMessageProcessor(target, this.methodName); + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + return this.processor.processMessage(requestMessage); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java index fc22f15..d22caa5 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java @@ -20,7 +20,7 @@ import org.springframework.integration.dsl.core.EndpointSpec; /** * @author Artem Bilan - * @since 4.0 + */ public interface EndpointConfigurer> { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java index 5854f52..6e90d0a 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java @@ -34,7 +34,7 @@ import org.springframework.util.ErrorHandler; /** * @author Artem Bilan -* @since 4.0 + */ public final class PollerSpec extends IntegrationComponentSpec { 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 481e8a8..fe74cc8 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 @@ -23,6 +23,7 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.io.File; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -32,10 +33,9 @@ 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.ListableBeanFactory; 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; @@ -59,6 +59,9 @@ 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.file.DefaultFileNameGenerator; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.FileWritingMessageHandler; import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.store.SimpleMessageStore; @@ -67,8 +70,11 @@ import org.springframework.integration.transformer.PayloadDeserializingTransform import org.springframework.integration.transformer.PayloadSerializingTransformer; import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; +import org.springframework.stereotype.Component; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.support.AnnotationConfigContextLoader; @@ -80,8 +86,10 @@ import org.springframework.test.context.support.AnnotationConfigContextLoader; @RunWith(SpringJUnit4ClassRunner.class) public class IntegrationFlowTests { + private static final File tmpDir = new File(System.getProperty("java.io.tmpdir")); + @Autowired - private BeanFactory beanFactory; + private ListableBeanFactory beanFactory; @Autowired @Qualifier("flow1QueueChannel") @@ -118,6 +126,18 @@ public class IntegrationFlowTests { @Qualifier("bridgeFlow2Output") private PollableChannel bridgeFlow2Output; + @Autowired + @Qualifier("fileFlow1Input") + private DirectChannel fileFlow1Input; + + @Autowired + @Qualifier("fileWritingMessageHandler") + private FileWritingMessageHandler fileWritingMessageHandler; + + @Autowired + @Qualifier("methodInvokingInput") + private DirectChannel methodInvokingInput; + @Test public void testPollingFlow() { for (int i = 0; i < 10; i++) { @@ -197,6 +217,34 @@ public class IntegrationFlowTests { } + @Test + public void testFileHandler() { + assertEquals(1, this.beanFactory.getBeansOfType(FileWritingMessageHandler.class).size()); + Message message = MessageBuilder.withPayload("foo").setHeader(FileHeaders.FILENAME, "foo").build(); + try { + this.fileFlow1Input.send(message); + fail("NullPointerException expected"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(MessageHandlingException.class)); + assertThat(e.getCause(), Matchers.instanceOf(NullPointerException.class)); + } + this.fileWritingMessageHandler.setFileNameGenerator(new DefaultFileNameGenerator()); + this.fileFlow1Input.send(message); + + assertTrue(new File(tmpDir, "foo").exists()); + } + + @Test + public void testMethodInvokingMessageHandler() { + QueueChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("world").setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel).build(); + this.methodInvokingInput.send(message); + Message receive = replyChannel.receive(5000); + assertNotNull(receive); + assertEquals("Hello, world", receive.getPayload()); + } + @Configuration @EnableIntegration @@ -315,6 +363,41 @@ public class IntegrationFlowTests { } + @Configuration + public static class ContextConfiguration4 { + + @Bean + public FileWritingMessageHandler fileWritingMessageHandler() { + return new FileWritingMessageHandler(tmpDir); + } + + @Bean + public IntegrationFlow fileFlow1() { + return IntegrationFlows.from(MessageChannels.direct("fileFlow1Input")) + .handle(this.fileWritingMessageHandler(), c -> { + FileWritingMessageHandler handler = c.get().getT2(); + handler.setFileNameGenerator(message -> null); + handler.setExpectReply(false); }) + .get(); + } + + @Bean + public IntegrationFlow methodInvokingFlow() { + return IntegrationFlows.from(MessageChannels.direct("methodInvokingInput")) + .handle("greetingService", null) + .get(); + } + } + + @Component("greetingService") + public static class GreetingService { + + public String greeting(String payload) { + return "Hello, " + payload; + } + } + + private static class InvalidLastComponentFlowContext { @Bean