This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 4f37187 CAMEL-17045: implements resume support for the camel-file
component
4f37187 is described below
commit 4f371871496d13f9855f24887afbb28e9ef5c2f8
Author: Otavio Rodolfo Piske <[email protected]>
AuthorDate: Thu Oct 7 17:09:49 2021 +0200
CAMEL-17045: implements resume support for the camel-file component
---
.../org/apache/camel/catalog/components/file.json | 1 +
.../component/file/FileEndpointConfigurer.java | 6 ++
.../component/file/FileEndpointUriFactory.java | 3 +-
.../org/apache/camel/component/file/file.json | 1 +
.../apache/camel/component/file/FileConsumer.java | 112 +++++++++++++--------
.../apache/camel/component/file/FileEndpoint.java | 20 ++++
.../camel/component/file/GenericFileConsumer.java | 33 +++---
.../file/consumer/FileConsumerResumeStrategy.java | 34 +++++++
.../component/file/consumer/FileResumeInfo.java | 43 ++++++++
.../endpoint/dsl/FileEndpointBuilderFactory.java | 38 +++++++
10 files changed, 235 insertions(+), 56 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/file.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/file.json
index 90a50d3..8f254fb 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/file.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/file.json
@@ -49,6 +49,7 @@
"pollStrategy": { "kind": "parameter", "displayName": "Poll Strategy",
"group": "consumer (advanced)", "label": "consumer,advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.spi.PollingConsumerPollStrategy", "deprecated": false,
"autowired": false, "secret": false, "description": "A pluggable
org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your
custom implementation to control error handling usually occurred during the
poll operation [...]
"probeContentType": { "kind": "parameter", "displayName": "Probe Content
Type", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": false,
"description": "Whether to enable probing of the content type. If enable then
the consumer uses Files#probeContentType(java.nio.file.Path) to determine the
content-type of the file, and store that as a he [...]
"processStrategy": { "kind": "parameter", "displayName": "Process
Strategy", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "object", "javaType":
"org.apache.camel.component.file.GenericFileProcessStrategy<java.io.File>",
"deprecated": false, "autowired": false, "secret": false, "description": "A
pluggable org.apache.camel.component.file.GenericFileProcessStrategy allowing
you to implement your own readLock option or similar. Can also be used [...]
+ "resumeStrategy": { "kind": "parameter", "displayName": "Resume Strategy",
"group": "consumer (advanced)", "label": "consumer,advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.component.file.consumer.FileConsumerResumeStrategy",
"deprecated": false, "autowired": false, "secret": false, "description": "Set a
resume strategy for files. This makes it possible to define a strategy for
resuming reading files after the last point before stopping the application.
[...]
"startingDirectoryMustExist": { "kind": "parameter", "displayName":
"Starting Directory Must Exist", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the starting directory must
exist. Mind that the autoCreate option is default enabled, which means the
starting directory is normally auto created if it doesn' [...]
"startingDirectoryMustHaveAccess": { "kind": "parameter", "displayName":
"Starting Directory Must Have Access", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the starting directory has
access permissions. Mind that the startingDirectoryMustExist parameter must be
set to true in order to verify that the di [...]
"appendChars": { "kind": "parameter", "displayName": "Append Chars",
"group": "producer", "label": "producer", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "description": "Used to append characters (text) after writing
files. This can for example be used to add new lines or other separators when
writing and appending new files or existing files. To specify new-line (slash-n
or slash-r) or tab (slash-t) [...]
diff --git
a/components/camel-file/src/generated/java/org/apache/camel/component/file/FileEndpointConfigurer.java
b/components/camel-file/src/generated/java/org/apache/camel/component/file/FileEndpointConfigurer.java
index 1e914ba..f291e0b 100644
---
a/components/camel-file/src/generated/java/org/apache/camel/component/file/FileEndpointConfigurer.java
+++
b/components/camel-file/src/generated/java/org/apache/camel/component/file/FileEndpointConfigurer.java
@@ -162,6 +162,8 @@ public class FileEndpointConfigurer extends
PropertyConfigurerSupport implements
case "renameUsingCopy":
target.setRenameUsingCopy(property(camelContext, boolean.class, value)); return
true;
case "repeatcount":
case "repeatCount": target.setRepeatCount(property(camelContext,
long.class, value)); return true;
+ case "resumestrategy":
+ case "resumeStrategy": target.setResumeStrategy(property(camelContext,
org.apache.camel.component.file.consumer.FileConsumerResumeStrategy.class,
value)); return true;
case "runlogginglevel":
case "runLoggingLevel":
target.setRunLoggingLevel(property(camelContext,
org.apache.camel.LoggingLevel.class, value)); return true;
case "scheduledexecutorservice":
@@ -338,6 +340,8 @@ public class FileEndpointConfigurer extends
PropertyConfigurerSupport implements
case "renameUsingCopy": return boolean.class;
case "repeatcount":
case "repeatCount": return long.class;
+ case "resumestrategy":
+ case "resumeStrategy": return
org.apache.camel.component.file.consumer.FileConsumerResumeStrategy.class;
case "runlogginglevel":
case "runLoggingLevel": return org.apache.camel.LoggingLevel.class;
case "scheduledexecutorservice":
@@ -515,6 +519,8 @@ public class FileEndpointConfigurer extends
PropertyConfigurerSupport implements
case "renameUsingCopy": return target.isRenameUsingCopy();
case "repeatcount":
case "repeatCount": return target.getRepeatCount();
+ case "resumestrategy":
+ case "resumeStrategy": return target.getResumeStrategy();
case "runlogginglevel":
case "runLoggingLevel": return target.getRunLoggingLevel();
case "scheduledexecutorservice":
diff --git
a/components/camel-file/src/generated/java/org/apache/camel/component/file/FileEndpointUriFactory.java
b/components/camel-file/src/generated/java/org/apache/camel/component/file/FileEndpointUriFactory.java
index 8387a9d..c77bc1c 100644
---
a/components/camel-file/src/generated/java/org/apache/camel/component/file/FileEndpointUriFactory.java
+++
b/components/camel-file/src/generated/java/org/apache/camel/component/file/FileEndpointUriFactory.java
@@ -20,7 +20,7 @@ public class FileEndpointUriFactory extends
org.apache.camel.support.component.E
private static final Set<String> PROPERTY_NAMES;
private static final Set<String> SECRET_PROPERTY_NAMES;
static {
- Set<String> props = new HashSet<>(94);
+ Set<String> props = new HashSet<>(95);
props.add("renameUsingCopy");
props.add("moveExistingFileStrategy");
props.add("fileName");
@@ -57,6 +57,7 @@ public class FileEndpointUriFactory extends
org.apache.camel.support.component.E
props.add("lazyStartProducer");
props.add("delay");
props.add("startScheduler");
+ props.add("resumeStrategy");
props.add("readLockMarkerFile");
props.add("readLockTimeout");
props.add("exceptionHandler");
diff --git
a/components/camel-file/src/generated/resources/org/apache/camel/component/file/file.json
b/components/camel-file/src/generated/resources/org/apache/camel/component/file/file.json
index 90a50d3..8f254fb 100644
---
a/components/camel-file/src/generated/resources/org/apache/camel/component/file/file.json
+++
b/components/camel-file/src/generated/resources/org/apache/camel/component/file/file.json
@@ -49,6 +49,7 @@
"pollStrategy": { "kind": "parameter", "displayName": "Poll Strategy",
"group": "consumer (advanced)", "label": "consumer,advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.spi.PollingConsumerPollStrategy", "deprecated": false,
"autowired": false, "secret": false, "description": "A pluggable
org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your
custom implementation to control error handling usually occurred during the
poll operation [...]
"probeContentType": { "kind": "parameter", "displayName": "Probe Content
Type", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": false,
"description": "Whether to enable probing of the content type. If enable then
the consumer uses Files#probeContentType(java.nio.file.Path) to determine the
content-type of the file, and store that as a he [...]
"processStrategy": { "kind": "parameter", "displayName": "Process
Strategy", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "object", "javaType":
"org.apache.camel.component.file.GenericFileProcessStrategy<java.io.File>",
"deprecated": false, "autowired": false, "secret": false, "description": "A
pluggable org.apache.camel.component.file.GenericFileProcessStrategy allowing
you to implement your own readLock option or similar. Can also be used [...]
+ "resumeStrategy": { "kind": "parameter", "displayName": "Resume Strategy",
"group": "consumer (advanced)", "label": "consumer,advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.component.file.consumer.FileConsumerResumeStrategy",
"deprecated": false, "autowired": false, "secret": false, "description": "Set a
resume strategy for files. This makes it possible to define a strategy for
resuming reading files after the last point before stopping the application.
[...]
"startingDirectoryMustExist": { "kind": "parameter", "displayName":
"Starting Directory Must Exist", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the starting directory must
exist. Mind that the autoCreate option is default enabled, which means the
starting directory is normally auto created if it doesn' [...]
"startingDirectoryMustHaveAccess": { "kind": "parameter", "displayName":
"Starting Directory Must Have Access", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the starting directory has
access permissions. Mind that the startingDirectoryMustExist parameter must be
set to true in order to verify that the di [...]
"appendChars": { "kind": "parameter", "displayName": "Append Chars",
"group": "producer", "label": "producer", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "description": "Used to append characters (text) after writing
files. This can for example be used to add new lines or other separators when
writing and appending new files or existing files. To specify new-line (slash-n
or slash-r) or tab (slash-t) [...]
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
index 50087fb..fa3cb13 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
@@ -31,6 +31,8 @@ import java.util.Set;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
+import org.apache.camel.component.file.consumer.FileConsumerResumeStrategy;
+import org.apache.camel.component.file.consumer.FileResumeInfo;
import org.apache.camel.util.FileUtil;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
@@ -42,6 +44,7 @@ import org.slf4j.LoggerFactory;
public class FileConsumer extends GenericFileConsumer<File> {
private static final Logger LOG =
LoggerFactory.getLogger(FileConsumer.class);
+ private final FileConsumerResumeStrategy resumeStrategy;
private String endpointPath;
private Set<String> extendedAttributes;
@@ -54,6 +57,8 @@ public class FileConsumer extends GenericFileConsumer<File> {
List<String> attributes =
Arrays.asList(endpoint.getExtendedAttributes().split(","));
this.extendedAttributes = new HashSet<>(attributes);
}
+
+ resumeStrategy = endpoint.getResumeStrategy();
}
@Override
@@ -83,25 +88,18 @@ public class FileConsumer extends GenericFileConsumer<File>
{
if (LOG.isTraceEnabled()) {
LOG.trace("Polling directory: {}, absolute path: {}",
directory.getPath(), directory.getAbsolutePath());
}
- File[] dirFiles = directory.listFiles();
- if (dirFiles == null || dirFiles.length == 0) {
- // no files in this directory to poll
- if (LOG.isTraceEnabled()) {
- LOG.trace("No files found in directory: {}",
directory.getPath());
- }
+ File[] files = listFiles(directory);
+ if (files == null || files.length == 0) {
return true;
- } else {
- // we found some files
- if (LOG.isTraceEnabled()) {
- LOG.trace("Found {} in directory: {}", dirFiles.length,
directory.getPath());
- }
}
- List<File> files = Arrays.asList(dirFiles);
+
if (getEndpoint().isPreSort()) {
- files.sort(Comparator.comparing(File::getAbsoluteFile));
+ Arrays.sort(files, Comparator.comparing(File::getAbsoluteFile));
}
- for (File file : dirFiles) {
+ List<File> filesArray = Arrays.asList(files);
+
+ for (File file : files) {
// check if we can continue polling in files
if (!canPollMoreFiles(fileList)) {
return false;
@@ -120,7 +118,7 @@ public class FileConsumer extends GenericFileConsumer<File>
{
= asGenericFile(endpointPath, file,
getEndpoint().getCharset(), getEndpoint().isProbeContentType());
if (file.isDirectory()) {
- if (endpoint.isRecursive() && depth < endpoint.getMaxDepth()
&& isValidFile(gf, true, files)) {
+ if (endpoint.isRecursive() && depth < endpoint.getMaxDepth()
&& isValidFile(gf, true, filesArray)) {
// recursive scan and add the sub files and folders
String subDirectory = fileName + File.separator +
file.getName();
boolean canPollMore = pollDirectory(subDirectory,
fileList, depth);
@@ -130,39 +128,15 @@ public class FileConsumer extends
GenericFileConsumer<File> {
}
} else {
// Windows can report false to a file on a share so regard it
- // always as a file (if its not a directory)
- if (depth >= endpoint.minDepth && isValidFile(gf, false,
files)) {
+ // always as a file (if it is not a directory)
+ if (depth >= endpoint.minDepth && isValidFile(gf, false,
filesArray)) {
LOG.trace("Adding valid file: {}", file);
// matched file so add
if (extendedAttributes != null) {
Path path = file.toPath();
Map<String, Object> allAttributes = new HashMap<>();
for (String attribute : extendedAttributes) {
- try {
- String prefix = null;
- if (attribute.endsWith(":*")) {
- prefix = attribute.substring(0,
attribute.length() - 1);
- } else if (attribute.equals("*")) {
- prefix = "basic:";
- }
-
- if (ObjectHelper.isNotEmpty(prefix)) {
- Map<String, Object> attributes =
Files.readAttributes(path, attribute);
- if (attributes != null) {
- for (Map.Entry<String, Object> entry :
attributes.entrySet()) {
- allAttributes.put(prefix +
entry.getKey(), entry.getValue());
- }
- }
- } else if (!attribute.contains(":")) {
- allAttributes.put("basic:" + attribute,
Files.getAttribute(path, attribute));
- } else {
- allAttributes.put(attribute,
Files.getAttribute(path, attribute));
- }
- } catch (IOException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Unable to read attribute {} on
file {}", attribute, file, e);
- }
- }
+ readAttributes(file, path, allAttributes,
attribute);
}
gf.setExtendedAttributes(allAttributes);
@@ -177,6 +151,60 @@ public class FileConsumer extends
GenericFileConsumer<File> {
return true;
}
+ private File[] listFiles(File directory) {
+ final File[] dirFiles = directory.listFiles();
+
+ if (dirFiles == null || dirFiles.length == 0) {
+ // no files in this directory to poll
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("No files found in directory: {}",
directory.getPath());
+ }
+ return null;
+ } else {
+ // we found some files
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Found {} in directory: {}", dirFiles.length,
directory.getPath());
+ }
+ }
+
+ if (resumeStrategy != null) {
+ FileResumeInfo resumeInfo = new FileResumeInfo(dirFiles);
+ resumeStrategy.resume(resumeInfo);
+
+ return resumeInfo.getOutputFiles();
+ }
+
+ return dirFiles;
+ }
+
+ private void readAttributes(File file, Path path, Map<String, Object>
allAttributes, String attribute) {
+ try {
+ String prefix = null;
+ if (attribute.endsWith(":*")) {
+ prefix = attribute.substring(0, attribute.length() - 1);
+ } else if (attribute.equals("*")) {
+ prefix = "basic:";
+ }
+
+ if (ObjectHelper.isNotEmpty(prefix)) {
+ Map<String, Object> attributes = Files.readAttributes(path,
attribute);
+ if (attributes != null) {
+ for (Map.Entry<String, Object> entry :
attributes.entrySet()) {
+ allAttributes.put(prefix + entry.getKey(),
entry.getValue());
+ }
+ }
+ } else if (!attribute.contains(":")) {
+ allAttributes.put("basic:" + attribute,
Files.getAttribute(path, attribute));
+ } else {
+ allAttributes.put(attribute, Files.getAttribute(path,
attribute));
+ }
+ } catch (IOException e) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Unable to read attribute {} on file {}", attribute,
file, e);
+ }
+ }
+ }
+
@Override
protected boolean isMatched(GenericFile<File> file, String doneFileName,
List<File> files) {
String onlyName = FileUtil.stripPath(doneFileName);
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/FileEndpoint.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/FileEndpoint.java
index dc51742..c57f0f3 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/FileEndpoint.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/FileEndpoint.java
@@ -31,6 +31,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.PollingConsumer;
import org.apache.camel.Processor;
+import org.apache.camel.component.file.consumer.FileConsumerResumeStrategy;
import org.apache.camel.component.file.strategy.FileMoveExistingStrategy;
import org.apache.camel.component.file.strategy.FileProcessStrategyFactory;
import org.apache.camel.spi.Metadata;
@@ -82,6 +83,9 @@ public class FileEndpoint extends GenericFileEndpoint<File> {
@UriParam(label = "producer,advanced")
private String chmodDirectory;
+ @UriParam(label = "consumer,advanced")
+ private FileConsumerResumeStrategy resumeStrategy;
+
public FileEndpoint() {
}
@@ -360,6 +364,22 @@ public class FileEndpoint extends
GenericFileEndpoint<File> {
this.extendedAttributes = extendedAttributes;
}
+ public FileConsumerResumeStrategy getResumeStrategy() {
+ return resumeStrategy;
+ }
+
+ /**
+ * Set a resume strategy for files. This makes it possible to define a
strategy for resuming reading files after the
+ * last point before stopping the application.
+ *
+ * See the {@link FileConsumerResumeStrategy} for implementation details
+ *
+ * @param resumeStrategy an instance of the resume strategy to be used
+ */
+ public void setResumeStrategy(FileConsumerResumeStrategy resumeStrategy) {
+ this.resumeStrategy = resumeStrategy;
+ }
+
/**
* Chmod value must be between 000 and 777; If there is a leading digit
like in 0755 we will ignore it.
*/
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
index a3fa67c..17448e1 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
@@ -590,21 +590,10 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
return false;
}
- // if its a file then check we have the file in the idempotent registry
+ // if it is a file then check we have the file in the idempotent
registry
// already
if (endpoint.isIdempotent()) {
- // use absolute file path as default key, but evaluate if an
- // expression key was configured
- String key = file.getAbsoluteFilePath();
- if (endpoint.getIdempotentKey() != null) {
- Exchange dummy = endpoint.createExchange(file);
- key = endpoint.getIdempotentKey().evaluate(dummy,
String.class);
- LOG.trace("Evaluated idempotentKey: {} for file: {}", key,
file);
- }
- if (key != null &&
endpoint.getIdempotentRepository().contains(key)) {
- LOG.trace(
- "This consumer is idempotent and the file has been
consumed before matching idempotentKey: {}. Will skip this file: {}",
- key, file);
+ if (notUnique(file)) {
return false;
}
}
@@ -615,6 +604,24 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
return endpoint.getInProgressRepository().add(absoluteFilePath);
}
+ private boolean notUnique(GenericFile<T> file) {
+ // use absolute file path as default key, but evaluate if an
+ // expression key was configured
+ String key = file.getAbsoluteFilePath();
+ if (endpoint.getIdempotentKey() != null) {
+ Exchange dummy = endpoint.createExchange(file);
+ key = endpoint.getIdempotentKey().evaluate(dummy, String.class);
+ LOG.trace("Evaluated idempotentKey: {} for file: {}", key, file);
+ }
+ if (key != null && endpoint.getIdempotentRepository().contains(key)) {
+ LOG.trace(
+ "This consumer is idempotent and the file has been
consumed before matching idempotentKey: {}. Will skip this file: {}",
+ key, file);
+ return true;
+ }
+ return false;
+ }
+
/**
* Strategy to perform file matching based on endpoint configuration.
* <p/>
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileConsumerResumeStrategy.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileConsumerResumeStrategy.java
new file mode 100644
index 0000000..fc4a991
--- /dev/null
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileConsumerResumeStrategy.java
@@ -0,0 +1,34 @@
+/*
+ * 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.consumer;
+
+import org.apache.camel.ResumeStrategy;
+
+/**
+ * Defines resume strategy for consumers of the file component.
+ */
+public interface FileConsumerResumeStrategy extends
ResumeStrategy<FileResumeInfo> {
+
+ /**
+ * Perform the resume operation. This runs in the scope of the
FileConsumer instance thread.
+ *
+ * @param resumeInfo resume information
+ */
+ @Override
+ void resume(FileResumeInfo resumeInfo);
+}
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileResumeInfo.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileResumeInfo.java
new file mode 100644
index 0000000..ae6e0eb
--- /dev/null
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/consumer/FileResumeInfo.java
@@ -0,0 +1,43 @@
+/*
+ * 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.consumer;
+
+import java.io.File;
+import java.util.Objects;
+
+public final class FileResumeInfo {
+ private final File[] inputFiles;
+ private File[] outputFiles;
+
+ public FileResumeInfo(File[] inputFiles) {
+ Objects.requireNonNull(inputFiles, "A list of input files must be
provided for the resume info");
+ this.inputFiles = inputFiles;
+ }
+
+ public File[] getInputFiles() {
+ return inputFiles;
+ }
+
+ public File[] getOutputFiles() {
+ return outputFiles;
+ }
+
+ public void setOutputFiles(File[] outputFiles) {
+ this.outputFiles = outputFiles;
+ }
+}
diff --git
a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FileEndpointBuilderFactory.java
b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FileEndpointBuilderFactory.java
index d88af29..d021942 100644
---
a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FileEndpointBuilderFactory.java
+++
b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FileEndpointBuilderFactory.java
@@ -2558,6 +2558,44 @@ public interface FileEndpointBuilderFactory {
return this;
}
/**
+ * Set a resume strategy for files. This makes it possible to define a
+ * strategy for resuming reading files after the last point before
+ * stopping the application. See the FileConsumerResumeStrategy for
+ * implementation details.
+ *
+ * The option is a:
+ *
<code>org.apache.camel.component.file.consumer.FileConsumerResumeStrategy</code>
type.
+ *
+ * Group: consumer (advanced)
+ *
+ * @param resumeStrategy the value to set
+ * @return the dsl builder
+ */
+ default AdvancedFileEndpointConsumerBuilder resumeStrategy(
+ Object resumeStrategy) {
+ doSetProperty("resumeStrategy", resumeStrategy);
+ return this;
+ }
+ /**
+ * Set a resume strategy for files. This makes it possible to define a
+ * strategy for resuming reading files after the last point before
+ * stopping the application. See the FileConsumerResumeStrategy for
+ * implementation details.
+ *
+ * The option will be converted to a
+ *
<code>org.apache.camel.component.file.consumer.FileConsumerResumeStrategy</code>
type.
+ *
+ * Group: consumer (advanced)
+ *
+ * @param resumeStrategy the value to set
+ * @return the dsl builder
+ */
+ default AdvancedFileEndpointConsumerBuilder resumeStrategy(
+ String resumeStrategy) {
+ doSetProperty("resumeStrategy", resumeStrategy);
+ return this;
+ }
+ /**
* Whether the starting directory must exist. Mind that the autoCreate
* option is default enabled, which means the starting directory is
* normally auto created if it doesn't exist. You can disable
autoCreate
