From 8cc38bc1b75294caa61d43983f8e612c1d944219 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 12 Aug 2014 22:25:45 +0300 Subject: [PATCH] DSL: Add `File.` Namespace Builder Factory * Introduce `ComponentsRegistration` *marker* to extract internal components from the `IntegrationComponentSpec` to be registered as bean in the application context * Add `ConsumerEndpointSpec#order(int order)` for the `order` of target `AbstractMessageHandler` * Fix `AggregatorSpec` to use `DefaultAggregatingMessageGroupProcessor` by default * Fix `IntegrationFlowBeanPostProcessor#generateBeanName` to check if `instance instanceof NamedComponent` and its `beanName` has been configured using `IntegrationComponentSpec#id` --- .../integration/dsl/AggregatorSpec.java | 3 +- .../dsl/IntegrationFlowBuilder.java | 11 +- .../integration/dsl/IntegrationFlows.java | 75 +++++++-- .../dsl/core/ComponentsRegistration.java | 28 ++++ .../dsl/core/ConsumerEndpointSpec.java | 13 +- .../IntegrationFlowBeanPostProcessor.java | 4 + .../dsl/core/MessagingProducerSpec.java | 2 +- .../integration/dsl/file/File.java | 55 +++++++ .../file/FileInboundChannelAdapterSpec.java | 138 ++++++++++++++++ .../file/FileWritingMessageHandlerSpec.java | 107 +++++++++++++ .../integration/dsl/file/TailAdapterSpec.java | 136 ++++++++++++++++ .../integration/dsl/file/package-info.java | 4 + .../dsl/test/IntegrationFlowTests.java | 148 ++++++++++++++---- 13 files changed, 670 insertions(+), 54 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/ComponentsRegistration.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/File.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileInboundChannelAdapterSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/TailAdapterSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/package-info.java diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java index a9573db..d23963d 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java @@ -17,6 +17,7 @@ package org.springframework.integration.dsl; import org.springframework.integration.aggregator.AggregatingMessageHandler; +import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor; import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor; import org.springframework.integration.aggregator.MessageGroupProcessor; import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor; @@ -26,7 +27,7 @@ import org.springframework.integration.aggregator.MethodInvokingMessageGroupProc */ public class AggregatorSpec extends CorrelationHandlerSpec { - private MessageGroupProcessor outputProcessor; + private MessageGroupProcessor outputProcessor = new DefaultAggregatingMessageGroupProcessor(); private boolean expireGroupsUponCompletion; 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 424f294..6b1a95e 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 @@ -22,7 +22,6 @@ import org.springframework.beans.factory.BeanCreationException; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler; import org.springframework.integration.aggregator.AggregatingMessageHandler; -import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor; import org.springframework.integration.aggregator.ResequencingMessageHandler; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.FixedSubscriberChannel; @@ -94,6 +93,13 @@ public final class IntegrationFlowBuilder { return this; } + IntegrationFlowBuilder addComponents(Collection components) { + for (Object component : components) { + this.flow.addComponent(component); + } + return this; + } + IntegrationFlowBuilder currentComponent(Object component) { this.currentComponent = component; return this; @@ -406,8 +412,7 @@ public final class IntegrationFlowBuilder { public IntegrationFlowBuilder aggregate(EndpointConfigurer> endpointConfigurer) { - return handle(new AggregatorSpec().outputProcessor(new DefaultAggregatingMessageGroupProcessor()).get(), - endpointConfigurer); + return handle(new AggregatorSpec().get(), endpointConfigurer); } public IntegrationFlowBuilder aggregate(ComponentConfigurer aggregatorConfigurer, diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index 3aedc0f..e93a455 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -20,6 +20,7 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.channel.MessageChannelSpec; +import org.springframework.integration.dsl.core.ComponentsRegistration; import org.springframework.integration.dsl.core.MessageSourceSpec; import org.springframework.integration.dsl.core.MessagingGatewaySpec; import org.springframework.integration.dsl.core.MessagingProducerSpec; @@ -56,22 +57,22 @@ public final class IntegrationFlows { return from(new FixedSubscriberChannelPrototype(messageChannelName)); } - public static IntegrationFlowBuilder from(MessageChannel messageChannel) { - return new IntegrationFlowBuilder().channel(messageChannel); - } - public static IntegrationFlowBuilder from(MessageChannelSpec messageChannelSpec) { return from(messageChannelSpec.get()); } - public static >> IntegrationFlowBuilder from(S - messageSourceSpec) { - return from(messageSourceSpec.get()); + public static IntegrationFlowBuilder from(MessageChannel messageChannel) { + return new IntegrationFlowBuilder().channel(messageChannel); } - public static >> IntegrationFlowBuilder from(S messageSource, - EndpointConfigurer endpointConfigurer) { - return from(messageSource.get(), endpointConfigurer); + public static >> IntegrationFlowBuilder + from(S messageSourceSpec) { + return from(messageSourceSpec, null); + } + + public static >> IntegrationFlowBuilder + from(S messageSourceSpec, EndpointConfigurer endpointConfigurer) { + return from(messageSourceSpec.get(), endpointConfigurer, registerComponents(messageSourceSpec)); } public static IntegrationFlowBuilder from(MessageSource messageSource) { @@ -80,31 +81,50 @@ public final class IntegrationFlows { public static IntegrationFlowBuilder from(MessageSource messageSource, EndpointConfigurer endpointConfigurer) { + return from(messageSource, endpointConfigurer, null); + } + + private static IntegrationFlowBuilder from(MessageSource messageSource, + EndpointConfigurer endpointConfigurer, + IntegrationFlowBuilder integrationFlowBuilder) { SourcePollingChannelAdapterSpec spec = new SourcePollingChannelAdapterSpec(messageSource); if (endpointConfigurer != null) { endpointConfigurer.configure(spec); } - return new IntegrationFlowBuilder() - .addComponent(spec) + if (integrationFlowBuilder == null) { + integrationFlowBuilder = new IntegrationFlowBuilder(); + } + return integrationFlowBuilder.addComponent(spec) .currentComponent(spec); } public static IntegrationFlowBuilder from(MessagingProducerSpec messagingProducerSpec) { - return from(messagingProducerSpec.get()); + return from(messagingProducerSpec.get(), registerComponents(messagingProducerSpec)); } public static IntegrationFlowBuilder from(MessageProducerSupport messageProducer) { + return from(messageProducer, null); + } + + private static IntegrationFlowBuilder from(MessageProducerSupport messageProducer, + IntegrationFlowBuilder integrationFlowBuilder) { DirectFieldAccessor dfa = new DirectFieldAccessor(messageProducer); MessageChannel outputChannel = (MessageChannel) dfa.getPropertyValue("outputChannel"); if (outputChannel == null) { outputChannel = new DirectChannel(); messageProducer.setOutputChannel(outputChannel); } - return from(outputChannel).addComponent(messageProducer); + if (integrationFlowBuilder == null) { + integrationFlowBuilder = from(outputChannel); + } + else { + integrationFlowBuilder.channel(outputChannel); + } + return integrationFlowBuilder.addComponent(messageProducer); } public static IntegrationFlowBuilder from(MessagingGatewaySpec inboundGatewaySpec) { - return from(inboundGatewaySpec.get()); + return from(inboundGatewaySpec.get(), registerComponents(inboundGatewaySpec)); } public static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway) { @@ -117,6 +137,31 @@ public final class IntegrationFlows { return from(outputChannel).addComponent(inboundGateway); } + private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway, + IntegrationFlowBuilder integrationFlowBuilder) { + DirectFieldAccessor dfa = new DirectFieldAccessor(inboundGateway); + MessageChannel outputChannel = (MessageChannel) dfa.getPropertyValue("requestChannel"); + if (outputChannel == null) { + outputChannel = new DirectChannel(); + inboundGateway.setRequestChannel(outputChannel); + } + if (integrationFlowBuilder == null) { + integrationFlowBuilder = from(outputChannel); + } + else { + integrationFlowBuilder.channel(outputChannel); + } + return integrationFlowBuilder.addComponent(inboundGateway); + } + + private static IntegrationFlowBuilder registerComponents(Object spec) { + if (spec instanceof ComponentsRegistration) { + return new IntegrationFlowBuilder() + .addComponents(((ComponentsRegistration) spec).getComponentsToRegister()); + } + return null; + } + private IntegrationFlows() { } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/ComponentsRegistration.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/ComponentsRegistration.java new file mode 100644 index 0000000..7db8656 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/ComponentsRegistration.java @@ -0,0 +1,28 @@ +/* + * Copyright 2014 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.core; + +import java.util.Collection; + +/** + * @author Artem Bilan + */ +public interface ComponentsRegistration { + + Collection getComponentsToRegister(); + +} 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 babec37..e35a1d4 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 @@ -23,13 +23,13 @@ import java.util.List; import org.aopalliance.aop.Advice; import org.springframework.integration.config.ConsumerEndpointFactoryBean; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.messaging.MessageHandler; /** * @author Artem Bilan - */ public abstract class ConsumerEndpointSpec, H extends MessageHandler> extends EndpointSpec { @@ -89,4 +89,15 @@ public abstract class ConsumerEndpointSpec, return _this(); } + public S order(int order) { + H handler = this.target.getT2(); + if (handler instanceof AbstractMessageHandler) { + ((AbstractMessageHandler) handler).setOrder(order); + } + else { + logger.warn("'order' can be applied only for AbstractMessageHandler"); + } + return _this(); + } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java index a6ea734..bb85ba8 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java @@ -34,6 +34,7 @@ import org.springframework.integration.config.SourcePollingChannelAdapterFactory import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec; import org.springframework.integration.dsl.support.MessageChannelReference; +import org.springframework.integration.support.context.NamedComponent; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; @@ -156,6 +157,9 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean } private String generateBeanName(Object instance) { + if (instance instanceof NamedComponent && ((NamedComponent) instance).getComponentName() != null) { + return ((NamedComponent) instance).getComponentName(); + } String generatedBeanName = instance.getClass().getName(); String id = instance.getClass().getName(); int counter = -1; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingProducerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingProducerSpec.java index 41848b4..5747cca 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingProducerSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingProducerSpec.java @@ -55,7 +55,7 @@ public abstract class MessagingProducerSpec receptionOrderComparator) { + return new FileInboundChannelAdapterSpec(receptionOrderComparator).directory(directory); + } + + public static FileWritingMessageHandlerSpec outboundAdapter(java.io.File destinationDirectory) { + return new FileWritingMessageHandlerSpec(destinationDirectory).expectReply(false); + } + + public static FileWritingMessageHandlerSpec outboundAdapter(String directoryExpression) { + return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(false); + } + + public static FileWritingMessageHandlerSpec outboundGateway(java.io.File destinationDirectory) { + return new FileWritingMessageHandlerSpec(destinationDirectory).expectReply(true); + } + + public static FileWritingMessageHandlerSpec outboundGateway(String directoryExpression) { + return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(true); + } + + public static TailAdapterSpec tailAdapter(java.io.File file) { + return new TailAdapterSpec().file(file); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileInboundChannelAdapterSpec.java new file mode 100644 index 0000000..51b4718 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileInboundChannelAdapterSpec.java @@ -0,0 +1,138 @@ +/* + * Copyright 2014 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.file; + +import java.io.File; +import java.util.Comparator; + +import org.springframework.integration.dsl.core.MessageSourceSpec; +import org.springframework.integration.file.DirectoryScanner; +import org.springframework.integration.file.FileLocker; +import org.springframework.integration.file.FileReadingMessageSource; +import org.springframework.integration.file.filters.AcceptAllFileListFilter; +import org.springframework.integration.file.filters.AcceptOnceFileListFilter; +import org.springframework.integration.file.filters.CompositeFileListFilter; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.filters.RegexPatternFileListFilter; +import org.springframework.integration.file.filters.SimplePatternFileListFilter; +import org.springframework.integration.file.locking.NioFileLocker; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + */ +public class FileInboundChannelAdapterSpec + extends MessageSourceSpec { + + private FileListFilter filter; + + private FileLocker locker; + + FileInboundChannelAdapterSpec() { + this.target = new FileReadingMessageSource(); + } + + FileInboundChannelAdapterSpec(Comparator receptionOrderComparator) { + this.target = new FileReadingMessageSource(receptionOrderComparator); + } + + FileInboundChannelAdapterSpec directory(File directory) { + this.target.setDirectory(directory); + return _this(); + } + + public FileInboundChannelAdapterSpec scanner(DirectoryScanner scanner) { + this.target.setScanner(scanner); + return _this(); + } + + public FileInboundChannelAdapterSpec autoCreateDirectory(boolean autoCreateDirectory) { + this.target.setAutoCreateDirectory(autoCreateDirectory); + return _this(); + } + + public FileInboundChannelAdapterSpec filter(FileListFilter filter) { + return filter(filter, false); + } + + public FileInboundChannelAdapterSpec filter(FileListFilter filter, boolean preventDuplicates) { + Assert.isNull(this.filter, + "The 'filter' (" + this.filter + ") is already configured for the FileReadingMessageSource"); + FileListFilter targetFilter = filter; + if (preventDuplicates) { + targetFilter = createCompositeWithAcceptOnceFilter(filter); + } + this.filter = targetFilter; + this.target.setFilter(targetFilter); + return _this(); + } + + public FileInboundChannelAdapterSpec preventDuplicatesFilter(boolean preventDuplicates) { + if (preventDuplicates) { + return filter(new AcceptOnceFileListFilter(), false); + } + else { + return filter(new AcceptAllFileListFilter(), false); + } + } + + public FileInboundChannelAdapterSpec patternFilter(String pattern) { + return patternFilter(pattern, true); + } + + public FileInboundChannelAdapterSpec patternFilter(String pattern, boolean preventDuplicates) { + return filter(new SimplePatternFileListFilter(pattern), preventDuplicates); + } + + public FileInboundChannelAdapterSpec regexFilter(String regex) { + return regexFilter(regex, true); + } + + public FileInboundChannelAdapterSpec regexFilter(String regex, boolean preventDuplicates) { + return filter(new RegexPatternFileListFilter(regex), preventDuplicates); + } + + private CompositeFileListFilter createCompositeWithAcceptOnceFilter(FileListFilter otherFilter) { + CompositeFileListFilter compositeFilter = new CompositeFileListFilter(); + compositeFilter.addFilter(new AcceptOnceFileListFilter()); + compositeFilter.addFilter(otherFilter); + return compositeFilter; + } + + public FileInboundChannelAdapterSpec locker(FileLocker locker) { + Assert.isNull(this.locker, + "The 'locker' (" + this.locker + ") is already configured for the FileReadingMessageSource"); + this.locker = locker; + this.target.setLocker(locker); + return _this(); + } + + public FileInboundChannelAdapterSpec nioLocker() { + return locker(new NioFileLocker()); + } + + public FileInboundChannelAdapterSpec scanEachPoll(boolean scanEachPoll) { + this.target.setScanEachPoll(scanEachPoll); + return _this(); + } + + @Override + protected FileReadingMessageSource doGet() { + throw new UnsupportedOperationException(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java new file mode 100644 index 0000000..9995ea6 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java @@ -0,0 +1,107 @@ +/* + * Copyright 2014 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.file; + +import java.util.Collection; +import java.util.Collections; + +import org.springframework.integration.dsl.core.ComponentsRegistration; +import org.springframework.integration.dsl.core.MessageHandlerSpec; +import org.springframework.integration.file.DefaultFileNameGenerator; +import org.springframework.integration.file.FileNameGenerator; +import org.springframework.integration.file.FileWritingMessageHandler; +import org.springframework.integration.file.support.FileExistsMode; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + */ +public class FileWritingMessageHandlerSpec + extends MessageHandlerSpec + implements ComponentsRegistration { + + private FileNameGenerator fileNameGenerator; + + private DefaultFileNameGenerator defaultFileNameGenerator; + + FileWritingMessageHandlerSpec(java.io.File destinationDirectory) { + this.target = new FileWritingMessageHandler(destinationDirectory); + } + + FileWritingMessageHandlerSpec(String directoryExpression) { + this.target = new FileWritingMessageHandler(PARSER.parseExpression(directoryExpression)); + } + + FileWritingMessageHandlerSpec expectReply(boolean expectReply) { + target.setExpectReply(expectReply); + return _this(); + } + + public FileWritingMessageHandlerSpec autoCreateDirectory(boolean autoCreateDirectory) { + target.setAutoCreateDirectory(autoCreateDirectory); + return _this(); + } + + public FileWritingMessageHandlerSpec temporaryFileSuffix(String temporaryFileSuffix) { + target.setTemporaryFileSuffix(temporaryFileSuffix); + return _this(); + } + + public FileWritingMessageHandlerSpec fileExistsMode(FileExistsMode fileExistsMode) { + target.setFileExistsMode(fileExistsMode); + return _this(); + } + + public FileWritingMessageHandlerSpec fileNameGenerator(FileNameGenerator fileNameGenerator) { + this.fileNameGenerator = fileNameGenerator; + target.setFileNameGenerator(fileNameGenerator); + return _this(); + } + + public FileWritingMessageHandlerSpec fileNameGeneratorExpression(String fileNameGeneratorExpression) { + Assert.isNull(this.fileNameGenerator, + "'fileNameGenerator' and 'fileNameGeneratorExpression' are mutually exclusive."); + this.defaultFileNameGenerator = new DefaultFileNameGenerator(); + this.defaultFileNameGenerator.setExpression(fileNameGeneratorExpression); + return _this(); + } + + + public FileWritingMessageHandlerSpec deleteSourceFiles(boolean deleteSourceFiles) { + target.setDeleteSourceFiles(deleteSourceFiles); + return _this(); + } + + public FileWritingMessageHandlerSpec charset(String charset) { + target.setCharset(charset); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + if (this.defaultFileNameGenerator != null) { + return Collections.singletonList(this.defaultFileNameGenerator); + } + return Collections.emptyList(); + } + + @Override + protected FileWritingMessageHandler doGet() { + throw new UnsupportedOperationException(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/TailAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/TailAdapterSpec.java new file mode 100644 index 0000000..27c0fbc --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/TailAdapterSpec.java @@ -0,0 +1,136 @@ +/* + * Copyright 2014 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.file; + +import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.dsl.core.MessagingProducerSpec; +import org.springframework.integration.file.config.FileTailInboundChannelAdapterFactoryBean; +import org.springframework.integration.file.tail.FileTailingMessageProducerSupport; +import org.springframework.messaging.MessageChannel; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + */ +public class TailAdapterSpec extends MessagingProducerSpec { + + private final FileTailInboundChannelAdapterFactoryBean factoryBean = new FileTailInboundChannelAdapterFactoryBean(); + + private MessageChannel outputChannel; + + private MessageChannel errorChannel; + + TailAdapterSpec() { + super(null); + this.factoryBean.setBeanFactory(new DefaultListableBeanFactory()); + } + + TailAdapterSpec file(java.io.File file) { + Assert.notNull(file); + this.factoryBean.setFile(file); + return _this(); + } + + public TailAdapterSpec nativeOptions(String nativeOptions) { + this.factoryBean.setNativeOptions(nativeOptions); + return _this(); + } + + public TailAdapterSpec taskExecutor(TaskExecutor taskExecutor) { + this.factoryBean.setTaskExecutor(taskExecutor); + return _this(); + } + + public TailAdapterSpec taskScheduler(TaskScheduler taskScheduler) { + this.factoryBean.setTaskScheduler(taskScheduler); + return _this(); + } + + public TailAdapterSpec delay(long delay) { + this.factoryBean.setDelay(delay); + return _this(); + } + + public TailAdapterSpec fileDelay(long fileDelay) { + this.factoryBean.setFileDelay(fileDelay); + return _this(); + } + + public TailAdapterSpec end(boolean end) { + this.factoryBean.setEnd(end); + return _this(); + } + + public TailAdapterSpec reopen(boolean reopen) { + this.factoryBean.setReopen(reopen); + return _this(); + } + + @Override + public TailAdapterSpec id(String id) { + this.factoryBean.setBeanName(id); + return _this(); + } + + @Override + public TailAdapterSpec phase(int phase) { + this.factoryBean.setPhase(phase); + return _this(); + } + + @Override + public TailAdapterSpec autoStartup(boolean autoStartup) { + this.factoryBean.setAutoStartup(autoStartup); + return _this(); + } + + @Override + public TailAdapterSpec outputChannel(MessageChannel outputChannel) { + this.outputChannel = outputChannel; + return _this(); + } + + @Override + public TailAdapterSpec errorChannel(MessageChannel errorChannel) { + this.errorChannel = errorChannel; + return _this(); + } + + @Override + protected FileTailingMessageProducerSupport doGet() { + if (this.outputChannel == null) { + this.factoryBean.setOutputChannel(new NullChannel()); + } + FileTailingMessageProducerSupport tailingMessageProducerSupport = null; + try { + this.factoryBean.afterPropertiesSet(); + tailingMessageProducerSupport = this.factoryBean.getObject(); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + if (this.errorChannel != null) { + tailingMessageProducerSupport.setErrorChannel(this.errorChannel); + } + tailingMessageProducerSupport.setOutputChannel(this.outputChannel); + return tailingMessageProducerSupport; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/package-info.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/package-info.java new file mode 100644 index 0000000..0294723 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides File Components support for Spring Integration Java DSL. + */ +package org.springframework.integration.dsl.file; 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 85756f9..c1b924e 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 @@ -17,11 +17,20 @@ package org.springframework.integration.dsl.test; import static org.hamcrest.Matchers.*; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; -import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; import java.io.FileOutputStream; import java.io.IOException; +import java.nio.charset.Charset; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -34,13 +43,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; -import com.mongodb.MongoClient; -import de.flapdoodle.embed.mongo.MongodExecutable; -import de.flapdoodle.embed.mongo.MongodStarter; -import de.flapdoodle.embed.mongo.config.MongodConfigBuilder; -import de.flapdoodle.embed.mongo.config.Net; -import de.flapdoodle.embed.mongo.distribution.Version; -import de.flapdoodle.embed.process.runtime.Network; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; @@ -92,15 +94,15 @@ import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.amqp.Amqp; import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.file.File; import org.springframework.integration.dsl.jms.Jms; import org.springframework.integration.dsl.support.Pollers; +import org.springframework.integration.dsl.support.Transformers; 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.file.tail.ApacheCommonsFileTailingMessageProducer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice; import org.springframework.integration.mongodb.store.MongoDbChannelMessageStore; @@ -111,6 +113,7 @@ import org.springframework.integration.store.PriorityCapableChannelMessageStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.MutableMessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.transformer.PayloadDeserializingTransformer; import org.springframework.integration.transformer.PayloadSerializingTransformer; import org.springframework.integration.xml.transformer.support.XPathExpressionEvaluatingHeaderValueMessageProcessor; @@ -135,6 +138,15 @@ import org.springframework.stereotype.Service; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.StreamUtils; + +import com.mongodb.MongoClient; +import de.flapdoodle.embed.mongo.MongodExecutable; +import de.flapdoodle.embed.mongo.MongodStarter; +import de.flapdoodle.embed.mongo.config.MongodConfigBuilder; +import de.flapdoodle.embed.mongo.config.Net; +import de.flapdoodle.embed.mongo.distribution.Version; +import de.flapdoodle.embed.process.runtime.Network; /** * @author Artem Bilan @@ -144,7 +156,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @DirtiesContext public class IntegrationFlowTests { - private static final File tmpDir = new File(System.getProperty("java.io.tmpdir")); + private static final java.io.File tmpDir = new java.io.File(System.getProperty("java.io.tmpdir")); private static int mongoPort; @@ -203,7 +215,7 @@ public class IntegrationFlowTests { private MessageChannel fileFlow1Input; @Autowired - @Qualifier("fileWritingMessageHandler") + @Qualifier("fileWriting.handler") private MessageHandler fileWritingMessageHandler; @Autowired @@ -337,11 +349,12 @@ public class IntegrationFlowTests { assertNotNull(message); assertEquals("" + i, message.getPayload()); } + this.controlBus.send("@integerEndpoint.stop()"); assertTrue(((ChannelInterceptorAware) this.outputChannel).getChannelInterceptors() .contains(this.testChannelInterceptor)); assertThat(this.testChannelInterceptor.getInvoked(), Matchers.greaterThanOrEqualTo(5)); - this.controlBus.send("@integerEndpoint.stop()"); + } @Test @@ -476,7 +489,7 @@ public class IntegrationFlowTests { dfa.setPropertyValue("fileNameGenerator", fileNameGenerator); this.fileFlow1Input.send(message); - assertTrue(new File(tmpDir, "foo").exists()); + assertTrue(new java.io.File(tmpDir, "foo").exists()); } @Test @@ -492,7 +505,7 @@ public class IntegrationFlowTests { } @Test - public void testLamdas() { + public void testLambdas() { QueueChannel replyChannel = new QueueChannel(); Message message = MessageBuilder.withPayload("World") .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) @@ -867,12 +880,10 @@ public class IntegrationFlowTests { @Test public void testMessageProducerFlow() throws Exception { - FileOutputStream file = new FileOutputStream(new File(tmpDir, "TailTest")); + FileOutputStream file = new FileOutputStream(new java.io.File(tmpDir, "TailTest")); for (int i = 0; i < 50; i++) { file.write((i + "\n").getBytes()); } - file.flush(); - file.close(); for (int i = 0; i < 50; i++) { Message message = this.tailChannel.receive(5000); @@ -880,6 +891,9 @@ public class IntegrationFlowTests { assertEquals("hello " + i, message.getPayload()); } assertNull(this.tailChannel.receive(1)); + + this.controlBus.send("@tailer.stop()"); + file.close(); } @@ -991,6 +1005,59 @@ public class IntegrationFlowTests { assertEquals("HELLO THROUGH THE JMS PIPELINE", receive.getPayload()); } + @Autowired + @Qualifier("fileReadingResultChannel") + private PollableChannel fileReadingResultChannel; + + @Test + public void testFileReadingFlow() throws Exception { + List evens = new ArrayList<>(25); + for (int i = 0; i < 50; i++) { + boolean even = i % 2 == 0; + String extension = even ? ".sitest" : ".foofile"; + if (even) { + evens.add(i); + } + FileOutputStream file = new FileOutputStream(new java.io.File(tmpDir, i + extension)); + file.write(("" + i).getBytes()); + file.flush(); + file.close(); + } + + Message message = fileReadingResultChannel.receive(10000); + assertNotNull(message); + Object payload = message.getPayload(); + assertThat(payload, instanceOf(List.class)); + @SuppressWarnings("unchecked") + List result = (List) payload; + assertEquals(25, result.size()); + result.forEach(s -> assertTrue(evens.contains(Integer.parseInt(s)))); + } + + + @Autowired + @Qualifier("fileWritingInput") + private MessageChannel fileWritingInput; + + @Autowired + @Qualifier("fileWritingResultChannel") + private PollableChannel fileWritingResultChannel; + + @Test + public void testFileWritingFlow() throws Exception { + String payload = "Spring Integration"; + this.fileWritingInput.send(new GenericMessage<>(payload)); + Message receive = this.fileWritingResultChannel.receive(1000); + assertNotNull(receive); + assertThat(receive.getPayload(), instanceOf(java.io.File.class)); + java.io.File resultFile = (java.io.File) receive.getPayload(); + assertThat(resultFile.getAbsolutePath(), + endsWith(TestUtils.applySystemFileSeparator("fileWritingFlow/foo.sitest"))); + String fileContent = StreamUtils.copyToString(new FileInputStream(resultFile), Charset.defaultCharset()); + assertEquals(payload, fileContent); + } + + @MessagingGateway(defaultRequestChannel = "controlBus") private static interface ControlBusGateway { @@ -1021,7 +1088,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow flow1() { return IntegrationFlows.from(this.integerMessageSource(), - c -> c.poller(Pollers.fixedRate(1000, 2000)) + c -> c.poller(Pollers.fixedRate(100)) .id("integerEndpoint") .autoStartup(false)) .fixedSubscriberChannel("integerChannel") @@ -1296,18 +1363,11 @@ public class IntegrationFlowTests { @Configuration public static class ContextConfiguration4 { - @Bean - public MessageHandler fileWritingMessageHandler() { - FileWritingMessageHandler fileWritingMessageHandler = new FileWritingMessageHandler(tmpDir); - fileWritingMessageHandler.setFileNameGenerator(message -> null); - fileWritingMessageHandler.setExpectReply(false); - return fileWritingMessageHandler; - } - @Bean public IntegrationFlow fileFlow1() { return IntegrationFlows.from("fileFlow1Input") - .handle(this.fileWritingMessageHandler()) + .handle(File.outboundAdapter(tmpDir).fileNameGenerator(message -> null), + c -> c.id("fileWriting")) .get(); } @@ -1464,10 +1524,9 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow tailFlow() { - ApacheCommonsFileTailingMessageProducer adapter = new ApacheCommonsFileTailingMessageProducer(); - adapter.setFile(new File(tmpDir, "TailTest")); - - return IntegrationFlows.from(adapter) + return IntegrationFlows.from(File.tailAdapter(new java.io.File(tmpDir, "TailTest")) + .delay(500) + .id("tailer")) .transform("hello "::concat) .channel(MessageChannels.queue("tailChannel")) .get(); @@ -1540,6 +1599,29 @@ public class IntegrationFlowTests { return MessageChannels.queue().get(); } + + @Bean + public IntegrationFlow fileReadingFlow() { + return IntegrationFlows + .from(File.inboundAdapter(tmpDir).patternFilter("*.sitest"), + e -> e.poller(Pollers.fixedDelay(100))) + .transform(Transformers.fileToString()) + .aggregate(a -> a.correlationExpression("1") + .releaseExpression("size() == 25"), null) + .channel(MessageChannels.queue("fileReadingResultChannel")) + .get(); + } + + @Bean + public IntegrationFlow fileWritingFlow() { + return IntegrationFlows.from("fileWritingInput") + .enrichHeaders(h -> h.header(FileHeaders.FILENAME, "foo.sitest") + .header("directory", new java.io.File(tmpDir, "fileWritingFlow"))) + .handle(File.outboundGateway("headers[directory]")) + .channel(MessageChannels.queue("fileWritingResultChannel")) + .get(); + } + } private static class RoutingTestBean {