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

INT-768: added LockFileFileListFilter and friends

Browse Source
This commit is contained in:
Iwein Fuld
2009-09-19 12:04:09 +00:00
parent 3f8072a683
commit b2f443d676
11 changed files with 675 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files

347
org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
View File

@@ -16,27 +16,23 @@
package org.springframework.integration.file;
import java.io.File;
import java.io.IOException;
import java.util.Comparator;
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;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.Resource;
import org.springframework.integration.aggregator.Resequencer;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.file.locking.FileLocker;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageSource;
import org.springframework.util.Assert;
import java.io.File;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.PriorityBlockingQueue;
/**
* {@link MessageSource} that creates messages from a file system directory. To
* prevent messages for certain files, you may supply a {@link FileListFilter}.
@@ -59,177 +55,208 @@ import org.springframework.util.Assert;
* <p/>
* FileReadingMessageSource is fully thread-safe under concurrent
* <code>receive()</code> invocations and message delivery callbacks.
*
*
* @author Iwein Fuld
* @author Mark Fisher
*/
public class FileReadingMessageSource implements MessageSource<File>,
InitializingBean {
InitializingBean {
private static final int INTERNAL_QUEUE_CAPACITY = 5;
private static final int INTERNAL_QUEUE_CAPACITY = 5;
private static final Log logger = LogFactory
.getLog(FileReadingMessageSource.class);
private static final Log logger = LogFactory
.getLog(FileReadingMessageSource.class);
private volatile File inputDirectory;
private volatile File inputDirectory;
private volatile boolean autoCreateDirectory = true;
private volatile boolean autoCreateDirectory = true;
/**
* {@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;
/**
* {@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 volatile FileListFilter filter = new AcceptOnceFileListFilter();
private volatile FileListFilter filter = new AcceptOnceFileListFilter();
private boolean scanEachPoll = false;
private volatile FileLocker locker = new NoopFileLocker();
/**
* Creates a FileReadingMessageSource with a naturally ordered queue.
*/
public FileReadingMessageSource() {
toBeReceived = new PriorityBlockingQueue<File>(INTERNAL_QUEUE_CAPACITY);
}
private boolean scanEachPoll = false;
/**
* Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue}
* ordered with the passed in {@link Comparator}
*
* No guarantees about file delivery order can be made under concurrent
* access.
*/
public FileReadingMessageSource(Comparator<File> receptionOrderComparator) {
toBeReceived = new PriorityBlockingQueue<File>(INTERNAL_QUEUE_CAPACITY,
receptionOrderComparator);
}
/**
* Creates a FileReadingMessageSource with a naturally ordered queue.
*/
public FileReadingMessageSource() {
toBeReceived = new PriorityBlockingQueue<File>(INTERNAL_QUEUE_CAPACITY);
}
/**
* Specify the input directory.
*/
public void setInputDirectory(Resource inputDirectory) {
Assert.notNull(inputDirectory, "inputDirectory must not be null");
try {
this.inputDirectory = inputDirectory.getFile();
} catch (IOException ioe) {
try {
// fallback to the URI
this.inputDirectory = new File(inputDirectory.getURI());
} catch (Exception e) {
throw new IllegalArgumentException(
"Unexpected IOException when looking for source directory: "
+ inputDirectory, ioe);
}
}
}
/**
* Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue}
* ordered with the passed in {@link Comparator}
* <p/>
* No guarantees about file delivery order can be made under concurrent
* access.
*/
public FileReadingMessageSource(Comparator<File> receptionOrderComparator) {
toBeReceived = new PriorityBlockingQueue<File>(INTERNAL_QUEUE_CAPACITY,
receptionOrderComparator);
}
/**
* 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.
*/
public void setAutoCreateDirectory(boolean autoCreateDirectory) {
this.autoCreateDirectory = autoCreateDirectory;
}
/**
* Specify the input directory.
*/
public void setInputDirectory(Resource inputDirectory) {
Assert.notNull(inputDirectory, "inputDirectory must not be null");
try {
this.inputDirectory = inputDirectory.getFile();
} catch (IOException ioe) {
try {
// fallback to the URI
this.inputDirectory = new File(inputDirectory.getURI());
} catch (Exception e) {
throw new IllegalArgumentException(
"Unexpected IOException when looking for source directory: "
+ inputDirectory, ioe);
}
}
}
/**
* Sets a {@link FileListFilter}. By default a
* {@link AcceptOnceFileListFilter} with no bounds is used. In most cases a
* customized {@link FileListFilter} will be needed to deal with
* modification and duplication concerns. If multiple filters are required a
* {@link CompositeFileListFilter} can be used to group them together.
* <p/>
* <b>The supplied filter must be thread safe.</b>.
*/
public void setFilter(FileListFilter filter) {
Assert.notNull(filter, "'filter' should not be null");
this.filter = filter;
}
/**
* 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.
*/
public void setAutoCreateDirectory(boolean autoCreateDirectory) {
this.autoCreateDirectory = autoCreateDirectory;
}
/**
* 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 PriorityBlockingQueue} that this class is keeping will more likely
* be out of sync with the filesystem if this flag is set to
* <code>false</code>, but it will change more often (causing reordering) if
* it is set to <code>true</code>.
*/
public void setScanEachPoll(boolean scanEachPoll) {
this.scanEachPoll = scanEachPoll;
}
/**
* Sets a {@link FileListFilter}. By default a
* {@link AcceptOnceFileListFilter} with no bounds is used. In most cases a
* customized {@link FileListFilter} will be needed to deal with
* modification and duplication concerns. If multiple filters are required a
* {@link CompositeFileListFilter} can be used to group them together.
* <p/>
* <b>The supplied filter must be thread safe.</b>.
*/
public void setFilter(FileListFilter filter) {
Assert.notNull(filter, "'filter' must not be null");
this.filter = filter;
}
public final void afterPropertiesSet() {
if (!this.inputDirectory.exists() && this.autoCreateDirectory) {
this.inputDirectory.mkdirs();
}
Assert.isTrue(this.inputDirectory.exists(), "Source directory ["
+ inputDirectory + "] does not exist.");
Assert.isTrue(this.inputDirectory.isDirectory(), "Source path ["
+ this.inputDirectory + "] does not point to a directory.");
Assert.isTrue(this.inputDirectory.canRead(), "Source directory ["
+ this.inputDirectory + "] is not readable.");
}
/**
* 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.
* <p/>
* <b>The supplied FileLocker must be thread safe</b>
*/
public void setLocker(FileLocker locker) {
Assert.notNull(locker, "'fileLocker' must not be null.");
this.locker = locker;
}
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();
// we can't rely on isEmpty for concurrency reasons
if (file != null) {
message = MessageBuilder.withPayload(file).build();
if (logger.isInfoEnabled()) {
logger.info("Created message: [" + message + "]");
}
}
return message;
}
/**
* 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 PriorityBlockingQueue} that this class is keeping will more likely
* be out of sync with the filesystem if this flag is set to
* <code>false</code>, but it will change more often (causing reordering) if
* it is set to <code>true</code>.
*/
public void setScanEachPoll(boolean scanEachPoll) {
this.scanEachPoll = scanEachPoll;
}
private void scanInputDirectory() {
File[] fileArray = inputDirectory.listFiles();
if (fileArray == null) {
throw new MessagingException(
"Either the path ["
+ this.inputDirectory
+ "] does not denote a directory, or an I/O error has occured.");
}
List<File> filteredFiles = this.filter.filterFiles(fileArray);
Set<File> freshFiles = new HashSet<File>(filteredFiles);
if (!freshFiles.isEmpty()) {
toBeReceived.addAll(freshFiles);
if (logger.isDebugEnabled()) {
logger.debug("Added to queue: " + freshFiles);
}
}
}
public final void afterPropertiesSet() {
if (!this.inputDirectory.exists() && this.autoCreateDirectory) {
this.inputDirectory.mkdirs();
}
Assert.isTrue(this.inputDirectory.exists(), "Source directory ["
+ inputDirectory + "] does not exist.");
Assert.isTrue(this.inputDirectory.isDirectory(), "Source path ["
+ this.inputDirectory + "] does not point to a directory.");
Assert.isTrue(this.inputDirectory.canRead(), "Source directory ["
+ this.inputDirectory + "] is not readable.");
}
/**
* Adds the failed message back to the 'toBeReceived' queue.
*/
public void onFailure(Message<File> failedMessage, Throwable t) {
if (logger.isWarnEnabled()) {
logger.warn("Failed to send: " + failedMessage);
}
toBeReceived.offer(failedMessage.getPayload());
}
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 && !locker.lock(file)) {
file = toBeReceived.poll();
}
if (file != null) {
message = MessageBuilder.withPayload(file).build();
if (logger.isInfoEnabled()) {
logger.info("Created message: [" + message + "]");
}
}
return message;
}
/**
* The message is just logged. It was already removed from the queue during
* the call to <code>receive()</code>
*/
public void onSend(Message<File> sentMessage) {
if (logger.isDebugEnabled()) {
logger.debug("Sent: " + sentMessage);
}
}
private void scanInputDirectory() {
File[] fileArray = inputDirectory.listFiles();
if (fileArray == null) {
throw new MessagingException(
"The path ["
+ this.inputDirectory
+ "] does not denote a properly accessible directory.");
}
List<File> filteredFiles = this.filter.filterFiles(fileArray);
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.
*/
public void onFailure(Message<File> failedMessage, Throwable t) {
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>
*/
public void onSend(Message<File> sentMessage) {
if (logger.isDebugEnabled()) {
logger.debug("Sent: " + sentMessage);
}
}
/**
* Implementation of FileLocker that doesn't provide any protection against duplicate listing.
*/
class NoopFileLocker implements FileLocker {
public boolean lock(File fileToLock) {
return true;
}
public void unlock(File fileToUnlock) {
//noop
}
}
}

38
org.springframework.integration.file/src/main/java/org/springframework/integration/file/NoopFileLocker.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* 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
}
}

38
org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* 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.locking;
import java.io.File;
/**
* A FileLocker is a strategy that can ensure that files are only processed a single time.
*
* Implementations are free implement any relation between locking and unlocking (e.g. the Noop
*
* @author Iwein Fuld
*/
public interface FileLocker {
/**
* Tries to lock the given file and returns <code>true</code> if it was successful, <code>false</code> otherwise.
*/
boolean lock(File fileToLock);
/**
* Unlocks the given file.
*/
void unlock(File fileToUnlock);
}

52
org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileListFilter.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* 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.locking;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.file.AbstractFileListFilter;
import static org.springframework.integration.file.locking.LockFileFileLocker.*;
import java.io.File;
/**
* LockFileFileListFilter keeps track of the files it should pick up by adding lock files in a directory. If this
* directory is the same as the directory of the original files addition precautions need to be taken to avoid treating
* the lock files as normal input.
*
* If different LockFileFileListFilters share their lock directory they are guaranteed not to pass the same file.
*
* @author Iwein Fuld
*/
public class LockFileFileListFilter extends AbstractFileListFilter {
private static final Log logger = LogFactory.getLog(LockFileFileListFilter.class);
private File workdir;
public LockFileFileListFilter(File workdir) {
this.workdir = workdir;
}
protected boolean accept(File file) {
File lockFile = new File(workdir, file.getName()+ LOCK_SUFFIX);
File preLockFile = new File(workdir, file.getName()+ PRELOCK_SUFFIX);
if (lockFile.exists()||preLockFile.exists()){
return false;
}
String name = file.getName();
return !name.endsWith(LOCK_SUFFIX)&&!name.endsWith(PRELOCK_SUFFIX);
}
}

79
org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileLocker.java Normal file
View File

@@ -0,0 +1,79 @@
/*
* 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.locking;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.util.Assert;
import java.io.File;
import java.io.IOException;
/**
* @author Iwein Fuld
*/
public class LockFileFileLocker implements FileLocker {
private static final Log logger = LogFactory.getLog(LockFileFileLocker.class);
private File workdir;
static final String LOCK_SUFFIX = ".lock";
static final String PRELOCK_SUFFIX = ".prelock";
public LockFileFileLocker(File workdir) {
Assert.notNull(workdir, "Work directy must not be null.");
Assert.isTrue(workdir.isDirectory(),"Work directy must be a directory.");
Assert.isTrue(workdir.canWrite(), "Work directy must be write accessible.");
this.workdir = workdir;
}
/**
* Makes a best effort attempt at locking the file atomically. The chances of success are wholly dependant on the
* underlying operating system.
*
* The locking mechanism will create a prelock file, remove write permissions from that file and move it to a lock
* file location.
*/
public boolean lock(File fileToLock) {
File lockFile = new File(workdir, fileToLock.getName() + LOCK_SUFFIX);
File preLockFile = new File(workdir, fileToLock.getName() + PRELOCK_SUFFIX);
if (lockFile.exists() || preLockFile.exists()) {
return false;
}
try {
preLockFile.createNewFile();
} catch (IOException e) {
logger.warn("Failed to lock file", e);
return false;
}
preLockFile.setWritable(false);
if (preLockFile.renameTo(lockFile)) {
return true;
}
preLockFile.delete();
logger.warn("Failed to lock file [" + fileToLock + "]");
return false;
}
/**
* Sets the lockFile for this file to be writable and then deletes it. If a lock happens concurrently, the delete
* should not succeed, so the lock will remain in place. This might be different on various operating systems.
*/
public void unlock(File fileToUnlock) {
File lockFile = new File(workdir, fileToUnlock.getName() + LOCK_SUFFIX);
lockFile.setWritable(true);
lockFile.delete();
}
}

192
org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java
View File

@@ -16,22 +16,19 @@
package org.springframework.integration.file;
import static org.mockito.Mockito.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import java.io.File;
import java.util.Comparator;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import static org.mockito.Mockito.*;
import org.mockito.runners.MockitoJUnit44Runner;
import org.springframework.core.io.Resource;
import org.springframework.integration.core.Message;
import org.springframework.integration.file.locking.FileLocker;
import java.io.File;
import java.util.Comparator;
/**
* @author Iwein Fuld
@@ -41,94 +38,119 @@ import org.springframework.integration.core.Message;
@RunWith(MockitoJUnit44Runner.class)
public class FileReadingMessageSourceTests {
private FileReadingMessageSource source;
private FileReadingMessageSource source;
@Mock
private File inputDirectoryMock;
@Mock
private File inputDirectoryMock;
@Mock
private Resource inputDirectoryResourceMock;
@Mock
private Resource inputDirectoryResourceMock;
@Mock
private File fileMock;
@Mock
private File fileMock;
@Mock
private Comparator<File> comparator;
@Mock
private FileLocker locker;
public void prepResource() throws Exception {
when(inputDirectoryResourceMock.exists()).thenReturn(true);
when(inputDirectoryResourceMock.getFile()).thenReturn(inputDirectoryMock);
when(inputDirectoryMock.canRead()).thenReturn(true);
}
@Mock
private Comparator<File> comparator;
@Before
public void initialize() throws Exception {
prepResource();
this.source = new FileReadingMessageSource(comparator);
source.setInputDirectory(inputDirectoryResourceMock);
}
public void prepResource() throws Exception {
when(inputDirectoryResourceMock.exists()).thenReturn(true);
when(inputDirectoryResourceMock.getFile()).thenReturn(inputDirectoryMock);
when(inputDirectoryMock.canRead()).thenReturn(true);
when(locker.lock(isA(File.class))).thenReturn(true);
}
@Test
public void straightProcess() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock });
source.onSend(source.receive());
}
@Before
public void initialize() throws Exception {
prepResource();
this.source = new FileReadingMessageSource(comparator);
source.setInputDirectory(inputDirectoryResourceMock);
source.setLocker(locker);
}
@Test
public void requeueOnFailure() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock });
Message received = source.receive();
assertNotNull(received);
source.onFailure(received, new RuntimeException("failed"));
assertEquals(received.getPayload(), source.receive().getPayload());
verify(inputDirectoryMock,times(1)).listFiles();
}
@Test
public void straightProcess() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
source.onSend(source.receive());
}
@Test
public void scanEachPoll() throws Exception {
File anotherFileMock = mock(File.class);
when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock, anotherFileMock });
source.setScanEachPoll(true);
assertNotNull(source.receive());
assertNotNull(source.receive());
assertNull(source.receive());
verify(inputDirectoryMock,times(3)).listFiles();
}
@Test
public void requeueOnFailure() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
Message received = source.receive();
assertNotNull(received);
source.onFailure(received, new RuntimeException("failed"));
assertEquals(received.getPayload(), source.receive().getPayload());
verify(inputDirectoryMock, times(1)).listFiles();
}
@Test
public void noDuplication() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock });
Message<File> received = source.receive();
assertNotNull(received);
assertEquals(fileMock, received.getPayload());
assertNull(source.receive());
verify(inputDirectoryMock,times(2)).listFiles();
}
@Test
public void scanEachPoll() throws Exception {
File anotherFileMock = mock(File.class);
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock, anotherFileMock});
source.setScanEachPoll(true);
assertNotNull(source.receive());
assertNotNull(source.receive());
assertNull(source.receive());
verify(inputDirectoryMock, times(3)).listFiles();
}
@Test(expected = IllegalArgumentException.class)
public void nullFilter() throws Exception {
source.setFilter(null);
}
@Test
public void noDuplication() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
Message<File> received = source.receive();
assertNotNull(received);
assertEquals(fileMock, received.getPayload());
assertNull(source.receive());
verify(inputDirectoryMock, times(2)).listFiles();
}
@Test
public void orderedReception() throws Exception {
File file1 = mock(File.class);
File file2 = mock(File.class);
File file3 = mock(File.class);
@Test(expected = IllegalArgumentException.class)
public void nullFilter() throws Exception {
source.setFilter(null);
}
// record the comparator to reverse order the files
when(comparator.compare(file1, file2)).thenReturn(1);
when(comparator.compare(file1, file3)).thenReturn(1);
when(comparator.compare(file2, file3)).thenReturn(1);
when(comparator.compare(file2, file1)).thenReturn(-1);
when(comparator.compare(file3, file1)).thenReturn(-1);
when(comparator.compare(file3, file2)).thenReturn(-1);
@Test
public void lockIsAquired() {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
Message<File> received = source.receive();
assertNotNull(received);
assertEquals(fileMock, received.getPayload());
verify(locker).lock(fileMock);
}
when(inputDirectoryMock.listFiles()).thenReturn(new File[] { file2, file3, file1 });
assertSame(file3, source.receive().getPayload());
assertSame(file2, source.receive().getPayload());
assertSame(file1, source.receive().getPayload());
assertNull(source.receive());
verify(inputDirectoryMock,times(2)).listFiles();
}
@Test
public void lockedFilesAreIgnored() {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
when(locker.lock(fileMock)).thenReturn(false);
Message<File> received = source.receive();
assertNull(received);
verify(locker).lock(fileMock);
}
@Test
public void orderedReception() throws Exception {
File file1 = mock(File.class);
File file2 = mock(File.class);
File file3 = mock(File.class);
// record the comparator to reverse order the files
when(comparator.compare(file1, file2)).thenReturn(1);
when(comparator.compare(file1, file3)).thenReturn(1);
when(comparator.compare(file2, file3)).thenReturn(1);
when(comparator.compare(file2, file1)).thenReturn(-1);
when(comparator.compare(file3, file1)).thenReturn(-1);
when(comparator.compare(file3, file2)).thenReturn(-1);
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{file2, file3, file1});
assertSame(file3, source.receive().getPayload());
assertSame(file2, source.receive().getPayload());
assertSame(file1, source.receive().getPayload());
assertNull(source.receive());
verify(inputDirectoryMock, times(2)).listFiles();
}
}

26
org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml Normal file
View File

@@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">
<!-- under test -->
<bean id="fileSource1" class="org.springframework.integration.file.FileReadingMessageSource"
p:inputDirectory="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"
p:locker-ref="locker"/>
<bean id="fileSource2" class="org.springframework.integration.file.FileReadingMessageSource"
p:inputDirectory="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"
p:locker-ref="locker" p:filter-ref="filter"/>
<bean id="locker" class="org.springframework.integration.file.locking.LockFileFileLocker">
<constructor-arg value="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"/>
</bean>
<bean id="filter" class="org.springframework.integration.file.locking.LockFileFileListFilter">
<constructor-arg value="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"/>
</bean>
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
</beans>

76
org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests.java Normal file
View File

@@ -0,0 +1,76 @@
/*
* 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.locking;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.test.matcher.PayloadMatcher;
import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload;
import org.junit.runner.RunWith;
import org.junit.Test;
import org.junit.BeforeClass;
import org.junit.Before;
import org.junit.Assert;
import org.junit.matchers.JUnitMatchers;
import static org.junit.Assert.assertThat;
import org.hamcrest.CoreMatchers;
import static org.hamcrest.CoreMatchers.nullValue;
import java.io.File;
import java.io.IOException;
/**
* @author Iwein Fuld
*/
@ContextConfiguration
@RunWith(org.springframework.test.context.junit4.SpringJUnit4ClassRunner.class)
public class FileLockingWithMultipleSourcesIntegrationTests {
private static File workdir;
@BeforeClass
public static void setupWorkDirectory() throws Exception {
workdir = new File(
new File(System.getProperty("java.io.tmpdir")),
FileLockingWithMultipleSourcesIntegrationTests.class.getSimpleName()
);
workdir.mkdir();
}
@Autowired
@Qualifier("fileSource1")
private FileReadingMessageSource fileSource1;
@Autowired
@Qualifier("fileSource2")
private FileReadingMessageSource fileSource2;
@Before
public void cleanoutWorkDir() {
for (File file : workdir.listFiles()) {
file.delete();
}
}
@Test
public void filePickedUpOnlyOnce() throws IOException {
File testFile = new File(workdir, "test");
testFile.createNewFile();
assertThat(fileSource1.receive(), hasPayload(testFile));
assertThat(fileSource2.receive(), nullValue());
}
}

70
org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/LockFileFileListFilterTests.java Normal file
View File

@@ -0,0 +1,70 @@
/*
* 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.locking;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
import org.junit.Before;
import org.junit.Test;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
/**
* @author Iwein Fuld
*/
public class LockFileFileListFilterTests {
private File workdir = new File(new File(System.getProperty("java.io.tmpdir")), this.getClass().getSimpleName());;
private FileLocker locker = new LockFileFileLocker(workdir);
@Before
public void setupWorkDir() {
workdir.mkdir();
cleanDirectory(workdir);
}
@Test
public void fileListedOnlyWhenNotLocked() throws IOException {
LockFileFileListFilter filter = new LockFileFileListFilter(workdir);
File testFile = new File(workdir, "test0");
testFile.createNewFile();
assertThat(filter.filterFiles(workdir.listFiles()).get(0), is(testFile));
locker.lock(testFile);
assertThat(filter.filterFiles(workdir.listFiles()), is((List)new ArrayList<File>()));
}
@Test
public void fileListedByOneFilterOnly() throws IOException {
LockFileFileListFilter filter1 = new LockFileFileListFilter(workdir);
LockFileFileListFilter filter2 = new LockFileFileListFilter(workdir);
File testFile = new File(workdir, "test1");
testFile.createNewFile();
assertThat(filter1.filterFiles(workdir.listFiles()).get(0), is(testFile));
locker.lock(testFile);
assertThat(filter2.filterFiles(workdir.listFiles()), is((List)new ArrayList<File>()));
}
private void cleanDirectory(File workdir) {
File[] files = workdir.listFiles();
for (File file : files) {
file.delete();
}
}
}