diff --git a/build.gradle b/build.gradle index 70c72b5639..38b0f15e2d 100644 --- a/build.gradle +++ b/build.gradle @@ -254,6 +254,7 @@ project('spring-integration-file') { 'org.springframework.integration.*;version="[2.2.0, 2.2.1)"', 'org.springframework.beans.*;version="[3.1.1, 4.0.0)"', 'org.springframework.expression.*;version="[3.1.1, 4.0.0)"', + 'org.springframework.context.expression.*;version="[3.1.1, 4.0.0)"', 'org.springframework.context;version="[3.1.1, 4.0.0)"', 'org.springframework.context.expression.*;version="[3.1.1, 4.0.0)"', 'org.springframework.core.*;version="[3.1.1, 4.0.0)"', diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/PseudoTransactionalMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/core/PseudoTransactionalMessageSource.java index 026b349abd..065c5f351e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/PseudoTransactionalMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/PseudoTransactionalMessageSource.java @@ -38,27 +38,40 @@ import org.springframework.transaction.support.TransactionSynchronization; * @since 2.2 * */ -public interface PseudoTransactionalMessageSource extends MessageSource { +public interface PseudoTransactionalMessageSource extends MessageSource { /** * Obtain the resource on which appropriate action needs * to be taken. * @return The resource. */ - Object getResource(); + V getResource(); /** * Invoked via {@link TransactionSynchronization} when the * transaction commits. * @param resource The resource to be "committed" */ - void afterCommit(Object resource); + void afterCommit(V resource); /** * Invoked via {@link TransactionSynchronization} when the * transaction rolls back. * @param resource */ - void afterRollback(Object resource); + void afterRollback(V resource); + + /** + * Called when there is no transaction and the receive() call completed. + * @param resource + */ + void afterReceiveNoTx(V resource); + + /** + * Called when there is no transaction and after the message was + * sent to the channel. + * @param resource + */ + void afterSendNoTx(V resource); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 458f91e4b4..60a150fedd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -99,13 +99,14 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme super.onInit(); } + @SuppressWarnings("unchecked") @Override protected boolean doPoll() { boolean isInTx = false; - PseudoTransactionalMessageSource messageSource = null; + PseudoTransactionalMessageSource messageSource = null; Object resource = null; if (this.isPseudoTxMessageSource) { - messageSource = (PseudoTransactionalMessageSource) this.source; + messageSource = (PseudoTransactionalMessageSource) this.source; resource = messageSource.getResource(); Assert.state(resource != null, "Pseudo Transactional Message Source returned null resource"); if (this.synchronizedTx && TransactionSynchronizationManager.isActualTransactionActive()) { @@ -122,17 +123,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme } finally { if (this.isPseudoTxMessageSource && !isInTx) { - /* - * If the message source implements PseudoTransactionalMessageSource and - * we're running from a transactional poller, the message source's afterCommit - * method will be called by the transaction interceptor, using the transaction - * synchronization callback, after the transaction is committed. - * - * If we are not running in a transaction, we invoke it manually, so the message - * source can take the appropriate action, immediately after the receive; - * this was the behavior before pseudo transaction support was added. - */ - messageSource.afterCommit(resource); + messageSource.afterReceiveNoTx(resource); } } if (this.logger.isDebugEnabled()){ @@ -143,6 +134,9 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme message = MessageHistory.write(message, this); } this.messagingTemplate.send(this.outputChannel, message); + if (this.isPseudoTxMessageSource && !isInTx) { + messageSource.afterSendNoTx(resource); + } return true; } if (this.logger.isDebugEnabled()){ @@ -191,21 +185,23 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme return false; } + @SuppressWarnings("unchecked") @Override protected void processResourceAfterCommit(PseudoTransactionalResourceHolder resourceHolder) { if (logger.isTraceEnabled()) { logger.trace("'Committing' pseudo-transactional resource"); } - ((PseudoTransactionalMessageSource) source).afterCommit(resourceHolder.getResource()); + ((PseudoTransactionalMessageSource) source).afterCommit(resourceHolder.getResource()); } + @SuppressWarnings("unchecked") @Override public void afterCompletion(int status) { if (status != TransactionSynchronization.STATUS_COMMITTED) { if (logger.isTraceEnabled()) { logger.trace("'Rolling back' pseudo-transactional resource"); } - ((PseudoTransactionalMessageSource) source).afterRollback(this.resourceHolder.getResource()); + ((PseudoTransactionalMessageSource) source).afterRollback(this.resourceHolder.getResource()); } super.afterCompletion(status); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/ExpressionUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/ExpressionUtils.java new file mode 100644 index 0000000000..eb37830969 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/ExpressionUtils.java @@ -0,0 +1,86 @@ +/* + * Copyright 2002-2012 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.util; + +import org.springframework.context.expression.MapAccessor; +import org.springframework.core.convert.ConversionService; +import org.springframework.expression.BeanResolver; +import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.expression.spel.support.StandardTypeConverter; + +/** + * Utility class with static methods for helping with establishing environments for + * SpEL expressions. + * + * @author Gary Russell + * @since 2.2 + * + */ +public class ExpressionUtils { + + /** + * Create a {@link StandardEvaluationContext} with a {@link MapAccessor} in its + * property accessor property. + * @return the evaluation context. + */ + public static StandardEvaluationContext createStandardEvaluationContext() { + return createStandardEvaluationContext(null, null); + } + + /** + * Create a {@link StandardEvaluationContext} with a {@link MapAccessor} in its + * property accessor property and the supplied {@link BeanResolver} in its + * beanResolver property. + * @param beanResolver the bean factory. + * @return the evaluation context. + */ + public static StandardEvaluationContext createStandardEvaluationContext(BeanResolver beanResolver) { + return createStandardEvaluationContext(beanResolver, null); + } + + /** + * Create a {@link StandardEvaluationContext} with a {@link MapAccessor} in its + * property accessor property and the supplied {@link ConversionService} in its + * conversionService property. + * @param conversionService the conversion service. + * @return the evaluation context. + */ + public static StandardEvaluationContext createStandardEvaluationContext(ConversionService conversionService) { + return createStandardEvaluationContext(null, conversionService); + } + + /** + * Create a {@link StandardEvaluationContext} with a {@link MapAccessor} in its + * property accessor property, the supplied {@link BeanResolver} in its + * beanResolver property, and the supplied {@link ConversionService} in its + * conversionService property. + * @param beanResolver the bean factory. + * @param conversionService the conversion service. + * @return the evaluation context. + */ + public static StandardEvaluationContext createStandardEvaluationContext(BeanResolver beanResolver, + ConversionService conversionService) { + StandardEvaluationContext evaluationContext = new StandardEvaluationContext(); + evaluationContext.addPropertyAccessor(new MapAccessor()); + if (beanResolver != null) { + evaluationContext.setBeanResolver(beanResolver); + } + if (conversionService != null) { + evaluationContext.setTypeConverter(new StandardTypeConverter(conversionService)); + } + return evaluationContext; + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java index 9b8b020e79..96a2fed6bf 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java @@ -44,7 +44,7 @@ public class PseudoTransactionalMessageSourceTests { final Object object = new Object(); final AtomicReference committed = new AtomicReference(); final AtomicReference rolledBack = new AtomicReference(); - adapter.setSource(new PseudoTransactionalMessageSource() { + adapter.setSource(new PseudoTransactionalMessageSource() { public Message receive() { return new GenericMessage("foo"); @@ -61,6 +61,12 @@ public class PseudoTransactionalMessageSourceTests { public void afterRollback(Object resource) { rolledBack.set(resource); } + + public void afterReceiveNoTx(Object resource) { + } + + public void afterSendNoTx(Object resource) { + } }); TransactionSynchronizationManager.initSynchronization(); @@ -81,7 +87,7 @@ public class PseudoTransactionalMessageSourceTests { final Object object = new Object(); final AtomicReference committed = new AtomicReference(); final AtomicReference rolledBack = new AtomicReference(); - adapter.setSource(new PseudoTransactionalMessageSource() { + adapter.setSource(new PseudoTransactionalMessageSource() { public Message receive() { return new GenericMessage("foo"); @@ -98,6 +104,12 @@ public class PseudoTransactionalMessageSourceTests { public void afterRollback(Object resource) { rolledBack.set(resource); } + + public void afterReceiveNoTx(Object resource) { + } + + public void afterSendNoTx(Object resource) { + } }); TransactionSynchronizationManager.initSynchronization(); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java index f5084f2746..b80561534d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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,8 +19,9 @@ package org.springframework.integration.file; /** * Pre-defined header names to be used when storing or retrieving * File-related values to/from integration Message Headers. - * + * * @author Mark Fisher + * @author Gary Russell */ public abstract class FileHeaders { @@ -34,4 +35,6 @@ public abstract class FileHeaders { public static final String REMOTE_FILE = PREFIX + "remoteFile"; + public static final String DISPOSITION_RESULT = PREFIX + "dispositionResult"; + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index 43d1b473d7..767f40af02 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -26,14 +26,23 @@ import java.util.concurrent.PriorityBlockingQueue; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.context.expression.BeanFactoryResolver; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; import org.springframework.integration.MessagingException; import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageSource; +import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.core.PseudoTransactionalMessageSource; +import org.springframework.integration.file.FileReadingMessageSource.FileMessageHolder; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.util.ExpressionUtils; import org.springframework.util.Assert; /** @@ -50,7 +59,7 @@ import org.springframework.util.Assert; * file-writing process renames each file as soon as it is ready for reading. A * pattern-matching filter that accepts only files that are ready (e.g. based on * a known suffix), composed with the default {@link AcceptOnceFileListFilter} - * would allow for this. + * would allow for this. *

* A {@link Comparator} can be used to ensure internal ordering of the Files in * a {@link PriorityBlockingQueue}. This does not provide the same guarantees as @@ -59,12 +68,13 @@ import org.springframework.util.Assert; *

* FileReadingMessageSource is fully thread-safe under concurrent * receive() invocations and message delivery callbacks. - * + * * @author Iwein Fuld * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ -public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource { +public class FileReadingMessageSource extends IntegrationObjectSupport implements PseudoTransactionalMessageSource { private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5; @@ -86,6 +96,16 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement private volatile boolean scanEachPoll = false; + private volatile Expression dispositionExpression; + + private final MessagingTemplate dispositionMessagingTemplate = new MessagingTemplate(); + + private volatile boolean dispostionResultChannelSet; + + private final ThreadLocal resources = new ThreadLocal(); + + private EvaluationContext evaluationContext = new StandardEvaluationContext(); + /** * Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity. @@ -98,7 +118,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement * Creates a FileReadingMessageSource with a bounded queue of the given * capacity. This can be used to reduce the memory footprint of this * component when reading from a large directory. - * + * * @param internalQueueCapacity * the size of the queue used to cache files to be received * internally. This queue can be made larger to optimize the @@ -124,7 +144,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement * size of the queue is mutually exclusive with ordering. No guarantees * about file delivery order can be made under concurrent access. *

- * + * * @param receptionOrderComparator * the comparator to be used to order the files in the internal * queue @@ -137,7 +157,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement /** * Specify the input directory. - * + * * @param directory to monitor */ public void setDirectory(File directory) { @@ -148,7 +168,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement /** * Optionally specify a custom scanner, for example the * {@link org.springframework.integration.file.RecursiveLeafOnlyDirectoryScanner} - * + * * @param scanner scanner implementation */ public void setScanner(DirectoryScanner scanner) { @@ -161,7 +181,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement * true. If set to false and the * source directory does not exist, an Exception will be thrown upon * initialization. - * + * * @param autoCreateDirectory * should the directory to be monitored be created when this * component starts up? @@ -180,7 +200,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement * can be used to group them together. *

* The supplied filter must be thread safe.. - * + * * @param filter a filter */ public void setFilter(FileListFilter filter) { @@ -193,7 +213,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement * duplicate processing. *

* The supplied FileLocker must be thread safe - * + * * @param locker a locker */ public void setLocker(FileLocker locker) { @@ -212,7 +232,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement * will more likely be out of sync with the file system if this flag is set * to false, but it will change more often (causing expensive * reordering) if it is set to true. - * + * * @param scanEachPoll * whether or not the component should re-scan (as opposed to not * rescanning until the entire backlog has been delivered) @@ -221,10 +241,27 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement this.scanEachPoll = scanEachPoll; } + public void setDispositionExpression(Expression dispositionExpression) { + Assert.notNull(dispositionExpression, "'dispositionExpression' must not be null"); + this.dispositionExpression = dispositionExpression; + } + + public void setDispositionResultChannel(MessageChannel dispositionResultChannel) { + Assert.notNull(dispositionResultChannel, "'dispositionResultChannel' must not be null"); + this.dispositionMessagingTemplate.setDefaultChannel(dispositionResultChannel); + this.dispostionResultChannelSet = true; + } + + public void setDispositionSendTimeout(long dispositionSendTimeout) { + this.dispositionMessagingTemplate.setSendTimeout(dispositionSendTimeout); + } + + @Override public String getComponentType() { return "file:inbound-channel-adapter"; } + @Override protected void onInit() { Assert.notNull(directory, "'directory' must not be null"); if (!this.directory.exists() && this.autoCreateDirectory) { @@ -236,6 +273,10 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement "Source path [" + this.directory + "] does not point to a directory."); Assert.isTrue(this.directory.canRead(), "Source directory [" + this.directory + "] is not readable."); + if (getBeanFactory() != null) { + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext( + new BeanFactoryResolver(getBeanFactory())); + } } public Message receive() throws MessagingException { @@ -259,6 +300,11 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement if (logger.isInfoEnabled()) { logger.info("Created message: [" + message + "]"); } + FileMessageHolder resource = this.resources.get(); + if (resource == null) { + this.resources.set(new FileMessageHolder()); + } + this.resources.get().setMessage(message); } return message; } @@ -276,7 +322,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement /** * Adds the failed message back to the 'toBeReceived' queue if there is room. - * + * * @param failedMessage * the {@link org.springframework.integration.Message} that failed */ @@ -290,7 +336,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement /** * The message is just logged. It was already removed from the queue during * the call to receive() - * + * * @param sentMessage * the message that was successfully delivered */ @@ -300,4 +346,59 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement } } + public FileMessageHolder getResource() { + FileMessageHolder resource = new FileMessageHolder(); + this.resources.set(resource); + return resource; + } + + public void afterCommit(FileMessageHolder resource) { + Assert.isInstanceOf(FileMessageHolder.class, resource); + FileMessageHolder fileResource = resource; + if (this.dispositionExpression != null) { + if (logger.isDebugEnabled()) { + logger.debug("Executing expression " + this.dispositionExpression.getExpressionString() + " on " + + fileResource.getMessage()); + } + Object result = this.dispositionExpression.getValue(this.evaluationContext, fileResource.getMessage()); + if (result != null) { + if (this.dispostionResultChannelSet) { + try { + Message message = MessageBuilder.fromMessage(fileResource.getMessage()) + .setHeader(FileHeaders.DISPOSITION_RESULT, result).build(); + this.dispositionMessagingTemplate.send(message); + } + catch (Exception e) { + logger.error("Error sending File Disposition Result", e); + } + } + } + } + this.resources.set(null); + } + + public void afterRollback(FileMessageHolder resource) { + // no op + } + + public void afterReceiveNoTx(FileMessageHolder resource) { + // no op + } + + public void afterSendNoTx(FileMessageHolder resource) { + this.afterCommit(resource); + } + + class FileMessageHolder { + + private Message message; + + Message getMessage() { + return message; + } + + void setMessage(Message message) { + this.message = message; + } + } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java index ccd75a6d53..7838c5d496 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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,39 +16,47 @@ package org.springframework.integration.file.config; -import org.w3c.dom.Element; - import org.springframework.beans.BeanMetadataElement; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; +import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.ExpressionFactoryBean; import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.file.locking.NioFileLocker; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Element; /** * Parser for the <inbound-channel-adapter> element of the 'file' namespace. - * + * * @author Iwein Fuld * @author Mark Fisher + * @author Gary Russell */ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser { - private static final String PACKAGE_NAME = "org.springframework.integration.file"; - - @Override protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - PACKAGE_NAME + ".config.FileReadingMessageSourceFactoryBean"); + BeanDefinitionBuilder builder = + BeanDefinitionBuilder.genericBeanDefinition(FileReadingMessageSourceFactoryBean.class); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "comparator"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "scanner"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "directory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-create-directory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-size"); + String dispositionExpression = element.getAttribute("disposition-expression"); + if (StringUtils.hasText(dispositionExpression)) { + RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class); + expressionDef.getConstructorArgumentValues().addGenericArgumentValue(dispositionExpression); + builder.addPropertyValue("dispositionExpression", expressionDef); + } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "disposition-result-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "disposition-send-timeout"); String filterBeanName = this.registerFilter(element, parserContext); String lockerBeanName = registerLocker(element, parserContext); if (lockerBeanName != null) { @@ -64,8 +72,8 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann String lockerBeanName = null; Element nioLocker = DomUtils.getChildElementByTagName(element, "nio-locker"); if (nioLocker != null) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - PACKAGE_NAME + ".locking.NioFileLocker"); + BeanDefinitionBuilder builder = + BeanDefinitionBuilder.genericBeanDefinition(NioFileLocker.class); lockerBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName( builder.getBeanDefinition(), parserContext.getRegistry()); } @@ -79,8 +87,8 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann } private String registerFilter(Element element, ParserContext parserContext) { - BeanDefinitionBuilder factoryBeanBuilder = BeanDefinitionBuilder.genericBeanDefinition( - PACKAGE_NAME + ".config.FileListFilterFactoryBean"); + BeanDefinitionBuilder factoryBeanBuilder = + BeanDefinitionBuilder.genericBeanDefinition(FileListFilterFactoryBean.class); factoryBeanBuilder.setRole(BeanDefinition.ROLE_SUPPORT); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(factoryBeanBuilder, element, "filter"); String filenamePattern = element.getAttribute("filename-pattern"); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java index 42bfb798dc..a0514069c7 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java @@ -21,8 +21,9 @@ import java.util.Comparator; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.beans.factory.FactoryBean; +import org.springframework.expression.Expression; +import org.springframework.integration.MessageChannel; import org.springframework.integration.file.DirectoryScanner; import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.filters.CompositeFileListFilter; @@ -56,6 +57,12 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean + + + + SpEL expression to be executed after the message has been sent. If running in a transactional + poller, it will be executed after the transaction commits. If running in a non-transactional + poller it will execute after the message is sent. Note that the actual point of execution + depends on any asynchronous handoffs on the downstream flow. It will be executed when the + current thread returns from the channel send. The root object of the expression is the + original message (with a File payload). Examples: "payload.delete()", + "payload.renameTo('/foo/bar/' + payload.name)", "@someBean.doSomething(payload)". + + + + + + + If a 'disposition-expression' is provided, and that expression returns a result, the result + is sent to this channel, with the original message payload and a 'file_dispositionResult' + header containing the result of the expression execution. + + + + + + + + + + + + If a 'disposition-expression' is provided, and that expression returns a result, the result + is sent to this disposition-result-channel. This timeout specifies how long to wait if + that channel might block (such as a bounded queue channel that is full). Default infinity. + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java index 3fc29efdf0..9e6cfc0061 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java @@ -16,20 +16,32 @@ package org.springframework.integration.file; -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mock; -import org.mockito.runners.MockitoJUnitRunner; -import org.springframework.integration.Message; +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertEquals; +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.mockito.Matchers.isA; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.io.File; import java.io.IOException; import java.util.Comparator; -import static org.hamcrest.CoreMatchers.is; -import static org.junit.Assert.*; -import static org.mockito.Mockito.*; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.Message; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.file.FileReadingMessageSource.FileMessageHolder; +import org.springframework.integration.message.GenericMessage; /** * @author Iwein Fuld @@ -147,4 +159,18 @@ public class FileReadingMessageSourceTests { assertNull(source.receive()); verify(inputDirectoryMock, times(2)).listFiles(); } + + @Test + public void disposition() { + source.setDispositionExpression(new LiteralExpression("foo")); + QueueChannel channel = new QueueChannel(); + source.setDispositionResultChannel(channel); + FileMessageHolder resource = source.getResource(); + File file = mock(File.class); + resource.setMessage(new GenericMessage(file)); + source.afterCommit(resource); + Message result = channel.receive(10000); + assertSame(file, result.getPayload()); + assertEquals("foo", result.getHeaders().get(FileHeaders.DISPOSITION_RESULT)); + } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests-context.xml index 8dc2f5c842..7e76e7bd6f 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests-context.xml @@ -12,6 +12,8 @@ + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests.java index ae0b43b06d..7cf6972260 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileToChannelIntegrationTests.java @@ -16,6 +16,13 @@ package org.springframework.integration.file; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.io.File; + import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; @@ -24,11 +31,6 @@ import org.springframework.integration.core.PollableChannel; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import java.io.File; - -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; - /** * @author Iwein Fuld */ @@ -41,15 +43,23 @@ public class FileToChannelIntegrationTests { @Autowired PollableChannel fileMessages; + @Autowired + PollableChannel resultChannel; + @Test(timeout = 2000) public void fileMessageToChannel() throws Exception { - File.createTempFile("test", null, inputDirectory).setLastModified(System.currentTimeMillis() - 1000); + File file = File.createTempFile("test", null, inputDirectory); + file.setLastModified(System.currentTimeMillis() - 1000); Message received = receiveFileMessage(); while (received == null) { Thread.sleep(50); received = receiveFileMessage(); } assertNotNull(received.getPayload()); + Message result = resultChannel.receive(10000); + assertNotNull(result); + assertEquals(Boolean.TRUE, result.getHeaders().get(FileHeaders.DISPOSITION_RESULT)); + assertTrue(!file.exists()); } @SuppressWarnings("unchecked") diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml index 168e6c93ed..074fd6cb78 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml @@ -14,10 +14,15 @@ directory="${java.io.tmpdir}" filter="filter" comparator="testComparator" + disposition-expression="payload.delete()" + disposition-result-channel="resultChannel" + disposition-send-timeout="123" auto-startup="false"> + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java index a29ca6d45d..70334a2a9a 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java @@ -27,14 +27,16 @@ import java.util.concurrent.PriorityBlockingQueue; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; - import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpression; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.file.DefaultDirectoryScanner; import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; +import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -92,6 +94,14 @@ public class FileInboundChannelAdapterParserTests { assertSame("comparator reference not set, ", expected, actual); } + @Test + public void disposition() throws Exception { + Object dispositionExpression = accessor.getPropertyValue("dispositionExpression"); + assertEquals(SpelExpression.class, dispositionExpression.getClass()); + assertEquals("payload.delete()", ((Expression) dispositionExpression).getExpressionString()); + assertSame(TestUtils.getPropertyValue(source, "dispositionMessagingTemplate.defaultChannel"), context.getBean("resultChannel")); + assertEquals(123L, TestUtils.getPropertyValue(source, "dispositionMessagingTemplate.sendTimeout")); + } static class TestComparator implements Comparator { diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java index d14a469cf0..b2739854e9 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java @@ -28,11 +28,8 @@ import javax.servlet.http.HttpServletResponse; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.context.expression.BeanFactoryResolver; -import org.springframework.context.expression.MapAccessor; -import org.springframework.core.convert.ConversionService; import org.springframework.expression.Expression; import org.springframework.expression.spel.support.StandardEvaluationContext; -import org.springframework.expression.spel.support.StandardTypeConverter; import org.springframework.http.HttpEntity; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; @@ -59,6 +56,7 @@ import org.springframework.integration.http.multipart.MultipartHttpInputMessage; import org.springframework.integration.http.support.DefaultHttpHeaderMapper; import org.springframework.integration.mapping.HeaderMapper; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.util.ExpressionUtils; import org.springframework.util.AntPathMatcher; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -502,17 +500,13 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor } protected StandardEvaluationContext createEvaluationContext(){ - StandardEvaluationContext evaluationContext = new StandardEvaluationContext(); - evaluationContext.addPropertyAccessor(new MapAccessor()); - BeanFactory beanFactory = this.getBeanFactory(); - if (beanFactory != null) { - evaluationContext.setBeanResolver(new BeanFactoryResolver(beanFactory)); + if (this.getBeanFactory() != null) { + return ExpressionUtils.createStandardEvaluationContext(new BeanFactoryResolver(this.getBeanFactory()), + this.getConversionService()); } - ConversionService conversionService = this.getConversionService(); - if (conversionService != null) { - evaluationContext.setTypeConverter(new StandardTypeConverter(conversionService)); + else { + return ExpressionUtils.createStandardEvaluationContext(this.getConversionService()); } - return evaluationContext; } private void validateSupportedMethods() { diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java index aa8decad62..8c0501dcf6 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/PseudoTransactionalMessageSourceTests.java @@ -60,7 +60,7 @@ public class PseudoTransactionalMessageSourceTests { assertTrue(rolledBack); } - public static class MessageSource implements PseudoTransactionalMessageSource { + public static class MessageSource implements PseudoTransactionalMessageSource { public Message receive() { if (doRollback) { @@ -83,5 +83,11 @@ public class PseudoTransactionalMessageSourceTests { latch2.countDown(); } + public void afterReceiveNoTx(Object resource) { + } + + public void afterSendNoTx(Object resource) { + } + } } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java index cffd5db545..c5376bfcd8 100644 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java @@ -39,7 +39,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Gary Russell */ -public class MailReceivingMessageSource implements PseudoTransactionalMessageSource { +public class MailReceivingMessageSource implements PseudoTransactionalMessageSource { private final Log logger = LogFactory.getLog(this.getClass()); @@ -77,17 +77,31 @@ public class MailReceivingMessageSource implements PseudoTransactionalMessageSou return null; } - public Object getResource() { + public MailReceiverContext getResource() { return this.mailReceiver.getTransactionContext(); } - public void afterCommit(Object context) { + public void afterCommit(MailReceiverContext context) { Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext"); - this.mailReceiver.closeContextAfterSuccess((MailReceiverContext) context); + this.mailReceiver.closeContextAfterSuccess(context); } - public void afterRollback(Object context) { + public void afterRollback(MailReceiverContext context) { Assert.isTrue(context instanceof MailReceiverContext, "Expected a MailReceiverContext"); - this.mailReceiver.closeContextAfterFailure((MailReceiverContext) context); + this.mailReceiver.closeContextAfterFailure(context); + } + + /** + * For backwards-compatibility; the mail adapter updates the status before the send. + */ + public void afterReceiveNoTx(MailReceiverContext resource) { + this.afterCommit(resource); + } + + /** + * For backwards-compatibility; the mail adapter updates the status before the send. + */ + public void afterSendNoTx(MailReceiverContext resource) { + // No op } }