diff --git a/org.springframework.integration.file/.classpath b/org.springframework.integration.file/.classpath
index 6a503bc291..af745b9da7 100644
--- a/org.springframework.integration.file/.classpath
+++ b/org.springframework.integration.file/.classpath
@@ -1,24 +1,24 @@
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/CompositeFileListFilter.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/CompositeFileListFilter.java
index 48fbf3f8af..a1ea42ee71 100644
--- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/CompositeFileListFilter.java
+++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/CompositeFileListFilter.java
@@ -16,71 +16,66 @@
package org.springframework.integration.file;
-import java.io.File;
-import java.io.FileFilter;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Set;
-
import org.springframework.util.Assert;
+import java.io.File;
+import java.io.FileFilter;
+import java.util.*;
+
/**
- * Composition that delegates to multiple {@link FileFilter}s. The composition
- * is AND based, meaning that a file must pass through each filter's
- * {@link #filterFiles(File)} method in order to be accepted by the composite.
- *
+ * Composition that delegates to multiple {@link FileFilter}s. The composition is AND based, meaning that a file must
+ * pass through each filter's {@link #filterFiles(java.io.File[])} method in order to be accepted by the composite.
+ *
* @author Iwein Fuld
* @author Mark Fisher
*/
public class CompositeFileListFilter implements FileListFilter {
- private final Set fileFilters;
+ private final Set fileFilters;
- public CompositeFileListFilter(FileListFilter... fileFilters) {
- this.fileFilters = new LinkedHashSet(Arrays.asList(fileFilters));
- }
+ public CompositeFileListFilter(FileListFilter... fileFilters) {
+ this.fileFilters = new LinkedHashSet(Arrays.asList(fileFilters));
+ }
- public CompositeFileListFilter(Collection fileFilters) {
- this.fileFilters = new LinkedHashSet(fileFilters);
- }
+ public CompositeFileListFilter(Collection fileFilters) {
+ this.fileFilters = new LinkedHashSet(fileFilters);
+ }
- /**
- * {@inheritDoc}
- *
- * This implementation delegates to a collection of filters and returns
- * only files that pass all the filters.
- */
- public List filterFiles(File[] files) {
- Assert.notNull(files, "'files' should not be null");
- List leftOver = Arrays.asList(files);
- for (FileListFilter fileFilter : this.fileFilters) {
- leftOver = fileFilter.filterFiles(leftOver.toArray(new File[] {}));
- }
- return leftOver;
- }
+ /**
+ * {@inheritDoc}
+ *
+ * This implementation delegates to a collection of filters and returns only files that pass all the filters.
+ */
+ public List filterFiles(File[] files) {
+ Assert.notNull(files, "'files' should not be null");
+ List leftOver = Arrays.asList(files);
+ for (FileListFilter fileFilter : this.fileFilters) {
+ leftOver = fileFilter.filterFiles(leftOver.toArray(new File[]{}));
+ }
+ return leftOver;
+ }
- /**
- * @see #addFilters(Collection)
- * @param filters one or more new filters to add
- * @return this CompositeFileFilter instance with the added filters
- */
- public CompositeFileListFilter addFilter(FileListFilter... filters) {
- return addFilters(Arrays.asList(filters));
- }
+ /**
+ * @param filters one or more new filters to add
+ * @return this CompositeFileFilter instance with the added filters
+ * @see #addFilters(Collection)
+ */
+ public CompositeFileListFilter addFilter(FileListFilter... filters) {
+ return addFilters(Arrays.asList(filters));
+ }
- /**
- * Add the new filters to this CompositeFileFilter while maintaining the
- * existing filters.
- *
- * @param filtersToAdd a list of filters to add
- * @return this CompositeFileFilter instance with the added filters
- */
- public CompositeFileListFilter addFilters(Collection filtersToAdd) {
- this.fileFilters.addAll(filtersToAdd);
+ /**
+ * Not thread safe. Only a single thread may add filters at a time.
+ *
+ * Add the new filters to this CompositeFileFilter while maintaining the existing filters.
+ *
+ * @param filtersToAdd a list of filters to add
+ * @return this CompositeFileFilter instance with the added filters
+ */
+ public CompositeFileListFilter addFilters(Collection filtersToAdd) {
+ this.fileFilters.addAll(filtersToAdd);
return this;
}
diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
index 5c07c1fd1b..b10d745eef 100644
--- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
+++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
@@ -143,9 +143,7 @@ public class FileReadingMessageSource implements MessageSource,
/**
* Optional. Sets a
* {@link org.springframework.integration.file.locking.FileLocker} to be
- * used instead of the default NoopFileLocker. Note that the locker is not
- * queried by this FileReadingMessageSource: integration with a
- * FileListFilter is an external concern.
+ * used to guard files against duplicate processing.
*
* The supplied FileLocker must be thread safe
*/
@@ -216,7 +214,7 @@ public class FileReadingMessageSource implements MessageSource,
}
/**
- * Adds the failed message back to the 'toBeReceived' queue.
+ * Adds the failed message back to the 'toBeReceived' queue if there is room.
*/
public void onFailure(Message failedMessage, Throwable t) {
if (logger.isWarnEnabled()) {
@@ -234,19 +232,4 @@ public class FileReadingMessageSource implements MessageSource,
logger.debug("Sent: " + sentMessage);
}
}
-
- /**
- * Implementation of FileLocker that doesn't provide any protection against
- * duplicate listing.
- */
- private static class NoopFileLocker implements FileLocker {
-
- public boolean lock(File fileToLock) {
- return true;
- }
-
- public void unlock(File fileToUnlock) {
- // noop
- }
- }
}
diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/NoopFileLocker.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/NoopFileLocker.java
deleted file mode 100644
index d8fc7fedbb..0000000000
--- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/NoopFileLocker.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Copyright 2002-2008 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 org.springframework.integration.file.locking.FileLocker;
-
-import java.io.File;
-
-/**
- * Implementation of FileLocker that doesn't provide any protection against duplicate listing. This is the default used
- * by the FileReadingMessageSource.
- *
- * @author Iwein Fuld
- */
-final class NoopFileLocker implements FileLocker {
-
- public boolean lock(File fileToLock) {
- return true;
- }
-
- public void unlock(File fileToUnlock) {
- //noop
- }
-}
-
diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java
index 773fb43eb1..aaa8da2dfb 100644
--- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java
+++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java
@@ -21,7 +21,7 @@ import org.springframework.integration.file.CompositeFileListFilter;
import org.springframework.integration.file.DirectoryScanner;
import org.springframework.integration.file.FileListFilter;
import org.springframework.integration.file.FileReadingMessageSource;
-import org.springframework.integration.file.locking.AbstractLockingFilter;
+import org.springframework.integration.file.locking.AbstractFileLockerFilter;
import java.io.File;
import java.util.Comparator;
@@ -39,7 +39,7 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean {
private volatile FileListFilter filter;
- private volatile AbstractLockingFilter locker;
+ private volatile AbstractFileLockerFilter locker;
private volatile Comparator comparator;
@@ -64,8 +64,8 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean {
}
public void setFilter(FileListFilter filter) {
- if (filter instanceof AbstractLockingFilter && this.locker == null) {
- this.setLocker((AbstractLockingFilter) filter);
+ if (filter instanceof AbstractFileLockerFilter && this.locker == null) {
+ this.setLocker((AbstractFileLockerFilter) filter);
}
this.filter = filter;
}
@@ -78,7 +78,7 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean {
this.autoCreateDirectory = autoCreateDirectory;
}
- public void setLocker(AbstractLockingFilter locker) {
+ public void setLocker(AbstractFileLockerFilter locker) {
this.locker = locker;
}
diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/AbstractLockingFilter.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/AbstractFileLockerFilter.java
similarity index 79%
rename from org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/AbstractLockingFilter.java
rename to org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/AbstractFileLockerFilter.java
index 50be386523..b42e5178bc 100644
--- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/AbstractLockingFilter.java
+++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/AbstractFileLockerFilter.java
@@ -21,16 +21,16 @@ import org.springframework.integration.file.AbstractFileListFilter;
import java.io.File;
/**
- * Convenience base class for implementing FileLockers that acquire a lock upon accepting a file. This is needed
+ * Convenience base class for implementing FileLockers that check a lock before accepting a file. This is needed
* when used in combination with a FileReadingMessageSource through a DirectoryScanner.
*
* @author Iwein Fuld
* @since 2.0
*
*/
-public abstract class AbstractLockingFilter extends AbstractFileListFilter implements FileLocker {
+public abstract class AbstractFileLockerFilter extends AbstractFileListFilter implements FileLocker {
protected final boolean accept(File file) {
- return lock(file);
+ return isLockable(file);
}
}
\ No newline at end of file
diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java
index 27ad326095..c0a12d0ac1 100644
--- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java
+++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java
@@ -25,7 +25,7 @@ import java.io.File;
* contract, defining these guarantees is up to the implementation.
*
* If a filter that respects locks is required extend
- * {@link org.springframework.integration.file.locking.AbstractLockingFilter} instead.
+ * {@link AbstractFileLockerFilter} instead.
*
* @author Iwein Fuld
* @since 2.0
@@ -40,6 +40,13 @@ public interface FileLocker {
*/
boolean lock(File fileToLock);
+ /**
+ * Checks whether the file passed in can be locked by this locker. This method never changes the locked state.
+ *
+ * @return true if the file was locked by another locker than this locker
+ */
+ boolean isLockable(File file);
+
/**
* Unlocks the given file.
*
diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java
index 951e4fabcc..707c1dd140 100644
--- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java
+++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java
@@ -26,56 +26,56 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
- * File locking strategy that uses java.nio. The locks taken by FileChannel are
- * shared with all the threads in a single JVM, so this locking strategy does
- * not prevent files being picked up multiple times within the same JVM.
- * {@link FileReadingMessageSource}s sharing a Locker will not pick up
- * the same files.
- *
- * This implementation will acquire or create a {@link FileLock} for the given
- * file. Caching locks might be expensive, so this locking strategy is not
- * recommended for scenarios where many files are accessed in parallel.
- *
+ * File locking strategy that uses java.nio. The locks taken by FileChannel are shared with all the threads in a single
+ * JVM, so this locking strategy does not prevent files being picked up multiple times within the same JVM.
+ * {@link FileReadingMessageSource}s sharing a Locker will not pick up the same files.
+ *
+ * This implementation will acquire or create a {@link FileLock} for the given file. Caching locks might be expensive,
+ * so this locking strategy is not recommended for scenarios where many files are accessed in parallel.
+ *
* @author Iwein Fuld
* @author Mark Fisher
* @since 2.0
*/
-public class NioFileLocker extends AbstractLockingFilter {
+public class NioFileLocker extends AbstractFileLockerFilter {
- private final ConcurrentMap lockCache = new ConcurrentHashMap();
+ private final ConcurrentMap lockCache = new ConcurrentHashMap();
/**
- * {@inheritDoc}
- *
- */
- public boolean lock(File fileToLock) {
- FileLock lock = lockCache.get(fileToLock);
- if (lock == null) {
- FileLock newLock = null;
- try {
- newLock = FileChannelCache.tryLockFor(fileToLock);
- } catch (IOException e) {
- throw new MessagingException("Failed to lock file: "
- + fileToLock, e);
- }
- if (newLock != null) {
- FileLock original = lockCache.putIfAbsent(fileToLock, newLock);
- lock = original != null ? original : newLock;
- }
- }
- return lock != null;
- }
+ * {@inheritDoc}
+ */
+ public boolean lock(File fileToLock) {
+ FileLock lock = lockCache.get(fileToLock);
+ if (lock == null) {
+ FileLock newLock = null;
+ try {
+ newLock = FileChannelCache.tryLockFor(fileToLock);
+ } catch (IOException e) {
+ throw new MessagingException("Failed to lock file: "
+ + fileToLock, e);
+ }
+ if (newLock != null) {
+ FileLock original = lockCache.putIfAbsent(fileToLock, newLock);
+ lock = original != null ? original : newLock;
+ }
+ }
+ return lock != null;
+ }
- public void unlock(File fileToUnlock) {
- FileLock fileLock = lockCache.get(fileToUnlock);
- try {
- if (fileLock != null) {
- fileLock.release();
- }
- FileChannelCache.closeChannelFor(fileToUnlock);
- } catch (IOException e) {
- throw new MessagingException("Failed to unlock file: "
- + fileToUnlock, e);
- }
+ public boolean isLockable(File file) {
+ return lockCache.containsKey(file) || !FileChannelCache.isLocked(file);
+ }
+
+ public void unlock(File fileToUnlock) {
+ FileLock fileLock = lockCache.get(fileToUnlock);
+ try {
+ if (fileLock != null) {
+ fileLock.release();
+ }
+ FileChannelCache.closeChannelFor(fileToUnlock);
+ } catch (IOException e) {
+ throw new MessagingException("Failed to unlock file: "
+ + fileToUnlock, e);
+ }
}
}
diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java
index d878395a6f..350fa3d2aa 100644
--- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java
+++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java
@@ -28,6 +28,7 @@ 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.*;
@@ -70,7 +71,7 @@ public class FileReadingMessageSourceTests {
@Test
public void straightProcess() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
- source.onSend(source.receive());
+ assertThat(source.receive().getPayload(), is(fileMock));
}
@Test
diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingNamespaceTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingNamespaceTests.java
index 2cab7404bb..28f74dd6bd 100644
--- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingNamespaceTests.java
+++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingNamespaceTests.java
@@ -77,11 +77,15 @@ public class FileLockingNamespaceTests {
assertThat(extractFromScanner("filter", nioLockingSource), is(CompositeFileListFilter.class));
}
- public static class StubLocker extends AbstractLockingFilter {
+ public static class StubLocker extends AbstractFileLockerFilter {
public boolean lock(File fileToLock) {
return true;
}
+ public boolean isLockable(File file) {
+ return true;
+ }
+
public void unlock(File fileToUnlock) {
//
}
diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml
index 0a36dc2775..69e1e6024c 100644
--- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml
+++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml
@@ -8,15 +8,15 @@
+ p:filter-ref="filter1" p:locker-ref="filter1"/>
+ p:filter-ref="filter2" p:locker-ref="filter2"/>
+ p:filter-ref="filter1" p:locker-ref="filter1"/>
diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/NioFileLockerTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/NioFileLockerTests.java
index 0408959136..5f4ab31acf 100644
--- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/NioFileLockerTests.java
+++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/NioFileLockerTests.java
@@ -15,11 +15,9 @@
*/
package org.springframework.integration.file.locking;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
import org.junit.Before;
-import org.junit.Test;
import org.junit.BeforeClass;
+import org.junit.Test;
import org.springframework.integration.file.FileListFilter;
import java.io.File;
@@ -27,6 +25,9 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
/**
* @author Iwein Fuld
*/
@@ -49,20 +50,22 @@ public class NioFileLockerTests {
@Test
public void fileListedByFirstFilter() throws IOException {
- FileListFilter filter = new NioFileLocker();
+ NioFileLocker filter = new NioFileLocker();
File testFile = new File(workdir, "test0");
testFile.createNewFile();
assertThat(filter.filterFiles(workdir.listFiles()).get(0), is(testFile));
+ filter.lock(testFile);
assertThat(filter.filterFiles(workdir.listFiles()).get(0), is(testFile));
}
@Test
- public void fileListedByOneFilterOnly() throws IOException {
- FileListFilter filter1 = new NioFileLocker();
+ public void fileNotListedWhenLockedByOtherFilter() throws IOException {
+ NioFileLocker filter1 = new NioFileLocker();
FileListFilter filter2 = new NioFileLocker();
File testFile = new File(workdir, "test1");
testFile.createNewFile();
assertThat(filter1.filterFiles(workdir.listFiles()).get(0), is(testFile));
+ filter1.lock(testFile);
assertThat(filter2.filterFiles(workdir.listFiles()), is((List)new ArrayList()));
}
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
index 27ac2d4de4..b30014798d 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
@@ -149,7 +149,7 @@ public class CorrelatingMessageHandlerTests {
}
});
- Thread.sleep(10);
+ Thread.sleep(20);
assertFalse(handler.forceComplete("key"));
bothMessagesHandled.await();