Author: tjsnell
Date: Mon Jul 4 12:38:25 2011
New Revision: 1142641
URL: http://svn.apache.org/viewvc?rev=1142641&view=rev
Log:
CAMEL-3894
Added min and maxDepth options.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileRecursiveDepthTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileConsumer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileConsumer.java?rev=1142641&r1=1142640&r2=1142641&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileConsumer.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileConsumer.java
Mon Jul 4 12:38:25 2011
@@ -35,9 +35,12 @@ public class FileConsumer extends Generi
this.endpointPath = endpoint.getConfiguration().getDirectory();
}
- protected boolean pollDirectory(String fileName, List<GenericFile<File>>
fileList) {
+ @Override
+ protected boolean pollDirectory(String fileName, List<GenericFile<File>>
fileList, int depth) {
log.trace("pollDirectory from fileName: {}", fileName);
+ depth++;
+
File directory = new File(fileName);
if (!directory.exists() || !directory.isDirectory()) {
log.debug("Cannot poll as directory does not exists or its not a
directory: {}", directory);
@@ -78,17 +81,17 @@ public class FileConsumer extends Generi
GenericFile<File> gf = asGenericFile(endpointPath, file);
if (file.isDirectory()) {
- if (endpoint.isRecursive() && isValidFile(gf, true)) {
+ if (endpoint.isRecursive() && isValidFile(gf, true) && depth <
endpoint.getMaxDepth()) {
// recursive scan and add the sub files and folders
String subDirectory = fileName + File.separator +
file.getName();
- boolean canPollMore = pollDirectory(subDirectory,
fileList);
+ boolean canPollMore = pollDirectory(subDirectory,
fileList, depth);
if (!canPollMore) {
return false;
}
}
} else {
// Windows can report false to a file on a share so regard it
always as a file (if its not a directory)
- if (isValidFile(gf, false)) {
+ if (isValidFile(gf, false) && depth >= endpoint.minDepth) {
if (isInProgress(gf)) {
if (log.isTraceEnabled()) {
log.trace("Skipping as file is already in
progress: {}", gf.getFileName());
@@ -99,6 +102,7 @@ public class FileConsumer extends Generi
fileList.add(gf);
}
}
+
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java?rev=1142641&r1=1142640&r2=1142641&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
Mon Jul 4 12:38:25 2011
@@ -78,7 +78,7 @@ public abstract class GenericFileConsume
// time how long time it takes to poll
StopWatch stop = new StopWatch();
- boolean limitHit = !pollDirectory(name, files);
+ boolean limitHit = !pollDirectory(name, files, 0);
long delta = stop.stop();
if (log.isDebugEnabled()) {
log.debug("Took {} to poll: {}", TimeUtils.printDuration(delta),
name);
@@ -150,7 +150,7 @@ public abstract class GenericFileConsume
// process the current exchange
processExchange(exchange);
}
-
+
// remove the file from the in progress list in case the batch was
limited by max messages per poll
while (exchanges.size() > 0) {
Exchange exchange = (Exchange) exchanges.poll();
@@ -237,7 +237,7 @@ public abstract class GenericFileConsume
* @param fileList current list of files gathered
* @return whether or not to continue polling, <tt>false</tt> means the
maxMessagesPerPoll limit has been hit
*/
- protected abstract boolean pollDirectory(String fileName,
List<GenericFile<T>> fileList);
+ protected abstract boolean pollDirectory(String fileName,
List<GenericFile<T>> fileList, int depth);
/**
* Sets the operations to be used.
@@ -450,7 +450,7 @@ public abstract class GenericFileConsume
fileExpressionResult = endpoint.getFileName().evaluate(dummy,
String.class);
}
}
-
+
@SuppressWarnings("unchecked")
private GenericFile<T> getExchangeFileProperty(Exchange exchange) {
return (GenericFile<T>)
exchange.getProperty(FileComponent.FILE_EXCHANGE_FILE);
@@ -459,7 +459,7 @@ public abstract class GenericFileConsume
@Override
protected void doStart() throws Exception {
super.doStart();
-
+
// prepare on startup
endpoint.getGenericFileProcessStrategy().prepareOnStartup(operations,
endpoint);
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java?rev=1142641&r1=1142640&r2=1142641&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java
Mon Jul 4 12:38:25 2011
@@ -69,6 +69,8 @@ public abstract class GenericFileEndpoin
protected boolean delete;
protected boolean flatten;
protected int maxMessagesPerPoll;
+ protected int maxDepth = Integer.MAX_VALUE;
+ protected int minDepth;
protected String tempPrefix;
protected Expression tempFileName;
protected boolean eagerDeleteTargetFile = true;
@@ -514,6 +516,23 @@ public abstract class GenericFileEndpoin
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
+ public int getMaxDepth() {
+ return maxDepth;
+ }
+
+ public void setMaxDepth(int maxDepth) {
+ this.maxDepth = maxDepth;
+ }
+
+ public int getMinDepth() {
+
+ return minDepth;
+ }
+
+ public void setMinDepth(int minDepth) {
+ this.minDepth = minDepth;
+ }
+
public IdempotentRepository<String> getInProgressRepository() {
return inProgressRepository;
}
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileRecursiveDepthTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileRecursiveDepthTest.java?rev=1142641&view=auto
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileRecursiveDepthTest.java
(added)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileRecursiveDepthTest.java
Mon Jul 4 12:38:25 2011
@@ -0,0 +1,98 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.file;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+
+public class FileRecursiveDepthTest extends ContextTestSupport {
+
+ @Override
+ protected void setUp() throws Exception {
+ deleteDirectory("target/depth");
+ super.setUp();
+ }
+
+
+ public void testDepth() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a2", "b2");
+
+ template.sendBodyAndHeader("file:target/depth", "a",
Exchange.FILE_NAME, "a.txt");
+ template.sendBodyAndHeader("file:target/depth", "b",
Exchange.FILE_NAME, "b.txt");
+ template.sendBodyAndHeader("file:target/depth/foo", "a2",
Exchange.FILE_NAME, "a2.txt");
+ template.sendBodyAndHeader("file:target/depth/foo/bar", "a3",
Exchange.FILE_NAME, "a.txt");
+ template.sendBodyAndHeader("file:target/depth/bar", "b2",
Exchange.FILE_NAME, "b2.txt");
+ template.sendBodyAndHeader("file:target/depth/bar/foo", "b3",
Exchange.FILE_NAME, "b.txt");
+
+ // only expect 2 of the 6 sent, those at depth 2
+ assertMockEndpointsSatisfied();
+ }
+
+
+ public void testDepthMin2Max99() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+
+ mock.expectedBodiesReceivedInAnyOrder("a2", "b2", "a3", "b3");
+
+ template.sendBodyAndHeader("file:target2/depth/2", "a",
Exchange.FILE_NAME, "a.txt");
+ template.sendBodyAndHeader("file:target2/depth/2", "b",
Exchange.FILE_NAME, "b.txt");
+ template.sendBodyAndHeader("file:target2/depth/2/bar", "b2",
Exchange.FILE_NAME, "b2.txt");
+ template.sendBodyAndHeader("file:target2/depth/2/foo", "a2",
Exchange.FILE_NAME, "a2.txt");
+ template.sendBodyAndHeader("file:target2/depth/2/foo/bar", "a3",
Exchange.FILE_NAME, "a3.txt");
+ template.sendBodyAndHeader("file:target2/depth/2/bar/foo", "b3",
Exchange.FILE_NAME, "b3.txt");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testMin1Max1() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+
+ mock.expectedBodiesReceivedInAnyOrder("a", "b");
+
+ template.sendBodyAndHeader("file:target/depth/3", "a",
Exchange.FILE_NAME, "a.txt");
+ template.sendBodyAndHeader("file:target/depth/3", "b",
Exchange.FILE_NAME, "b.txt");
+ template.sendBodyAndHeader("file:target/depth/3/foo", "a2",
Exchange.FILE_NAME, "a.txt");
+ template.sendBodyAndHeader("file:target/depth/3/foo/bar", "a3",
Exchange.FILE_NAME, "a.txt");
+ template.sendBodyAndHeader("file:target/depth/3/bar", "b2",
Exchange.FILE_NAME, "b.txt");
+ template.sendBodyAndHeader("file:target/depth/3/bar/foo", "b3",
Exchange.FILE_NAME, "b.txt");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ from("file:target/depth?recursive=true&minDepth=2&maxDepth=2")
+ .convertBodyTo(String.class)
+ .to("mock:result");
+
+
from("file:target2/depth/2?recursive=true&minDepth=2&maxDepth=99")
+ .convertBodyTo(String.class)
+ .to("mock:result");
+
+
from("file:target/depth/3?recursive=true&noop=true&minDepth=1&maxDepth=1")
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ };
+ }
+}
Modified:
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java?rev=1142641&r1=1142640&r2=1142641&view=diff
==============================================================================
---
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java
(original)
+++
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java
Mon Jul 4 12:38:25 2011
@@ -37,7 +37,8 @@ public class FtpConsumer extends RemoteF
this.endpointPath = endpoint.getConfiguration().getDirectory();
}
- protected boolean pollDirectory(String fileName,
List<GenericFile<FTPFile>> fileList) {
+ @Override
+ protected boolean pollDirectory(String fileName,
List<GenericFile<FTPFile>> fileList, int depth) {
String currentDir = null;
if (isStepwise()) {
// must remember current dir so we stay in that directory after
the poll
Modified:
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java?rev=1142641&r1=1142640&r2=1142641&view=diff
==============================================================================
---
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java
(original)
+++
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java
Mon Jul 4 12:38:25 2011
@@ -36,7 +36,8 @@ public class SftpConsumer extends Remote
this.endpointPath = endpoint.getConfiguration().getDirectory();
}
- protected boolean pollDirectory(String fileName,
List<GenericFile<ChannelSftp.LsEntry>> fileList) {
+ @Override
+ protected boolean pollDirectory(String fileName,
List<GenericFile<ChannelSftp.LsEntry>> fileList, int depth) {
String currentDir = null;
if (isStepwise()) {
// must remember current dir so we stay in that directory after
the poll
