From 295af552421fe4184e1bdbb751b7b09dde9f1c6f Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 5 Nov 2010 13:53:59 -0400 Subject: [PATCH] polishing --- .../file/DefaultDirectoryScanner.java | 93 ++-- .../file/DefaultFileNameGenerator.java | 5 +- .../integration/file/DirectoryScanner.java | 85 ++-- .../integration/file/FileHeaders.java | 2 +- .../integration/file/FileLocker.java | 2 +- .../integration/file/FileNameGenerator.java | 2 +- .../file/FileReadingMessageSource.java | 440 ++++++++++-------- .../file/FileWritingMessageHandler.java | 11 +- .../file/HeadDirectoryScanner.java | 39 +- .../RecursiveLeafOnlyDirectoryScanner.java | 50 +- ...nelAdapterWithRecursiveDirectoryTests.java | 2 +- 11 files changed, 402 insertions(+), 329 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/DefaultDirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/DefaultDirectoryScanner.java index aff9f362ec..87b41a67cb 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/DefaultDirectoryScanner.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/DefaultDirectoryScanner.java @@ -13,66 +13,71 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.file; +import java.io.File; +import java.util.Arrays; +import java.util.List; + import org.springframework.integration.MessagingException; import org.springframework.integration.file.entries.AcceptOnceEntryFileListFilter; import org.springframework.integration.file.entries.EntryListFilter; -import java.io.File; - -import java.util.List; - - /** - * Default directory scanner and base class for other directory scanners. It takes care of the default interrelations - * between filtering, scanning and locking. - * + * Default directory scanner and base class for other directory scanners. + * Manages the default interrelations between filtering, scanning and locking. + * * @author Iwein Fuld * @since 2.0 */ public class DefaultDirectoryScanner implements DirectoryScanner { - private EntryListFilter filter = new AcceptOnceEntryFileListFilter(); - private FileLocker locker; - public final List listFiles(File directory) throws IllegalArgumentException { - File[] files = listEligibleFiles(directory); + private volatile EntryListFilter filter = new AcceptOnceEntryFileListFilter(); - if (files == null) { - throw new MessagingException("The path [" + directory + "] does not denote a properly accessible directory."); - } + private volatile FileLocker locker; - return this.filter.filterEntries(files); - } - /** - * Subclasses may refine the listing strategy by overriding this method. The files returned here are passed onto the - * filter. - * - * @param directory root directory to use for listing - * @return the files this scanner should consider - */ - protected File[] listEligibleFiles(File directory) { - return directory.listFiles(); - } + public void setFilter(EntryListFilter filter) { + this.filter = filter; + } - public void setFilter(EntryListFilter filter) { - this.filter = filter; - } + /** + * {@inheritDoc} + */ + public final void setLocker(FileLocker locker) { + this.locker = locker; + } - /** - * {@inheritDoc} - *

- * This class takes the minimal implementation and merely delegates to the locker if set. - */ - public final boolean tryClaim(File file) { - return (locker == null) || locker.lock(file); - } - /** - * {@inheritDoc} - */ - public final void setLocker(FileLocker locker) { - this.locker = locker; - } + /** + * {@inheritDoc} + *

+ * This class takes the minimal implementation and merely delegates to the + * locker if set. + */ + public final boolean tryClaim(File file) { + return (this.locker == null) || this.locker.lock(file); + } + + public final List listFiles(File directory) throws IllegalArgumentException { + File[] files = listEligibleFiles(directory); + if (files == null) { + throw new MessagingException("The path [" + directory + + "] does not denote a properly accessible directory."); + } + return (this.filter != null) ? this.filter.filterEntries(files) : Arrays.asList(files); + } + + /** + * Subclasses may refine the listing strategy by overriding this method. The + * files returned here are passed onto the filter. + * + * @param directory root directory to use for listing + * @return the files this scanner should consider + */ + protected File[] listEligibleFiles(File directory) { + return directory.listFiles(); + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/DefaultFileNameGenerator.java b/spring-integration-file/src/main/java/org/springframework/integration/file/DefaultFileNameGenerator.java index ad46cc549b..20fdc3a97e 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/DefaultFileNameGenerator.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/DefaultFileNameGenerator.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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. @@ -49,8 +49,7 @@ public class DefaultFileNameGenerator implements FileNameGenerator { public String generateFileName(Message message) { Object filenameProperty = message.getHeaders().get(this.headerName); - if (filenameProperty instanceof String - && StringUtils.hasText((String) filenameProperty)) { + if (filenameProperty instanceof String && StringUtils.hasText((String) filenameProperty)) { return (String) filenameProperty; } if (message.getPayload() instanceof File) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/DirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/DirectoryScanner.java index bcd3d69396..c9090e3277 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/DirectoryScanner.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/DirectoryScanner.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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. @@ -22,52 +22,57 @@ import java.io.File; import java.util.List; /** - * Strategy for scanning directories. Implementations may select all children and grandchildren of the scanned directory - * in any order. This interface is intended to enable the customization of selection, locking and ordering of files in a - * directory like RecursiveDirectoryScanner. If the only requirement is to ignore certain files a EntryListFilter - * implementation should suffice. - * + * Strategy for scanning directories. Implementations may select all children + * and grandchildren of the scanned directory in any order. This interface is + * intended to enable the customization of selection, locking and ordering of + * files in a directory like RecursiveDirectoryScanner. If the only requirement + * is to ignore certain files a EntryListFilter implementation should suffice. * - * * @author Iwein Fuld */ public interface DirectoryScanner { - /** - * Scans the directory according to the strategy particular to this implementation and returns the selected files as - * a File array. This method may never return files that are rejected by the filter. - * - * @param directory the directory to scan for files - * @return a list of files representing the content of the directory - * @throws IllegalArgumentException thrown if the input is incorrect - */ - List listFiles(File directory) throws IllegalArgumentException; + /** + * Scans the directory according to the strategy particular to this + * implementation and returns the selected files as a File array. This + * method may never return files that are rejected by the filter. + * + * @param directory the directory to scan for files + * @return a list of files representing the content of the directory + * @throws IllegalArgumentException if the input is incorrect + */ + List listFiles(File directory) throws IllegalArgumentException; - /** - * Sets a custom filter to be used by this scanner. The filter will get a chance to reject files before the scanner - * presents them through its listFiles method. A scanner may use additional filtering that is out of the control of - * the provided filter. - * - * @param filter the custom filter to be used - */ - void setFilter(EntryListFilter filter); + /** + * Sets a custom filter to be used by this scanner. The filter will get a + * chance to reject files before the scanner presents them through its + * listFiles method. A scanner may use additional filtering that is out of + * the control of the provided filter. + * + * @param filter + * the custom filter to be used + */ + void setFilter(EntryListFilter filter); + /** + * Sets a custom locker to be used by this scanner. The locker will get a + * chance to lock files and reject claims on files that are already locked. + * + * @param locker + * the custom locker to be used + */ + void setLocker(FileLocker locker); - /** - * Claim the file to process. It is up to the implementation to decide what additional safe guards are required to - * attain a claim to the file. But if a locker is set implementations MUST invoke its lock method and - * MUST return false if the locker did not grant the lock. - * - * @param file file to be claimed - * @return true if the claim was granted false otherwise - */ - boolean tryClaim(File file); + /** + * Claim the file to process. It is up to the implementation to decide what + * additional safe guards are required to attain a claim to the file. But if + * a locker is set implementations MUST invoke its lock method + * and MUST return false if the locker did not grant the lock. + * + * @param file + * file to be claimed + * @return true if the claim was granted false otherwise + */ + boolean tryClaim(File file); - /** - * Sets a custom locker to be used by this scanner. The locker will get a chance to lock files and reject claims on - * files that are already locked. - * - * @param locker the custom locker to be used - */ - void setLocker(FileLocker locker); } 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 d14537ddfb..3bfbdaf65f 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-2009 the original author or authors. + * Copyright 2002-2010 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. diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileLocker.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileLocker.java index e96883d806..06b02a358f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileLocker.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileLocker.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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. diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileNameGenerator.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileNameGenerator.java index 7fda8e92a3..0c7ca53aeb 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileNameGenerator.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileNameGenerator.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2010 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. 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 3eade29e94..89b39b9d8d 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-2009 the original author or authors. + * Copyright 2002-2010 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. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.file; import java.io.File; @@ -34,232 +35,275 @@ import org.springframework.integration.file.entries.EntryListFilter; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; - /** - * {@link MessageSource} that creates messages from a file system directory. To prevent messages for certain files, you - * may supply a {@link org.springframework.integration.file.entries.EntryListFilter}. By default, - * an {@link org.springframework.integration.file.entries.AcceptOnceEntryFileListFilter} is used. It ensures files are - * picked up only once from the directory. + * {@link MessageSource} that creates messages from a file system directory. To + * prevent messages for certain files, you may supply a + * {@link org.springframework.integration.file.entries.EntryListFilter}. By + * default, an + * {@link org.springframework.integration.file.entries.AcceptOnceEntryFileListFilter} + * is used. It ensures files are picked up only once from the directory. *

- * A common problem with reading files is that a file may be detected before it is ready. The default {@link - * org.springframework.integration.file.entries.AcceptOnceEntryFileListFilter} does not prevent this. In most cases, this can be prevented if the 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 org.springframework.integration.file.entries.AcceptOnceEntryFileListFilter} would allow for - * this. See {@link org.springframework.integration.file.entries.CompositeEntryListFilter} for a way to do this. + * A common problem with reading files is that a file may be detected before it + * is ready. The default + * {@link org.springframework.integration.file.entries.AcceptOnceEntryFileListFilter} + * does not prevent this. In most cases, this can be prevented if the + * 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 org.springframework.integration.file.entries.AcceptOnceEntryFileListFilter} + * would allow for this. See + * {@link org.springframework.integration.file.entries.CompositeEntryListFilter} + * for a way to do 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 a {@link ResequencingMessageGroupProcessor}, but in cases where writing files and failure - * downstream are rare it might be sufficient. + * 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 + * a {@link ResequencingMessageGroupProcessor}, but in cases where writing files + * and failure downstream are rare it might be sufficient. *

- * FileReadingMessageSource is fully thread-safe under concurrent receive() invocations and message - * delivery callbacks. - * + * FileReadingMessageSource is fully thread-safe under concurrent + * receive() invocations and message delivery callbacks. + * * @author Iwein Fuld * @author Mark Fisher * @author Oleg Zhurakousky */ -public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource{ - private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5; - private static final Log logger = LogFactory.getLog(FileReadingMessageSource.class); - private volatile File directory; - private volatile DirectoryScanner scanner = new DefaultDirectoryScanner(); - private volatile boolean autoCreateDirectory = true; +public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource { - /* - * {@link PriorityBlockingQueue#iterator()} throws {@link java.util.ConcurrentModificationException} in Java 5. - * There is no locking around the queue, so there is also no iteration. - */ - private final Queue toBeReceived; - private boolean scanEachPoll = false; + private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5; - /** - * Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity. - */ - public FileReadingMessageSource() { - this(null); - } + private static final Log logger = LogFactory.getLog(FileReadingMessageSource.class); - /** - * 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 directory scanning. With scanEachPoll set to false - * and the queue to a large size, it will be filled once and then completely emptied - * before a new directory listing is done. This is particularly useful to reduce scans - * of large numbers of files in a directory. - */ - public FileReadingMessageSource(int internalQueueCapacity) { - this(null); - Assert.isTrue(internalQueueCapacity > 0, "Cannot create a queue with non positive capacity"); - this.setScanner(new HeadDirectoryScanner(internalQueueCapacity)); - } - /** - * Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue} ordered with the passed in {@link - * Comparator} - *

- * The size of the queue used should be large enough to hold all the files in the input directory in order to sort - * all of them, so restricting the 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 - */ - public FileReadingMessageSource(Comparator receptionOrderComparator) { - toBeReceived = new PriorityBlockingQueue(DEFAULT_INTERNAL_QUEUE_CAPACITY, receptionOrderComparator); - } + private volatile File directory; - /** - * Specify the input directory. - * - * @param directory to monitor - */ - public void setDirectory(File directory) { - Assert.notNull(directory, "directory must not be null"); - this.directory = directory; - } + private volatile DirectoryScanner scanner = new DefaultDirectoryScanner(); - /** - * Optionally specify a custom scanner, for example the {@link org.springframework.integration.file.RecursiveLeafOnlyDirectoryScanner} - * - * @param scanner scanner impl - */ - public void setScanner(DirectoryScanner scanner) { - this.scanner = scanner; - } + private volatile boolean autoCreateDirectory = true; - /** - * Specify whether to create the source directory automatically if it does not yet exist upon initialization. By - * default, this value is 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? - */ - public void setAutoCreateDirectory(boolean autoCreateDirectory) { - this.autoCreateDirectory = autoCreateDirectory; - } + /* + * {@link PriorityBlockingQueue#iterator()} throws + * {@link java.util.ConcurrentModificationException} in Java 5. + * There is no locking around the queue, so there is also no iteration. + */ + private final Queue toBeReceived; - /** - * Sets a {@link org.springframework.integration.file.entries.EntryListFilter}. By default a {@link org.springframework.integration.file.entries.AcceptOnceEntryFileListFilter} with no bounds is used. In most - * cases a customized {@link org.springframework.integration.file.entries.EntryListFilter} will be needed to deal with modification and duplication concerns. If - * multiple filters are required a {@link org.springframework.integration.file.entries.CompositeEntryListFilter} can be used to group them together. - *

- * The supplied filter must be thread safe.. - * - * @param filter a filter - */ - public void setFilter(EntryListFilter filter) { - Assert.notNull(filter, "'filter' must not be null"); - this.scanner.setFilter(filter); - } + private volatile boolean scanEachPoll = false; - /** - * Optional. Sets a {@link FileLocker} to be used to guard files - * against duplicate processing. - *

- * The supplied FileLocker must be thread safe - * - * @param locker a locker - */ - public void setLocker(FileLocker locker) { - Assert.notNull(locker, "'fileLocker' must not be null."); - this.scanner.setLocker(locker); - } - /** - * Optional. Set this flag if you want to make sure the internal queue is refreshed with the latest content of the - * input directory on each poll. - *

- * By default this implementation will empty its queue before looking at the directory again. In cases where order - * is relevant it is important to consider the effects of setting this flag. The internal {@link - * java.util.concurrent.BlockingQueue} that this class is keeping 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) - */ - public void setScanEachPoll(boolean scanEachPoll) { - this.scanEachPoll = scanEachPoll; - } + /** + * Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity. + */ + public FileReadingMessageSource() { + this(null); + } - protected void onInit() { - Assert.notNull(directory, "'directory' must not be set before initialization"); + /** + * 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 + * directory scanning. With scanEachPoll set to false and the + * queue to a large size, it will be filled once and then + * completely emptied before a new directory listing is done. + * This is particularly useful to reduce scans of large numbers + * of files in a directory. + */ + public FileReadingMessageSource(int internalQueueCapacity) { + this(null); + Assert.isTrue(internalQueueCapacity > 0, + "Cannot create a queue with non positive capacity"); + this.setScanner(new HeadDirectoryScanner(internalQueueCapacity)); + } - if (!this.directory.exists() && this.autoCreateDirectory) { - this.directory.mkdirs(); - } + /** + * Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue} + * ordered with the passed in {@link Comparator} + *

+ * The size of the queue used should be large enough to hold all the files + * in the input directory in order to sort all of them, so restricting the + * 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 + */ + public FileReadingMessageSource(Comparator receptionOrderComparator) { + this.toBeReceived = new PriorityBlockingQueue( + DEFAULT_INTERNAL_QUEUE_CAPACITY, receptionOrderComparator); + } - Assert.isTrue(this.directory.exists(), "Source directory [" + directory + "] does not exist."); - Assert.isTrue(this.directory.isDirectory(), "Source path [" + this.directory + "] does not point to a directory."); - Assert.isTrue(this.directory.canRead(), "Source directory [" + this.directory + "] is not readable."); - } - public Message receive() throws MessagingException { - Message message = null; + /** + * Specify the input directory. + * + * @param directory to monitor + */ + public void setDirectory(File directory) { + Assert.notNull(directory, "directory must not be null"); + this.directory = directory; + } - // rescan only if needed or explicitly configured - if (scanEachPoll || toBeReceived.isEmpty()) { - scanInputDirectory(); - } + /** + * Optionally specify a custom scanner, for example the + * {@link org.springframework.integration.file.RecursiveLeafOnlyDirectoryScanner} + * + * @param scanner scanner implementation + */ + public void setScanner(DirectoryScanner scanner) { + this.scanner = scanner; + } - File file = toBeReceived.poll(); + /** + * Specify whether to create the source directory automatically if it does + * not yet exist upon initialization. By default, this value is + * 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? + */ + public void setAutoCreateDirectory(boolean autoCreateDirectory) { + this.autoCreateDirectory = autoCreateDirectory; + } - // file == null means the queue was empty - // we can't rely on isEmpty for concurrency reasons - while ((file != null) && !scanner.tryClaim(file)) { - file = toBeReceived.poll(); - } + /** + * Sets a + * {@link org.springframework.integration.file.entries.EntryListFilter}. By + * default a + * {@link org.springframework.integration.file.entries.AcceptOnceEntryFileListFilter} + * with no bounds is used. In most cases a customized + * {@link org.springframework.integration.file.entries.EntryListFilter} will + * be needed to deal with modification and duplication concerns. If multiple + * filters are required a + * {@link org.springframework.integration.file.entries.CompositeEntryListFilter} + * can be used to group them together. + *

+ * The supplied filter must be thread safe.. + * + * @param filter a filter + */ + public void setFilter(EntryListFilter filter) { + Assert.notNull(filter, "'filter' must not be null"); + this.scanner.setFilter(filter); + } - if (file != null) { - message = MessageBuilder.withPayload(file).build(); + /** + * Optional. Sets a {@link FileLocker} to be used to guard files against + * duplicate processing. + *

+ * The supplied FileLocker must be thread safe + * + * @param locker a locker + */ + public void setLocker(FileLocker locker) { + Assert.notNull(locker, "'fileLocker' must not be null."); + this.scanner.setLocker(locker); + } - if (logger.isInfoEnabled()) { - logger.info("Created message: [" + message + "]"); - } - } - - return message; - } - - private void scanInputDirectory() { - List filteredFiles = scanner.listFiles(directory); - Set freshFiles = new HashSet(filteredFiles); - - if (!freshFiles.isEmpty()) { - toBeReceived.addAll(freshFiles); - - if (logger.isDebugEnabled()) { - logger.debug("Added to queue: " + freshFiles); - } - } - } - - /** - * Adds the failed message back to the 'toBeReceived' queue if there is room. - * - * @param failedMessage the {@link org.springframework.integration.Message} that blew up - */ - public void onFailure(Message failedMessage) { - if (logger.isWarnEnabled()) { - logger.warn("Failed to send: " + failedMessage); - } - - toBeReceived.offer(failedMessage.getPayload()); - } - - /** - * 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 - */ - public void onSend(Message sentMessage) { - if (logger.isDebugEnabled()) { - logger.debug("Sent: " + sentMessage); - } - } + /** + * Optional. Set this flag if you want to make sure the internal queue is + * refreshed with the latest content of the input directory on each poll. + *

+ * By default this implementation will empty its queue before looking at the + * directory again. In cases where order is relevant it is important to + * consider the effects of setting this flag. The internal + * {@link java.util.concurrent.BlockingQueue} that this class is keeping + * 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) + */ + public void setScanEachPoll(boolean scanEachPoll) { + this.scanEachPoll = scanEachPoll; + } public String getComponentType() { return "file:inbound-channel-adapter"; } + + protected void onInit() { + Assert.notNull(directory, "'directory' must not be null"); + if (!this.directory.exists() && this.autoCreateDirectory) { + this.directory.mkdirs(); + } + Assert.isTrue(this.directory.exists(), + "Source directory [" + directory + "] does not exist."); + Assert.isTrue(this.directory.isDirectory(), + "Source path [" + this.directory + "] does not point to a directory."); + Assert.isTrue(this.directory.canRead(), + "Source directory [" + this.directory + "] is not readable."); + } + + public Message receive() throws MessagingException { + Message message = null; + + // rescan only if needed or explicitly configured + if (scanEachPoll || toBeReceived.isEmpty()) { + scanInputDirectory(); + } + + File file = toBeReceived.poll(); + + // file == null means the queue was empty + // we can't rely on isEmpty for concurrency reasons + while ((file != null) && !scanner.tryClaim(file)) { + file = toBeReceived.poll(); + } + + if (file != null) { + message = MessageBuilder.withPayload(file).build(); + if (logger.isInfoEnabled()) { + logger.info("Created message: [" + message + "]"); + } + } + return message; + } + + private void scanInputDirectory() { + List filteredFiles = scanner.listFiles(directory); + Set freshFiles = new HashSet(filteredFiles); + if (!freshFiles.isEmpty()) { + toBeReceived.addAll(freshFiles); + if (logger.isDebugEnabled()) { + logger.debug("Added to queue: " + freshFiles); + } + } + } + + /** + * Adds the failed message back to the 'toBeReceived' queue if there is room. + * + * @param failedMessage + * the {@link org.springframework.integration.Message} that failed + */ + public void onFailure(Message failedMessage) { + if (logger.isWarnEnabled()) { + logger.warn("Failed to send: " + failedMessage); + } + toBeReceived.offer(failedMessage.getPayload()); + } + + /** + * 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 + */ + public void onSend(Message sentMessage) { + if (logger.isDebugEnabled()) { + logger.debug("Sent: " + sentMessage); + } + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index b89fe2f196..ba5478d040 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java @@ -204,9 +204,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand return resultFile; } - private File handleByteArrayMessage(byte[] bytes, File originalFile, File tempFile, File resultFile) - throws IOException { - + private File handleByteArrayMessage(byte[] bytes, File originalFile, File tempFile, File resultFile) throws IOException { FileCopyUtils.copy(bytes, tempFile); tempFile.renameTo(resultFile); if (this.deleteSourceFiles && originalFile != null) { @@ -215,11 +213,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand return resultFile; } - private File handleStringMessage(String content, File originalFile, File tempFile, File resultFile) - throws IOException { - - OutputStreamWriter writer = new OutputStreamWriter( - new FileOutputStream(tempFile), this.charset); + private File handleStringMessage(String content, File originalFile, File tempFile, File resultFile) throws IOException { + OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(tempFile), this.charset); FileCopyUtils.copy(content, writer); tempFile.renameTo(resultFile); if (this.deleteSourceFiles && originalFile != null) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java index 6a3c291b24..999a56df71 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.file; import org.springframework.integration.file.entries.EntryListFilter; @@ -22,28 +23,32 @@ import java.io.File; import java.util.Arrays; import java.util.List; - /** - * A custom scanner that only returns the first maxNumberOfFiles elements from a directory listing. This is - * useful to limit the number of File objects in memory and therefore mutually exclusive with AcceptOnceFileListFilter. - * + * A custom scanner that only returns the first maxNumberOfFiles + * elements from a directory listing. This is useful to limit the number of File + * objects in memory and therefore mutually exclusive with AcceptOnceFileListFilter. + * * @author Iwein Fuld - * @since 2.0.0 + * @since 2.0 */ public class HeadDirectoryScanner extends DefaultDirectoryScanner { - public HeadDirectoryScanner(int maxNumberOfFiles) { - this.setFilter(new HeadFilter(maxNumberOfFiles)); - } - private class HeadFilter implements EntryListFilter { - private final int maxNumberOfFiles; + public HeadDirectoryScanner(int maxNumberOfFiles) { + this.setFilter(new HeadFilter(maxNumberOfFiles)); + } - public HeadFilter(int maxNumberOfFiles) { - this.maxNumberOfFiles = maxNumberOfFiles; - } - public List filterEntries(File[] files) { - return Arrays.asList(files).subList(0, Math.min(files.length, maxNumberOfFiles)); - } - } + private static class HeadFilter implements EntryListFilter { + + private final int maxNumberOfFiles; + + public HeadFilter(int maxNumberOfFiles) { + this.maxNumberOfFiles = maxNumberOfFiles; + } + + public List filterEntries(File[] files) { + return Arrays.asList(files).subList(0, Math.min(files.length, maxNumberOfFiles)); + } + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java index f417e98993..de531f4d29 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java @@ -1,3 +1,19 @@ +/* + * Copyright 2002-2010 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.file; import java.io.File; @@ -6,23 +22,27 @@ import java.util.Arrays; import java.util.List; /** - * DirectoryScanner that lists all files inside a directory and subdirectories, without limit. This scanner should not - * be used with directories that contain a vast number of files or on deep trees, as all the file names will be read + * DirectoryScanner that lists all files inside a directory and subdirectories, + * without limit. This scanner should not be used with directories that contain + * a vast number of files or on deep trees, as all the file names will be read * into memory and the scanning will be done recursively. - * + * * @author Iwein Fuld */ public class RecursiveLeafOnlyDirectoryScanner extends DefaultDirectoryScanner { - protected File[] listEligibleFiles(File directory) throws IllegalArgumentException { - File[] rootFiles = directory.listFiles(); - List files = new ArrayList(rootFiles.length); - for (File rootFile : rootFiles) { - if (rootFile.isDirectory()) { - files.addAll(Arrays.asList(listEligibleFiles(rootFile))); - } else { - files.add(rootFile); - } - } - return files.toArray(new File[files.size()]); - } + + protected File[] listEligibleFiles(File directory) throws IllegalArgumentException { + File[] rootFiles = directory.listFiles(); + List files = new ArrayList(rootFiles.length); + for (File rootFile : rootFiles) { + if (rootFile.isDirectory()) { + files.addAll(Arrays.asList(listEligibleFiles(rootFile))); + } + else { + files.add(rootFile); + } + } + return files.toArray(new File[files.size()]); + } + } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/recursive/FileInboundChannelAdapterWithRecursiveDirectoryTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/recursive/FileInboundChannelAdapterWithRecursiveDirectoryTests.java index b8792d06fd..247c189654 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/recursive/FileInboundChannelAdapterWithRecursiveDirectoryTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/recursive/FileInboundChannelAdapterWithRecursiveDirectoryTests.java @@ -61,7 +61,7 @@ public class FileInboundChannelAdapterWithRecursiveDirectoryTests { } @SuppressWarnings("unchecked") - @Test(timeout = 2000) + @Test(timeout = 3000) public void shouldReturnFilesMultipleLevels() throws IOException { //when