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

Replaced FileFilter usage with FileListFilter

Browse Source
This commit is contained in:
Iwein Fuld
2008-09-11 16:25:05 +00:00
parent e17bb4f11a
commit 280da19191
9 changed files with 109 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
org.springframework.integration.file/src/main/java/org/springframework/integration/file/AcceptOnceFileFilter.java
View File

@@ -1,11 +1,12 @@
package org.springframework.integration.file;
import java.io.File;
import java.io.FileFilter;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
class AcceptOnceFileFilter implements FileFilter {
public class AcceptOnceFileFilter implements FileListFilter {
private final Queue<File> seen;
@@ -19,7 +20,17 @@ class AcceptOnceFileFilter implements FileFilter {
seen = new LinkedBlockingQueue<File>();
}
public boolean accept(File pathname) {
public List<File> filterFiles(File[] files) {
List<File> accepted = new ArrayList<File>();
for (File file : files) {
if (accept(file)) {
accepted.add(file);
}
}
return accepted;
}
private boolean accept(File pathname) {
synchronized (monitor) {
if (!seen.contains(pathname)) {
if (!seen.offer(pathname)) {

34
org.springframework.integration.file/src/main/java/org/springframework/integration/file/CompositeFileFilter.java
View File

@@ -20,37 +20,37 @@ import java.io.FileFilter;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
/**
* Composition that delegates to multiple {@link FileFilter}s. The compostition
* is AND based, meaning that all filters must {@link #accept(File)} in order
* for a file to be accepted by the composite.
* is AND based, meaning that all filters must {@link #filterFiles(File)} in
* order for a file to be accepted by the composite.
*
* @author Iwein Fuld
*/
class CompositeFileFilter implements FileFilter {
public class CompositeFileFilter implements FileListFilter {
private final Set<FileFilter> fileFilters;
private final Set<FileListFilter> fileFilters;
public CompositeFileFilter(FileFilter... fileFilters) {
this.fileFilters = new HashSet<FileFilter>(Arrays.asList(fileFilters));
public CompositeFileFilter(FileListFilter... fileFilters) {
this.fileFilters = new HashSet<FileListFilter>(Arrays.asList(fileFilters));
}
public CompositeFileFilter(HashSet<FileFilter> fileFilters) {
this.fileFilters = new HashSet<FileFilter>(fileFilters);
public CompositeFileFilter(Collection<FileListFilter> fileFilters) {
this.fileFilters = new HashSet<FileListFilter>(fileFilters);
}
/**
* {@inheritDoc}
*/
public boolean accept(File pathname) {
for (FileFilter fileFilter : fileFilters) {
if (!fileFilter.accept(pathname)) {
return false;
}
public List<File> filterFiles(File[] files) {
List<File> leftOver = Arrays.asList(files);
for (FileListFilter fileFilter : fileFilters) {
leftOver = fileFilter.filterFiles(leftOver.toArray(new File[] {}));
}
return true;
return leftOver;
}
/**
@@ -58,7 +58,7 @@ class CompositeFileFilter implements FileFilter {
* @param filters one or more new filters to be used
* @return a new CompositeFileFilter with the additional filters
*/
public CompositeFileFilter addFilter(FileFilter... filters) {
public CompositeFileFilter addFilter(FileListFilter... filters) {
return addFilters(Arrays.asList(filters));
}
@@ -69,8 +69,8 @@ class CompositeFileFilter implements FileFilter {
* @param filtersToAdd
* @return a new CompositeFileFilter with the added filters
*/
public CompositeFileFilter addFilters(Collection<FileFilter> filtersToAdd) {
HashSet<FileFilter> newFilterSet = new HashSet<FileFilter>(filtersToAdd);
public CompositeFileFilter addFilters(Collection<FileListFilter> filtersToAdd) {
HashSet<FileListFilter> newFilterSet = new HashSet<FileListFilter>(filtersToAdd);
newFilterSet.addAll(fileFilters);
return new CompositeFileFilter(newFilterSet);
}

10
org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileListFilter.java Normal file
View File

@@ -0,0 +1,10 @@
package org.springframework.integration.file;
import java.io.File;
import java.util.List;
public interface FileListFilter {
List<File> filterFiles(File[] files);
}

15
org.springframework.integration.file/src/main/java/org/springframework/integration/file/ModificationTimeFileFilter.java
View File

@@ -1,15 +0,0 @@
package org.springframework.integration.file;
import java.io.File;
import java.io.FileFilter;
class ModificationTimeFileFilter implements FileFilter {
private final long modificationTime;
public ModificationTimeFileFilter(long modificationTime) {
this.modificationTime = modificationTime;
}
public boolean accept(File file) {
return modificationTime <= file.lastModified();
}
}

69
org.springframework.integration.file/src/main/java/org/springframework/integration/file/PollableFileSource.java
View File

@@ -17,10 +17,10 @@ package org.springframework.integration.file;
import java.io.File;
import java.io.FileFilter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.PriorityBlockingQueue;
import org.apache.commons.logging.Log;
@@ -50,21 +50,13 @@ import org.springframework.util.Assert;
*/
public class PollableFileSource implements PollableSource<File>, MessageDeliveryAware<File>, InitializingBean {
private static final Log log = LogFactory.getLog(PollableFileSource.class);
private static final Log logger = LogFactory.getLog(PollableFileSource.class);
private volatile File inputDirectory;
private volatile Queue<File> fileQueue = new PriorityBlockingQueue<File>();
private final Queue<File> toBeReceived = new PriorityBlockingQueue<File>();
private volatile FileFilter filter = new AcceptOnceFileFilter();
/**
* Sets a queue to be used to hold files that are not processed yet. By
* default a {@link PriorityBlockingQueue} with natural ordering is used.
*/
public void setQueue(Queue<File> queue) {
this.fileQueue = queue;
}
private volatile FileListFilter filter = new AcceptOnceFileFilter();
public void setInputDirectory(File inputDirectory) {
this.inputDirectory = inputDirectory;
@@ -78,8 +70,8 @@ public class PollableFileSource implements PollableSource<File>, MessageDelivery
* {@link CompositeFileFilter} can be used to group them together <p/>
* <b>Note that the supplied filter must be thread safe</b>.
*/
public void setFilter(FileFilter filter) {
Assert.notNull(filter);
public void setFilter(FileListFilter filter) {
Assert.notNull(filter, "'filter' should not be null");
this.filter = filter;
}
@@ -89,23 +81,16 @@ public class PollableFileSource implements PollableSource<File>, MessageDelivery
Assert.isTrue(this.inputDirectory.canRead(), "No read permissions on " + inputDirectory);
}
/**
* {@inheritDoc}
*
* @return the Message created by the {@link #messageCreator} based on the
* next file from the {@link #fileQueue}. Existence of the file is not
* guaranteed, so the consumer of the message needs to check this.
*/
public Message<File> receive() throws MessagingException {
traceState();
refreshQueue();
Message<File> message = null;
File file = fileQueue.poll();
File file = toBeReceived.poll();
// we can't rely on isEmpty for concurrency reasons
if (file != null) {
message = new GenericMessage<File>(file);
if (log.isInfoEnabled()) {
log.info("Created message: [" + message + "]");
if (logger.isInfoEnabled()) {
logger.info("Created message: [" + message + "]");
}
}
traceState();
@@ -113,41 +98,33 @@ public class PollableFileSource implements PollableSource<File>, MessageDelivery
}
private void refreshQueue() {
List<File> freshFiles = new ArrayList<File>(processFileList(Arrays.asList(inputDirectory.listFiles(filter))));
List<File> filteredFiles = filter.filterFiles((inputDirectory.listFiles()));
Set<File> freshFiles = new HashSet<File>(filteredFiles);
if (!freshFiles.isEmpty()) {
// don't duplicate what's on the queue already
freshFiles.removeAll(fileQueue);
fileQueue.addAll(freshFiles);
if (log.isDebugEnabled()) {
log.debug("Added to queue: " + freshFiles);
freshFiles.removeAll(toBeReceived);
toBeReceived.addAll(freshFiles);
if (logger.isDebugEnabled()) {
logger.debug("Added to queue: " + freshFiles);
}
}
}
/**
* TODO point to FileFilter options
* @param files
* @return
*/
protected List<File> processFileList(List<File> files) {
return files;
}
/**
* In concurrent scenarios onFailure() might cause failing files to be
* ignored. If this is not acceptable access to this method should be
* synchronized on this instance externally.
*/
public void onFailure(Message<File> failedMessage, Throwable t) {
if (log.isWarnEnabled()) {
log.warn("Failed to send: " + failedMessage);
if (logger.isWarnEnabled()) {
logger.warn("Failed to send: " + failedMessage);
}
fileQueue.add(failedMessage.getPayload());
toBeReceived.add(failedMessage.getPayload());
}
public void onSend(Message<File> sentMessage) {
if (log.isDebugEnabled()) {
log.debug("Sent: " + sentMessage);
if (logger.isDebugEnabled()) {
logger.debug("Sent: " + sentMessage);
}
}
@@ -155,8 +132,8 @@ public class PollableFileSource implements PollableSource<File>, MessageDelivery
* utility method to trace the stateful collections of this instance.
*/
private void traceState() {
if (log.isTraceEnabled()) {
log.trace("Files to be received: [" + fileQueue + "]");
if (logger.isTraceEnabled()) {
logger.trace("Files to be received: [" + toBeReceived + "]");
}
}
}