You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2021/10/12 13:03:02 UTC
[camel] branch main updated: CAMEL-17045: implements resume support
for the camel-file component
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 <op...@redhat.com>
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