You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2008/07/27 19:46:29 UTC
svn commit: r680159 - in /activemq/camel/trunk:
camel-core/src/main/java/org/apache/camel/component/file/
camel-core/src/test/java/org/apache/camel/component/file/
components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/
components/ca...
Author: davsclaus
Date: Sun Jul 27 10:46:28 2008
New Revision: 680159
URL: http://svn.apache.org/viewvc?rev=680159&view=rev
Log:
Polished exclusive read lock for FileConsumer and FTPConsumer. Not possible to unit test in single JVM with unit test and Camel running on same JVM. See FileLock javadoc.
Modified:
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileConsumer.java
activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileExclusiveReadTest.java
activemq/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java
activemq/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java
activemq/camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpExclusiveReadTest.java
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileConsumer.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileConsumer.java?rev=680159&r1=680158&r2=680159&view=diff
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileConsumer.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileConsumer.java Sun Jul 27 10:46:28 2008
@@ -141,7 +141,7 @@
boolean handled = DeadLetterChannel.isFailureHandled(exchange);
if (LOG.isDebugEnabled()) {
- LOG.debug("Done processing file: " + file + ". Status is: " + (failed ? "failed: " + failed + ", handled by failure processor: " + handled : "OK"));
+ LOG.debug("Done processing file: " + file + ". Status is: " + (failed ? "failed: " + failed + ", handled by failure processor: " + handled : "processed OK"));
}
if (!failed || handled) {
@@ -170,26 +170,24 @@
protected void acquireExclusiveRead(File file) throws IOException {
if (LOG.isTraceEnabled()) {
- LOG.trace("Acquiring exclusive read (avoid reading file that is in progress of being written) to " + file);
+ LOG.trace("Waiting for exclusive lock to file: " + file);
}
// try to acquire rw lock on the file before we can consume it
FileChannel channel = new RandomAccessFile(file, "rw").getChannel();
try {
FileLock lock = channel.lock();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Acquired exclusive lock: " + lock + " to file: " + file);
+ }
// just release it now we dont want to hold it during the rest of the processing
lock.release();
} finally {
// must close channel
- ObjectHelper.close(channel, "FileConsumer during acquiring of exclusive read", LOG);
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Acquired exclusive read to: " + file);
+ ObjectHelper.close(channel, "FileConsumer during acquiring of exclusive lock", LOG);
}
}
-
/**
* Strategy when the file was processed and a commit should be executed.
*
Modified: activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileExclusiveReadTest.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileExclusiveReadTest.java?rev=680159&r1=680158&r2=680159&view=diff
==============================================================================
--- activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileExclusiveReadTest.java (original)
+++ activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileExclusiveReadTest.java Sun Jul 27 10:46:28 2008
@@ -53,13 +53,13 @@
mock.assertIsSatisfied();
}
- // TODO: Fix on Bamboo
+ // TODO: Not possible to test in the same JVM (see javadoc for FileLock)
public void xxxtestPollFileWhileSlowFileIsBeingWritten() throws Exception {
deleteDirectory("./target/exclusiveread");
createDirectory("./target/exclusiveread/slowfile");
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedMessageCount(1);
- mock.expectedBodiesReceived("Hello World");
+ mock.expectedBodiesReceived("Hello WorldLine #0Line #1Line #2Bye Worl");
createSlowFile();
@@ -68,24 +68,17 @@
private void createSlowFile() throws Exception {
LOG.info("Creating a slow file ...");
-
File file = new File("./target/exclusiveread/slowfile/hello.txt");
FileOutputStream fos = new FileOutputStream(file);
-
- // get a lock so we are the only one working on this file
FileLock lock = fos.getChannel().lock();
-
- byte[] buffer = "Hello World".getBytes();
- ByteBuffer bb = ByteBuffer.wrap(buffer);
- for (int i = 0; i < buffer.length; i++) {
+ fos.write("Hello World".getBytes());
+ for (int i = 0; i < 3; i++) {
+ Thread.sleep(1000);
+ fos.write(("Line #" + i).getBytes());
LOG.info("Appending to slowfile");
- Thread.sleep(300);
}
- LOG.info("Writing to file");
- fos.write(buffer);
- LOG.info("Releasing lock");
+ fos.write("Bye World".getBytes());
lock.release();
- LOG.info("Closing file");
fos.close();
LOG.info("... done creating slowfile");
}
Modified: activemq/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java?rev=680159&r1=680158&r2=680159&view=diff
==============================================================================
--- activemq/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java (original)
+++ activemq/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java Sun Jul 27 10:46:28 2008
@@ -210,20 +210,25 @@
}
protected void acquireExclusiveRead(FTPClient client, FTPFile ftpFile) throws IOException {
- LOG.trace("Acquiring exclusive read (avoid reading file that is in progress of being written)");
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Waiting for exclusive lock to file: " + ftpFile);
+ }
// the trick is to try to rename the file, if we can rename then we have exclusive read
- // since its a remote file we can not use java.nio to get a RW access
+ // since its a remote file we can not use java.nio to get a RW lock
String originalName = ftpFile.getName();
String newName = originalName + ".camelExclusiveRead";
boolean exclusive = false;
while (! exclusive) {
exclusive = client.rename(originalName, newName);
if (exclusive) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Acquired exclusive lock to file: " + originalName);
+ }
// rename it back so we can read it
client.rename(newName, originalName);
} else {
- LOG.trace("Exclusive read not granted. Sleeping for 1000 millis");
+ LOG.trace("Exclusive lock not granted. Sleeping for 1000 millis.");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
@@ -231,9 +236,6 @@
}
}
}
- if (LOG.isDebugEnabled()) {
- LOG.debug("Acquired exclusive read to: " + originalName);
- }
}
private String remoteServer() {
Modified: activemq/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java?rev=680159&r1=680158&r2=680159&view=diff
==============================================================================
--- activemq/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java (original)
+++ activemq/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java Sun Jul 27 10:46:28 2008
@@ -220,7 +220,9 @@
}
protected void acquireExclusiveRead(ChannelSftp.LsEntry sftpFile) throws SftpException {
- LOG.trace("Acquiring exclusive read (avoid reading file that is in progress of being written)");
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Waiting for exclusive lock to file: " + sftpFile);
+ }
// the trick is to try to rename the file, if we can rename then we have exclusive read
// since its a remote file we can not use java.nio to get a RW access
@@ -236,10 +238,13 @@
}
if (exclusive) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Acquired exclusive lock to file: " + originalName);
+ }
// rename it back so we can read it
channel.rename(newName, originalName);
} else {
- LOG.trace("Exclusive read not granted. Sleeping for 1000 millis");
+ LOG.trace("Exclusive lock not granted. Sleeping for 1000 millis");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
@@ -247,9 +252,6 @@
}
}
}
- if (LOG.isDebugEnabled()) {
- LOG.debug("Acquired exclusive read to: " + originalName);
- }
}
private String remoteServer() {
Modified: activemq/camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpExclusiveReadTest.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpExclusiveReadTest.java?rev=680159&r1=680158&r2=680159&view=diff
==============================================================================
--- activemq/camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpExclusiveReadTest.java (original)
+++ activemq/camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpExclusiveReadTest.java Sun Jul 27 10:46:28 2008
@@ -38,7 +38,7 @@
return port;
}
- // TODO fix on Bamboo
+ // TODO: Not possible to test in single JVM
public void testPollFileWhileSlowFileIsBeingWritten() throws Exception {
/*deleteDirectory("./res/home");
createDirectory("./res/home/slowfile");