Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

polishing

Browse Source
This commit is contained in:
Mark Fisher
2010-11-05 13:53:59 -04:00
parent 225a4a3322
commit 295af55242
11 changed files with 402 additions and 329 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
spring-integration-file/src/main/java/org/springframework/integration/file/DefaultDirectoryScanner.java
View File

@@ -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<File> filter = new AcceptOnceEntryFileListFilter<File>();
private FileLocker locker;
public final List<File> listFiles(File directory) throws IllegalArgumentException {
File[] files = listEligibleFiles(directory);
private volatile EntryListFilter<File> filter = new AcceptOnceEntryFileListFilter<File>();
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<File> filter) {
this.filter = filter;
}
public void setFilter(EntryListFilter<File> filter) {
this.filter = filter;
}
/**
* {@inheritDoc}
*/
public final void setLocker(FileLocker locker) {
this.locker = locker;
}
/**
* {@inheritDoc}
* <p/>
* 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}
* <p/>
* 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<File> 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();
}
}

5
spring-integration-file/src/main/java/org/springframework/integration/file/DefaultFileNameGenerator.java
View File

@@ -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) {

85
spring-integration-file/src/main/java/org/springframework/integration/file/DirectoryScanner.java
View File

@@ -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<File> 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<File> 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<File> 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<File> 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 <code>lock</code> method and
* MUST return <code>false<code> 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 <code>lock</code> method
* and MUST return <code>false<code> 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);
}

2
spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java
View File

@@ -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.

2
spring-integration-file/src/main/java/org/springframework/integration/file/FileLocker.java
View File

@@ -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.

2
spring-integration-file/src/main/java/org/springframework/integration/file/FileNameGenerator.java
View File

@@ -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.

440
spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
View File

@@ -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.
* <p/>
* 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.
* <p/>
* 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.
* <p/>
* FileReadingMessageSource is fully thread-safe under concurrent <code>receive()</code> invocations and message
* delivery callbacks.
*
* FileReadingMessageSource is fully thread-safe under concurrent
* <code>receive()</code> invocations and message delivery callbacks.
*
* @author Iwein Fuld
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource<File>{
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<File> {
/*
* {@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<File> 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}
* <p/>
* 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.
* <p/>
*
* @param receptionOrderComparator the comparator to be used to order the files in the internal queue
*/
public FileReadingMessageSource(Comparator<File> receptionOrderComparator) {
toBeReceived = new PriorityBlockingQueue<File>(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 <emphasis>true</emphasis>. If set to <emphasis>false</emphasis> 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<File> 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.
* <p/>
* <b>The supplied filter must be thread safe.</b>.
*
* @param filter a filter
*/
public void setFilter(EntryListFilter<File> 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.
* <p/>
* <b>The supplied FileLocker must be thread safe</b>
*
* @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.
* <p/>
* 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 <code>false</code>, but it will change more often (causing expensive reordering) if
* it is set to <code>true</code>.
*
* @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}
* <p/>
* 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.
* <p/>
*
* @param receptionOrderComparator
* the comparator to be used to order the files in the internal
* queue
*/
public FileReadingMessageSource(Comparator<File> receptionOrderComparator) {
this.toBeReceived = new PriorityBlockingQueue<File>(
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<File> receive() throws MessagingException {
Message<File> 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
* <emphasis>true</emphasis>. If set to <emphasis>false</emphasis> 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.
* <p/>
* <b>The supplied filter must be thread safe.</b>.
*
* @param filter a filter
*/
public void setFilter(EntryListFilter<File> 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.
* <p/>
* <b>The supplied FileLocker must be thread safe</b>
*
* @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<File> filteredFiles = scanner.listFiles(directory);
Set<File> freshFiles = new HashSet<File>(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<File> 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 <code>receive()</code>
*
* @param sentMessage the message that was successfully delivered
*/
public void onSend(Message<File> 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.
* <p/>
* 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 <code>false</code>, but it will change more often (causing expensive
* reordering) if it is set to <code>true</code>.
*
* @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<File> receive() throws MessagingException {
Message<File> 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<File> filteredFiles = scanner.listFiles(directory);
Set<File> freshFiles = new HashSet<File>(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<File> 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 <code>receive()</code>
*
* @param sentMessage
* the message that was successfully delivered
*/
public void onSend(Message<File> sentMessage) {
if (logger.isDebugEnabled()) {
logger.debug("Sent: " + sentMessage);
}
}
}

11
spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java
View File

@@ -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) {

39
spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java
View File

@@ -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 <code>maxNumberOfFiles</code> 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 <code>maxNumberOfFiles</code>
* 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<File> {
private final int maxNumberOfFiles;
public HeadDirectoryScanner(int maxNumberOfFiles) {
this.setFilter(new HeadFilter(maxNumberOfFiles));
}
public HeadFilter(int maxNumberOfFiles) {
this.maxNumberOfFiles = maxNumberOfFiles;
}
public List<File> filterEntries(File[] files) {
return Arrays.asList(files).subList(0, Math.min(files.length, maxNumberOfFiles));
}
}
private static class HeadFilter implements EntryListFilter<File> {
private final int maxNumberOfFiles;
public HeadFilter(int maxNumberOfFiles) {
this.maxNumberOfFiles = maxNumberOfFiles;
}
public List<File> filterEntries(File[] files) {
return Arrays.asList(files).subList(0, Math.min(files.length, maxNumberOfFiles));
}
}
}

50
spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java
View File

@@ -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<File> files = new ArrayList<File>(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<File> files = new ArrayList<File>(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()]);
}
}

2
spring-integration-file/src/test/java/org/springframework/integration/file/recursive/FileInboundChannelAdapterWithRecursiveDirectoryTests.java
View File

@@ -61,7 +61,7 @@ public class FileInboundChannelAdapterWithRecursiveDirectoryTests {
}
@SuppressWarnings("unchecked")
@Test(timeout = 2000)
@Test(timeout = 3000)
public void shouldReturnFilesMultipleLevels() throws IOException {
//when