From e6d1b2f9c39bd69ace74a8463cf36e51975c5239 Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Fri, 12 Feb 2010 19:31:10 +0000 Subject: [PATCH] INT-947: cleaning out some obsolete code, splitting locking and filtering a bit better. --- .../.classpath | 42 ++++---- .../file/CompositeFileListFilter.java | 95 +++++++++---------- .../file/FileReadingMessageSource.java | 21 +--- .../integration/file/NoopFileLocker.java | 38 -------- .../FileReadingMessageSourceFactoryBean.java | 10 +- ...ter.java => AbstractFileLockerFilter.java} | 6 +- .../integration/file/locking/FileLocker.java | 9 +- .../file/locking/NioFileLocker.java | 86 ++++++++--------- .../file/FileReadingMessageSourceTests.java | 3 +- .../locking/FileLockingNamespaceTests.java | 6 +- ...ultipleSourcesIntegrationTests-context.xml | 6 +- .../file/locking/NioFileLockerTests.java | 15 +-- .../CorrelatingMessageHandlerTests.java | 2 +- 13 files changed, 147 insertions(+), 192 deletions(-) delete mode 100644 org.springframework.integration.file/src/main/java/org/springframework/integration/file/NoopFileLocker.java rename org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/{AbstractLockingFilter.java => AbstractFileLockerFilter.java} (79%) 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();