This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 d02bbc7c0e8 CAMEL-21733: camel-core - Poll EIP to support DynamicAware
to reuse endpoints during dynamic poll EIP (#17092)
d02bbc7c0e8 is described below
commit d02bbc7c0e8036903799d40acb1637e0d27ff78b
Author: Claus Ibsen <[email protected]>
AuthorDate: Sun Feb 9 12:09:18 2025 +0100
CAMEL-21733: camel-core - Poll EIP to support DynamicAware to reuse
endpoints during dynamic poll EIP (#17092)
* CAMEL-21733: camel-core - Poll EIP to support DynamicAware to reuse
endpoints during dynamic poll EIP
---
.../apache/camel/catalog/models/pollEnrich.json | 3 +-
.../apache/camel/catalog/schemas/camel-spring.xsd | 9 ++
.../org/apache/camel/poll-dynamic/azure-files | 2 +
.../org/apache/camel/send-dynamic/azure-files | 2 +
.../camel/component/file/azure/FilesConsumer.java | 20 ++-
.../file/azure/FilesPollDynamicAware.java | 24 +++
.../file/azure/FilesSendDynamicAware.java | 24 +++
.../services/org/apache/camel/poll-dynamic/file | 2 +
.../apache/camel/component/file/FileConsumer.java | 25 ++--
.../camel/component/file/FilePollDynamicAware.java | 23 +++
.../camel/component/file/GenericFileConsumer.java | 46 ++++--
.../file/GenericFilePollDynamicAware.java | 67 +++++++++
.../component/file/GenericFilePollingConsumer.java | 34 +++--
.../services/org/apache/camel/poll-dynamic/ftp | 2 +
.../services/org/apache/camel/poll-dynamic/ftps | 2 +
.../services/org/apache/camel/poll-dynamic/sftp | 2 +
.../camel/component/file/remote/FtpConsumer.java | 27 ++--
.../component/file/remote/FtpPollDynamicAware.java | 24 +++
.../file/remote/FtpsPollDynamicAware.java | 24 +++
.../component/file/remote/RemoteFileConsumer.java | 8 +-
.../camel/component/file/remote/SftpConsumer.java | 14 +-
.../file/remote/SftpPollDynamicAware.java | 24 +++
.../remote/RemoteFileIgnoreDoPollErrorTest.java | 12 +-
.../integration/PollDynamicFileNameTest.java | 105 +++++++++++++
.../services/org/apache/camel/send-dynamic/scp | 2 +
.../camel/component/scp/ScpSendDynamicAware.java | 24 +++
.../services/org/apache/camel/poll-dynamic/smb | 2 +
.../apache/camel/component/smb/SmbConsumer.java | 18 ++-
.../camel/component/smb/SmbPollDynamicAware.java | 24 +++
.../component/smb/PollDynamicFileNameTest.java | 99 +++++++++++++
.../apache/camel/spi/annotations/PollDynamic.java | 33 +++++
.../org/apache/camel/DynamicPollingConsumer.java | 65 ++++++++
...SendDynamicAware.java => PollDynamicAware.java} | 47 ++----
.../org/apache/camel/spi/SendDynamicAware.java | 4 +-
.../org/apache/camel/model/pollEnrich.json | 3 +-
.../apache/camel/model/PollEnrichDefinition.java | 42 ++++++
.../camel/processor/PollDynamicAwareResolver.java | 67 +++++++++
.../org/apache/camel/processor/PollEnricher.java | 163 ++++++++++++++++++---
.../apache/camel/reifier/PollEnrichReifier.java | 3 +
...> PollDynamicFileNameOptimizeDisabledTest.java} | 31 +++-
.../enricher/PollDynamicFileNameTest.java | 26 +++-
.../management/mbean/ManagedPollEnricherMBean.java | 6 +
.../management/mbean/ManagedPollEnricher.java | 10 ++
.../camel/support/ScheduledPollConsumer.java | 12 ++
.../support/component/PollDynamicAwareSupport.java | 144 ++++++++++++++++++
.../java/org/apache/camel/xml/in/ModelParser.java | 1 +
.../java/org/apache/camel/xml/out/ModelWriter.java | 1 +
.../org/apache/camel/yaml/out/ModelWriter.java | 1 +
.../ROOT/pages/camel-4x-upgrade-guide-4_11.adoc | 18 +++
.../modules/ROOT/pages/camel-4x-upgrade-guide.adoc | 1 +
.../dsl/yaml/deserializers/ModelDeserializers.java | 6 +
.../generated/resources/schema/camelYamlDsl.json | 5 +
.../apache/camel/spi/annotations/PollDynamic.java | 33 +++++
53 files changed, 1278 insertions(+), 138 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/pollEnrich.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/pollEnrich.json
index 85409de3ace..5644c4a8a54 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/pollEnrich.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/pollEnrich.json
@@ -24,7 +24,8 @@
"timeout": { "index": 9, "kind": "attribute", "displayName": "Timeout",
"group": "common", "required": false, "type": "duration", "javaType":
"java.lang.String", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "-1", "description": "Timeout in millis when polling from the
external service. The timeout has influence about the poll enrich behavior. It
basically operations in three different modes: negative value - Waits until a
message is available and then ret [...]
"cacheSize": { "index": 10, "kind": "attribute", "displayName": "Cache
Size", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired":
false, "secret": false, "description": "Sets the maximum size used by the
org.apache.camel.spi.ConsumerCache which is used to cache and reuse consumers
when uris are reused. Beware that when using dynamic endpoints then it affects
how well the cache can be utiliz [...]
"ignoreInvalidEndpoint": { "index": 11, "kind": "attribute",
"displayName": "Ignore Invalid Endpoint", "group": "advanced", "label":
"advanced", "required": false, "type": "boolean", "javaType":
"java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Ignore the invalidate endpoint exception
when try to create a producer with that endpoint" },
- "autoStartComponents": { "index": 12, "kind": "attribute", "displayName":
"Auto Start Components", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Whether to auto startup components when poll enricher is starting up." }
+ "allowOptimisedComponents": { "index": 12, "kind": "attribute",
"displayName": "Allow Optimised Components", "group": "advanced", "label":
"advanced", "required": false, "type": "boolean", "javaType":
"java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": true, "description": "Whether to allow components to optimise
if they are org.apache.camel.spi.SendDynamicAware ." },
+ "autoStartComponents": { "index": 13, "kind": "attribute", "displayName":
"Auto Start Components", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Whether to auto startup components when poll enricher is starting up." }
},
"exchangeProperties": {
"CamelToEndpoint": { "index": 0, "kind": "exchangeProperty",
"displayName": "To Endpoint", "label": "producer", "required": false,
"javaType": "String", "deprecated": false, "autowired": false, "secret": false,
"description": "Endpoint URI where this Exchange is being sent to" }
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
index 06008c197e7..216510d472e 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
@@ -11816,6 +11816,15 @@ storing too many non frequent used producers.
<xs:documentation xml:lang="en">
<![CDATA[
Ignore the invalidate endpoint exception when try to create a producer with
that endpoint. Default value: false
+]]>
+ </xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
+ <xs:attribute name="allowOptimisedComponents" type="xs:string">
+ <xs:annotation>
+ <xs:documentation xml:lang="en">
+<![CDATA[
+Whether to allow components to optimise if they are
org.apache.camel.spi.SendDynamicAware . Default value: true
]]>
</xs:documentation>
</xs:annotation>
diff --git
a/components/camel-azure/camel-azure-files/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/azure-files
b/components/camel-azure/camel-azure-files/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/azure-files
new file mode 100644
index 00000000000..c9844afecd1
--- /dev/null
+++
b/components/camel-azure/camel-azure-files/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/azure-files
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.file.azure.FilesPollDynamicAware
diff --git
a/components/camel-azure/camel-azure-files/src/generated/resources/META-INF/services/org/apache/camel/send-dynamic/azure-files
b/components/camel-azure/camel-azure-files/src/generated/resources/META-INF/services/org/apache/camel/send-dynamic/azure-files
new file mode 100644
index 00000000000..d7daea26902
--- /dev/null
+++
b/components/camel-azure/camel-azure-files/src/generated/resources/META-INF/services/org/apache/camel/send-dynamic/azure-files
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.file.azure.FilesSendDynamicAware
diff --git
a/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/FilesConsumer.java
b/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/FilesConsumer.java
index 4a8bd13cbd9..80ac733f457 100644
---
a/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/FilesConsumer.java
+++
b/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/FilesConsumer.java
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.function.Supplier;
import com.azure.storage.file.share.models.ShareFileItem;
+import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.api.management.ManagedResource;
@@ -89,15 +90,17 @@ public class FilesConsumer extends
RemoteFileConsumer<ShareFileItem> {
@Override
protected boolean pollDirectory(
+ Exchange dynamic,
String path, List<GenericFile<ShareFileItem>> fileList,
int depth) {
LOG.trace("pollDirectory({},,{})", path, depth);
- return doPollDirectory(FilesPath.trimTrailingSeparator(path), null,
fileList, depth);
+ return doPollDirectory(dynamic, FilesPath.trimTrailingSeparator(path),
null, fileList, depth);
}
@Override
protected boolean doPollDirectory(
+ Exchange dynamic,
String path, String dirName,
List<GenericFile<ShareFileItem>> fileList, int depth) {
LOG.trace("doPollDirectory({},{},,{})", path, dirName, depth);
@@ -119,7 +122,7 @@ public class FilesConsumer extends
RemoteFileConsumer<ShareFileItem> {
}
for (var fileItem : listedFileItems) {
- if (handleFileItem(path, fileList, depth + 1, listedFileItems,
fileItem)) {
+ if (handleFileItem(dynamic, path, fileList, depth + 1,
listedFileItems, fileItem)) {
return false;
}
}
@@ -128,6 +131,7 @@ public class FilesConsumer extends
RemoteFileConsumer<ShareFileItem> {
}
private boolean handleFileItem(
+ Exchange dynamic,
String path, List<GenericFile<ShareFileItem>> polledFiles,
int depth, ShareFileItem[] listedFileItems, ShareFileItem
fileItem) {
if (LOG.isTraceEnabled()) {
@@ -140,16 +144,17 @@ public class FilesConsumer extends
RemoteFileConsumer<ShareFileItem> {
}
if (fileItem.isDirectory()) {
- if (handleDirectory(path, polledFiles, depth, listedFileItems,
fileItem)) {
+ if (handleDirectory(dynamic, path, polledFiles, depth,
listedFileItems, fileItem)) {
return true;
}
} else {
- handleFile(path, polledFiles, depth, listedFileItems, fileItem);
+ handleFile(dynamic, path, polledFiles, depth, listedFileItems,
fileItem);
}
return false;
}
private boolean handleDirectory(
+ Exchange dynamic,
String path, List<GenericFile<ShareFileItem>> polledFiles,
int depth, ShareFileItem[] listedFileItems, ShareFileItem dir) {
@@ -157,10 +162,10 @@ public class FilesConsumer extends
RemoteFileConsumer<ShareFileItem> {
Supplier<GenericFile<ShareFileItem>> remote =
Suppliers.memorize(() -> asRemoteFile(path, dir));
String absoluteFilePath = FilesPath.concat(path, dir.getName());
Supplier<String> relative = getRelativeFilePath(endpointPath,
path, absoluteFilePath, dir);
- if (isValidFile(remote, dir.getName(), absoluteFilePath, relative,
true, listedFileItems)) {
+ if (isValidFile(dynamic, remote, dir.getName(), absoluteFilePath,
relative, true, listedFileItems)) {
String dirName = dir.getName();
String dirPath = FilesPath.concat(path, dirName);
- boolean canPollMore = doSafePollSubDirectory(dirPath, dirName,
polledFiles, depth);
+ boolean canPollMore = doSafePollSubDirectory(dynamic, dirPath,
dirName, polledFiles, depth);
if (!canPollMore) {
return true;
}
@@ -170,13 +175,14 @@ public class FilesConsumer extends
RemoteFileConsumer<ShareFileItem> {
}
private void handleFile(
+ Exchange dynamic,
String path, List<GenericFile<ShareFileItem>> polledFiles, int
depth,
ShareFileItem[] listedFileItems, ShareFileItem file) {
if (depth >= endpoint.getMinDepth()) {
Supplier<GenericFile<ShareFileItem>> remote =
Suppliers.memorize(() -> asRemoteFile(path, file));
String absoluteFilePath = FilesPath.concat(path, file.getName());
Supplier<String> relative = getRelativeFilePath(endpointPath,
path, absoluteFilePath, file);
- if (isValidFile(remote, file.getName(), absoluteFilePath,
relative, false,
+ if (isValidFile(dynamic, remote, file.getName(), absoluteFilePath,
relative, false,
listedFileItems)) {
polledFiles.add(remote.get());
}
diff --git
a/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/FilesPollDynamicAware.java
b/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/FilesPollDynamicAware.java
new file mode 100644
index 00000000000..a7c54d39281
--- /dev/null
+++
b/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/FilesPollDynamicAware.java
@@ -0,0 +1,24 @@
+/*
+ * 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.azure;
+
+import org.apache.camel.component.file.GenericFilePollDynamicAware;
+import org.apache.camel.spi.annotations.PollDynamic;
+
+@PollDynamic(FilesComponent.SCHEME)
+public class FilesPollDynamicAware extends GenericFilePollDynamicAware {
+}
diff --git
a/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/FilesSendDynamicAware.java
b/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/FilesSendDynamicAware.java
new file mode 100644
index 00000000000..a224262eb0d
--- /dev/null
+++
b/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/FilesSendDynamicAware.java
@@ -0,0 +1,24 @@
+/*
+ * 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.azure;
+
+import org.apache.camel.component.file.GenericFileSendDynamicAware;
+import org.apache.camel.spi.annotations.SendDynamic;
+
+@SendDynamic(FilesComponent.SCHEME)
+public class FilesSendDynamicAware extends GenericFileSendDynamicAware {
+}
diff --git
a/components/camel-file/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/file
b/components/camel-file/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/file
new file mode 100644
index 00000000000..122b90b3c89
--- /dev/null
+++
b/components/camel-file/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/file
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.file.FilePollDynamicAware
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 d2b6dfbd57b..58df9d374c2 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
@@ -74,7 +74,7 @@ public class FileConsumer extends GenericFileConsumer<File>
implements ResumeAwa
return exchange;
}
- private boolean pollDirectory(File directory, List<GenericFile<File>>
fileList, int depth) {
+ private boolean pollDirectory(Exchange dynamic, File directory,
List<GenericFile<File>> fileList, int depth) {
depth++;
if (LOG.isTraceEnabled()) {
@@ -89,14 +89,14 @@ public class FileConsumer extends GenericFileConsumer<File>
implements ResumeAwa
Arrays.sort(files, Comparator.comparing(File::getAbsoluteFile));
}
- if (processPolledFiles(fileList, depth, files)) {
+ if (processPolledFiles(dynamic, fileList, depth, files)) {
return false;
}
return true;
}
- private boolean processPolledFiles(List<GenericFile<File>> fileList, int
depth, File[] files) {
+ private boolean processPolledFiles(Exchange dynamic,
List<GenericFile<File>> fileList, int depth, File[] files) {
for (File file : files) {
// check if we can continue polling in files
if (!canPollMoreFiles(fileList)) {
@@ -125,7 +125,7 @@ public class FileConsumer extends GenericFileConsumer<File>
implements ResumeAwa
}
}
- if (processEntry(fileList, depth, file, gf, files)) {
+ if (processEntry(dynamic, fileList, depth, file, gf, files)) {
return true;
}
}
@@ -133,23 +133,25 @@ public class FileConsumer extends
GenericFileConsumer<File> implements ResumeAwa
}
private boolean processEntry(
+ Exchange dynamic,
List<GenericFile<File>> fileList, int depth, File file,
Supplier<GenericFile<File>> gf, File[] files) {
if (file.isDirectory()) {
- return processDirectoryEntry(fileList, depth, file, gf, files);
+ return processDirectoryEntry(dynamic, fileList, depth, file, gf,
files);
} else {
- processFileEntry(fileList, depth, file, gf, files);
+ processFileEntry(dynamic, fileList, depth, file, gf, files);
}
return false;
}
private void processFileEntry(
+ Exchange dynamic,
List<GenericFile<File>> fileList, int depth, File file,
Supplier<GenericFile<File>> gf, File[] files) {
// Windows can report false to a file on a share so regard it
// always as a file (if it is not a directory)
if (depth >= endpoint.minDepth) {
boolean valid
- = isValidFile(gf, file.getName(), file.getAbsolutePath(),
+ = isValidFile(dynamic, gf, file.getName(),
file.getAbsolutePath(),
getRelativeFilePath(endpointPath, null, null,
file),
false, files);
if (valid) {
@@ -168,14 +170,15 @@ public class FileConsumer extends
GenericFileConsumer<File> implements ResumeAwa
}
private boolean processDirectoryEntry(
+ Exchange dynamic,
List<GenericFile<File>> fileList, int depth, File file,
Supplier<GenericFile<File>> gf, File[] files) {
if (endpoint.isRecursive() && depth < endpoint.getMaxDepth()) {
boolean valid
- = isValidFile(gf, file.getName(), file.getAbsolutePath(),
+ = isValidFile(dynamic, gf, file.getName(),
file.getAbsolutePath(),
getRelativeFilePath(endpointPath, null, null,
file),
true, files);
if (valid) {
- boolean canPollMore = pollDirectory(file, fileList, depth);
+ boolean canPollMore = pollDirectory(dynamic, file, fileList,
depth);
return !canPollMore;
}
}
@@ -194,7 +197,7 @@ public class FileConsumer extends GenericFileConsumer<File>
implements ResumeAwa
}
@Override
- protected boolean pollDirectory(String fileName, List<GenericFile<File>>
fileList, int depth) {
+ protected boolean pollDirectory(Exchange dynamic, String fileName,
List<GenericFile<File>> fileList, int depth) {
LOG.trace("pollDirectory from fileName: {}", fileName);
File directory = new File(fileName);
@@ -206,7 +209,7 @@ public class FileConsumer extends GenericFileConsumer<File>
implements ResumeAwa
return true;
}
- return pollDirectory(directory, fileList, depth);
+ return pollDirectory(dynamic, directory, fileList, depth);
}
private File[] listFiles(File directory) {
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/FilePollDynamicAware.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/FilePollDynamicAware.java
new file mode 100644
index 00000000000..e0a63b50650
--- /dev/null
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/FilePollDynamicAware.java
@@ -0,0 +1,23 @@
+/*
+ * 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.spi.annotations.PollDynamic;
+
+@PollDynamic("file")
+public class FilePollDynamicAware extends GenericFilePollDynamicAware {
+}
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 7f87cb2af1d..de0ffcf65a4 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
@@ -33,6 +33,7 @@ import org.apache.camel.Processor;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.ShutdownRunningTask;
import org.apache.camel.support.EmptyAsyncCallback;
+import org.apache.camel.support.MessageHelper;
import org.apache.camel.support.ScheduledBatchPollingConsumer;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.CastUtils;
@@ -107,11 +108,16 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
*/
protected abstract Exchange createExchange(GenericFile<T> file);
+ @Override
+ protected int poll() throws Exception {
+ return poll(null);
+ }
+
/**
* Poll for files
*/
@Override
- public int poll() throws Exception {
+ protected int poll(Exchange dynamic) throws Exception {
// must prepare on startup the very first time
if (!prepareOnStartup) {
// prepare on startup
@@ -138,7 +144,7 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
StopWatch stop = new StopWatch();
boolean limitHit;
try {
- limitHit = !pollDirectory(name, files, 0);
+ limitHit = !pollDirectory(dynamic, name, files, 0);
} catch (Exception e) {
// during poll directory we add files to the in progress
repository,
// in case of any exception thrown after this work
@@ -336,7 +342,7 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
* @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, int depth);
+ protected abstract boolean pollDirectory(Exchange dynamic, String
fileName, List<GenericFile<T>> fileList, int depth);
/**
* Sets the operations to be used.
@@ -429,7 +435,6 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
// must use full name when downloading so we have the correct path
final String name = target.getAbsoluteFilePath();
try {
-
if (isRetrieveFile()) {
if (!tryRetrievingFile(exchange, name, target,
absoluteFileName, file)) {
return false;
@@ -587,9 +592,10 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
* @return <tt>true</tt> to include the file, <tt>false</tt>
to skip it
*/
protected boolean isValidFile(
+ Exchange dynamic,
Supplier<GenericFile<T>> file, String name, String
absoluteFilePath,
Supplier<String> relativeFilePath, boolean isDirectory, T[] files)
{
- if (!isMatched(file, name, absoluteFilePath, relativeFilePath,
isDirectory, files)) {
+ if (!isMatched(dynamic, file, name, absoluteFilePath,
relativeFilePath, isDirectory, files)) {
LOG.trace("File did not match. Will skip this file: {}", name);
return false;
}
@@ -610,7 +616,7 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
// if it is a file then check we have the file in the idempotent
registry
// already
if (Boolean.TRUE.equals(endpoint.isIdempotent())) {
- if (notUnique(file, absoluteFilePath)) {
+ if (notUnique(dynamic, file, absoluteFilePath)) {
return false;
}
}
@@ -621,13 +627,13 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
return endpoint.getInProgressRepository().add(absoluteFilePath);
}
- private boolean notUnique(Supplier<GenericFile<T>> file, String
absoluteFilePath) {
+ private boolean notUnique(Exchange dynamic, Supplier<GenericFile<T>> file,
String absoluteFilePath) {
boolean answer = false;
// use absolute file path as default key, but evaluate if an
// expression key was configured
String key = absoluteFilePath;
if (endpoint.getIdempotentKey() != null) {
- Exchange dummy = endpoint.createExchange(file.get());
+ Exchange dummy = createDummy(dynamic, file);
key = endpoint.getIdempotentKey().evaluate(dummy, String.class);
LOG.trace("Evaluated idempotentKey: {} for file: {}", key, file);
}
@@ -686,6 +692,7 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
* @return <tt>true</tt> if the file is matched,
<tt>false</tt> if not
*/
protected boolean isMatched(
+ Exchange dynamic,
Supplier<GenericFile<T>> file, String name, String
absoluteFilePath,
Supplier<String> relativeFilePath, boolean isDirectory, T[] files)
{
@@ -722,9 +729,8 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
}
if (isDirectory && endpoint.getFilterDirectory() != null) {
- // create a dummy exchange as Exchange is needed for expression
- // evaluation
- Exchange dummy = endpoint.createExchange(file.get());
+ // create a dummy exchange as Exchange is needed for expression
evaluation
+ Exchange dummy = createDummy(dynamic, file);
boolean matches = endpoint.getFilterDirectory().matches(dummy);
if (!matches) {
return false;
@@ -742,7 +748,7 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
if (endpoint.getFileName() != null) {
// create a dummy exchange as Exchange is needed for expression
evaluation
- Exchange dummy = endpoint.createExchange(file.get());
+ Exchange dummy = createDummy(dynamic, file);
String result = evaluateFileExpression(dummy);
if (result != null) {
if (!name.equals(result)) {
@@ -753,7 +759,7 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
if (endpoint.getFilterFile() != null) {
// create a dummy exchange as Exchange is needed for expression
evaluation
- Exchange dummy = endpoint.createExchange(file.get());
+ Exchange dummy = createDummy(dynamic, file);
boolean matches = endpoint.getFilterFile().matches(dummy);
if (!matches) {
return false;
@@ -849,6 +855,20 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
return result;
}
+ protected Exchange createDummy(Exchange dynamic, Supplier<GenericFile<T>>
file) {
+ Exchange dummy = endpoint.createExchange(file.get());
+ if (dynamic != null) {
+ // enrich with data from dynamic source
+ if (dynamic.getMessage().hasHeaders()) {
+ MessageHelper.copyHeaders(dynamic.getMessage(),
dummy.getMessage(), true);
+ if (dynamic.hasVariables()) {
+ dummy.getVariables().putAll(dynamic.getVariables());
+ }
+ }
+ }
+ return dummy;
+ }
+
@SuppressWarnings("unchecked")
private GenericFile<T> getExchangeFileProperty(Exchange exchange) {
return (GenericFile<T>)
exchange.getProperty(ExchangePropertyKey.FILE_EXCHANGE_FILE);
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollDynamicAware.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollDynamicAware.java
new file mode 100644
index 00000000000..0e002f60df3
--- /dev/null
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollDynamicAware.java
@@ -0,0 +1,67 @@
+/*
+ * 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 java.util.Map;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.support.component.PollDynamicAwareSupport;
+import org.apache.camel.util.URISupport;
+
+public abstract class GenericFilePollDynamicAware extends
PollDynamicAwareSupport {
+
+ public static final String PROP_FILE_NAME = "fileName";
+
+ @Override
+ public boolean isLenientProperties() {
+ return false;
+ }
+
+ @Override
+ public DynamicAwareEntry prepare(Exchange exchange, String uri, String
originalUri) throws Exception {
+ Map<String, Object> properties = endpointProperties(exchange, uri);
+ return new DynamicAwareEntry(uri, originalUri, properties, null);
+ }
+
+ @Override
+ public String resolveStaticUri(Exchange exchange, DynamicAwareEntry entry)
throws Exception {
+ String uri = entry.getUri();
+ // windows path problems such as C:\temp was by simple language
evaluated \t as a tab character
+ // which should then be reversed
+ uri = uri.replace("\t", "\\\\t");
+
+ boolean fileName = entry.getProperties().containsKey(PROP_FILE_NAME);
+ // if any of the above are in use, then they should not be pre
evaluated
+ // and we need to rebuild a new uri with them as-is
+ if (fileName) {
+ Map<String, Object> params = entry.getProperties();
+ Map<String, Object> originalParams =
URISupport.parseQuery(URISupport.extractQuery(entry.getOriginalUri()));
+ compute(originalParams, PROP_FILE_NAME, params);
+ return asEndpointUri(exchange, uri, params);
+ } else {
+ return uri;
+ }
+ }
+
+ private static void compute(Map<String, Object> originalParams, String
propFileName, Map<String, Object> params) {
+ Object val = originalParams.get(propFileName);
+ if (val != null) {
+ params.put(propFileName, val.toString());
+ }
+ }
+
+}
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollingConsumer.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollingConsumer.java
index 68c9474b44e..4f29e6358ac 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollingConsumer.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollingConsumer.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.file;
import org.apache.camel.Consumer;
+import org.apache.camel.DynamicPollingConsumer;
import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.spi.PollingConsumerPollStrategy;
@@ -26,7 +27,7 @@ import org.apache.camel.util.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class GenericFilePollingConsumer extends EventDrivenPollingConsumer {
+public class GenericFilePollingConsumer extends EventDrivenPollingConsumer
implements DynamicPollingConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(GenericFilePollingConsumer.class);
private final long delay;
@@ -65,11 +66,11 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer {
}
@Override
- public Exchange receiveNoWait() {
+ public Exchange receiveNoWait(Exchange exchange) {
if (LOG.isTraceEnabled()) {
LOG.trace("receiveNoWait polling file: {}",
getConsumer().getEndpoint());
}
- int polled = doReceive(0);
+ int polled = doReceive(exchange, 0);
if (polled > 0) {
return super.receive(0);
} else {
@@ -78,11 +79,11 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer {
}
@Override
- public Exchange receive() {
+ public Exchange receive(Exchange exchange) {
if (LOG.isTraceEnabled()) {
LOG.trace("receive polling file: {}", getConsumer().getEndpoint());
}
- int polled = doReceive(Long.MAX_VALUE);
+ int polled = doReceive(exchange, Long.MAX_VALUE);
if (polled > 0) {
return super.receive();
} else {
@@ -91,11 +92,11 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer {
}
@Override
- public Exchange receive(long timeout) {
+ public Exchange receive(Exchange exchange, long timeout) {
if (LOG.isTraceEnabled()) {
LOG.trace("receive({}) polling file: {}", timeout,
getConsumer().getEndpoint());
}
- int polled = doReceive(timeout);
+ int polled = doReceive(exchange, timeout);
if (polled > 0) {
return super.receive(timeout);
} else {
@@ -103,7 +104,22 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer {
}
}
- protected int doReceive(long timeout) {
+ @Override
+ public Exchange receiveNoWait() {
+ return receiveNoWait(null);
+ }
+
+ @Override
+ public Exchange receive() {
+ return receive(null);
+ }
+
+ @Override
+ public Exchange receive(long timeout) {
+ return receive(null, timeout);
+ }
+
+ protected int doReceive(Exchange exchange, long timeout) {
int retryCounter = -1;
boolean done = false;
Throwable cause = null;
@@ -130,7 +146,7 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer {
boolean begin = pollStrategy.begin(getConsumer(),
getEndpoint());
if (begin) {
retryCounter++;
- polledMessages = getConsumer().poll();
+ polledMessages = getConsumer().poll(exchange);
LOG.trace("Polled {} messages", polledMessages);
if (polledMessages == 0 && sendEmptyMessageWhenIdle) {
diff --git
a/components/camel-ftp/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/ftp
b/components/camel-ftp/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/ftp
new file mode 100644
index 00000000000..ff02a276d11
--- /dev/null
+++
b/components/camel-ftp/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/ftp
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.file.remote.FtpPollDynamicAware
diff --git
a/components/camel-ftp/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/ftps
b/components/camel-ftp/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/ftps
new file mode 100644
index 00000000000..5797fc23040
--- /dev/null
+++
b/components/camel-ftp/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/ftps
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.file.remote.FtpsPollDynamicAware
diff --git
a/components/camel-ftp/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/sftp
b/components/camel-ftp/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/sftp
new file mode 100644
index 00000000000..e56ca8dc8b3
--- /dev/null
+++
b/components/camel-ftp/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/sftp
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.file.remote.SftpPollDynamicAware
diff --git
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java
index d7f931c019d..2b7c5dced54 100644
---
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java
+++
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpConsumer.java
@@ -94,7 +94,7 @@ public class FtpConsumer extends RemoteFileConsumer<FTPFile> {
}
@Override
- protected boolean pollDirectory(String fileName,
List<GenericFile<FTPFile>> fileList, int depth) {
+ protected boolean pollDirectory(Exchange dynamic, 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
@@ -105,7 +105,7 @@ public class FtpConsumer extends
RemoteFileConsumer<FTPFile> {
// strip trailing slash
fileName = FileUtil.stripTrailingSeparator(fileName);
- boolean answer = doPollDirectory(fileName, null, fileList, depth);
+ boolean answer = doPollDirectory(dynamic, fileName, null, fileList,
depth);
if (currentDir != null) {
operations.changeCurrentDirectory(currentDir);
}
@@ -113,8 +113,9 @@ public class FtpConsumer extends
RemoteFileConsumer<FTPFile> {
return answer;
}
- protected boolean pollSubDirectory(String absolutePath, String dirName,
List<GenericFile<FTPFile>> fileList, int depth) {
- boolean answer = doSafePollSubDirectory(absolutePath, dirName,
fileList, depth);
+ protected boolean pollSubDirectory(
+ Exchange dynamic, String absolutePath, String dirName,
List<GenericFile<FTPFile>> fileList, int depth) {
+ boolean answer = doSafePollSubDirectory(dynamic, absolutePath,
dirName, fileList, depth);
// change back to parent directory when finished polling sub directory
if (isStepwise()) {
operations.changeToParentDirectory();
@@ -123,7 +124,8 @@ public class FtpConsumer extends
RemoteFileConsumer<FTPFile> {
}
@Override
- protected boolean doPollDirectory(String absolutePath, String dirName,
List<GenericFile<FTPFile>> fileList, int depth) {
+ protected boolean doPollDirectory(
+ Exchange dynamic, String absolutePath, String dirName,
List<GenericFile<FTPFile>> fileList, int depth) {
LOG.trace("doPollDirectory from absolutePath: {}, dirName: {}",
absolutePath, dirName);
depth++;
@@ -150,7 +152,7 @@ public class FtpConsumer extends
RemoteFileConsumer<FTPFile> {
}
for (FTPFile file : files) {
- if (handleFtpEntries(absolutePath, fileList, depth, files, file)) {
+ if (handleFtpEntries(dynamic, absolutePath, fileList, depth,
files, file)) {
return false;
}
}
@@ -159,6 +161,7 @@ public class FtpConsumer extends
RemoteFileConsumer<FTPFile> {
}
private boolean handleFtpEntries(
+ Exchange dynamic,
String absolutePath, List<GenericFile<FTPFile>> fileList, int
depth, FTPFile[] files, FTPFile file) {
if (LOG.isTraceEnabled()) {
LOG.trace("FtpFile[name={}, dir={}, file={}]", file.getName(),
file.isDirectory(), file.isFile());
@@ -170,11 +173,11 @@ public class FtpConsumer extends
RemoteFileConsumer<FTPFile> {
}
if (file.isDirectory()) {
- if (handleDirectory(absolutePath, fileList, depth, files, file)) {
+ if (handleDirectory(dynamic, absolutePath, fileList, depth, files,
file)) {
return true;
}
} else if (file.isFile()) {
- handleFile(absolutePath, fileList, depth, files, file);
+ handleFile(dynamic, absolutePath, fileList, depth, files, file);
} else {
LOG.debug("Ignoring unsupported remote file type: {}", file);
}
@@ -182,6 +185,7 @@ public class FtpConsumer extends
RemoteFileConsumer<FTPFile> {
}
private boolean handleDirectory(
+ Exchange dynamic,
String absolutePath, List<GenericFile<FTPFile>> fileList, int
depth, FTPFile[] files, FTPFile file) {
if (endpoint.isRecursive() && depth < endpoint.getMaxDepth()) {
// calculate the absolute file path using util class
@@ -190,11 +194,11 @@ public class FtpConsumer extends
RemoteFileConsumer<FTPFile> {
Supplier<GenericFile<FTPFile>> remote
= Suppliers.memorize(() -> asRemoteFile(absolutePath,
absoluteFilePath, file, getEndpoint().getCharset()));
Supplier<String> relativePath = getRelativeFilePath(endpointPath,
null, absolutePath, file);
- if (isValidFile(remote, file.getName(), absoluteFilePath,
relativePath, true, files)) {
+ if (isValidFile(dynamic, remote, file.getName(), absoluteFilePath,
relativePath, true, files)) {
// recursive scan and add the sub files and folders
String subDirectory = file.getName();
String path = ObjectHelper.isNotEmpty(absolutePath) ?
absolutePath + "/" + subDirectory : subDirectory;
- boolean canPollMore = pollSubDirectory(path, subDirectory,
fileList, depth);
+ boolean canPollMore = pollSubDirectory(dynamic, path,
subDirectory, fileList, depth);
if (!canPollMore) {
return true;
}
@@ -204,6 +208,7 @@ public class FtpConsumer extends
RemoteFileConsumer<FTPFile> {
}
private void handleFile(
+ Exchange dynamic,
String absolutePath, List<GenericFile<FTPFile>> fileList, int
depth, FTPFile[] files, FTPFile file) {
if (depth >= endpoint.getMinDepth()) {
// calculate the absolute file path using util class
@@ -212,7 +217,7 @@ public class FtpConsumer extends
RemoteFileConsumer<FTPFile> {
Supplier<GenericFile<FTPFile>> remote
= Suppliers.memorize(() -> asRemoteFile(absolutePath,
absoluteFilePath, file, getEndpoint().getCharset()));
Supplier<String> relativePath = getRelativeFilePath(endpointPath,
null, absolutePath, file);
- if (isValidFile(remote, file.getName(), absoluteFilePath,
relativePath, false, files)) {
+ if (isValidFile(dynamic, remote, file.getName(), absoluteFilePath,
relativePath, false, files)) {
// matched file so add
fileList.add(remote.get());
}
diff --git
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpPollDynamicAware.java
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpPollDynamicAware.java
new file mode 100644
index 00000000000..4311acd528c
--- /dev/null
+++
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpPollDynamicAware.java
@@ -0,0 +1,24 @@
+/*
+ * 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.remote;
+
+import org.apache.camel.component.file.GenericFilePollDynamicAware;
+import org.apache.camel.spi.annotations.PollDynamic;
+
+@PollDynamic("ftp")
+public class FtpPollDynamicAware extends GenericFilePollDynamicAware {
+}
diff --git
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpsPollDynamicAware.java
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpsPollDynamicAware.java
new file mode 100644
index 00000000000..3693f994150
--- /dev/null
+++
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpsPollDynamicAware.java
@@ -0,0 +1,24 @@
+/*
+ * 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.remote;
+
+import org.apache.camel.component.file.GenericFilePollDynamicAware;
+import org.apache.camel.spi.annotations.PollDynamic;
+
+@PollDynamic("ftps")
+public class FtpsPollDynamicAware extends GenericFilePollDynamicAware {
+}
diff --git
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
index 0b268ae4ee3..0031178e20f 100644
---
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
+++
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
@@ -251,11 +251,12 @@ public abstract class RemoteFileConsumer<T> extends
GenericFileConsumer<T> {
* maxMessagesPerPoll limit
has been hit
* @throws GenericFileOperationFailedException if the exception during
doPollDirectory can not be ignored
*/
- protected boolean doSafePollSubDirectory(String absolutePath, String
dirName, List<GenericFile<T>> fileList, int depth) {
+ protected boolean doSafePollSubDirectory(
+ Exchange dynamic, String absolutePath, String dirName,
List<GenericFile<T>> fileList, int depth) {
try {
LOG.trace("Polling sub directory: {} from: {}", absolutePath,
endpoint);
// Try to poll the directory
- return doPollDirectory(absolutePath, dirName, fileList, depth);
+ return doPollDirectory(dynamic, absolutePath, dirName, fileList,
depth);
} catch (Exception e) {
LOG.debug("Caught exception {}", e.getMessage());
if (ignoreCannotRetrieveFile(absolutePath, null, e)) {
@@ -285,5 +286,6 @@ public abstract class RemoteFileConsumer<T> extends
GenericFileConsumer<T> {
* @return whether or not to continue polling, <tt>false</tt>
means the maxMessagesPerPoll limit has
* been hit
*/
- protected abstract boolean doPollDirectory(String absolutePath, String
dirName, List<GenericFile<T>> fileList, int depth);
+ protected abstract boolean doPollDirectory(
+ Exchange dynamic, String absolutePath, String dirName,
List<GenericFile<T>> fileList, int depth);
}
diff --git
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java
index d231d6a3fd8..d27e336ba6c 100644
---
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java
+++
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpConsumer.java
@@ -86,7 +86,7 @@ public class SftpConsumer extends
RemoteFileConsumer<SftpRemoteFile> {
}
@Override
- protected boolean pollDirectory(String fileName,
List<GenericFile<SftpRemoteFile>> fileList, int depth) {
+ protected boolean pollDirectory(Exchange dynamic, String fileName,
List<GenericFile<SftpRemoteFile>> fileList, int depth) {
String currentDir = null;
if (isStepwise()) {
// must remember current dir so we stay in that directory after the
@@ -97,7 +97,7 @@ public class SftpConsumer extends
RemoteFileConsumer<SftpRemoteFile> {
// strip trailing slash
fileName = FileUtil.stripTrailingSeparator(fileName);
- boolean answer = doPollDirectory(fileName, null, fileList, depth);
+ boolean answer = doPollDirectory(dynamic, fileName, null, fileList,
depth);
if (currentDir != null) {
operations.changeCurrentDirectory(currentDir);
}
@@ -106,8 +106,9 @@ public class SftpConsumer extends
RemoteFileConsumer<SftpRemoteFile> {
}
protected boolean pollSubDirectory(
+ Exchange dynamic,
String absolutePath, String dirName,
List<GenericFile<SftpRemoteFile>> fileList, int depth) {
- boolean answer = doSafePollSubDirectory(absolutePath, dirName,
fileList, depth);
+ boolean answer = doSafePollSubDirectory(dynamic, absolutePath,
dirName, fileList, depth);
// change back to parent directory when finished polling sub directory
if (isStepwise()) {
operations.changeToParentDirectory();
@@ -117,6 +118,7 @@ public class SftpConsumer extends
RemoteFileConsumer<SftpRemoteFile> {
@Override
protected boolean doPollDirectory(
+ Exchange dynamic,
String absolutePath, String dirName,
List<GenericFile<SftpRemoteFile>> fileList, int depth) {
LOG.trace("doPollDirectory from absolutePath: {}, dirName: {}",
absolutePath, dirName);
@@ -168,11 +170,11 @@ public class SftpConsumer extends
RemoteFileConsumer<SftpRemoteFile> {
= Suppliers.memorize(
() -> asRemoteFile(absolutePath,
absoluteFilePath, file, getEndpoint().getCharset()));
Supplier<String> relativePath =
getRelativeFilePath(endpointPath, null, absolutePath, file);
- if (isValidFile(remote, file.getFilename(),
absoluteFilePath, relativePath, true, files)) {
+ if (isValidFile(dynamic, remote, file.getFilename(),
absoluteFilePath, relativePath, true, files)) {
// recursive scan and add the sub files and folders
String subDirectory = file.getFilename();
String path = ObjectHelper.isNotEmpty(absolutePath) ?
absolutePath + "/" + subDirectory : subDirectory;
- boolean canPollMore = pollSubDirectory(path,
subDirectory, fileList, depth);
+ boolean canPollMore = pollSubDirectory(dynamic, path,
subDirectory, fileList, depth);
if (!canPollMore) {
return false;
}
@@ -188,7 +190,7 @@ public class SftpConsumer extends
RemoteFileConsumer<SftpRemoteFile> {
= Suppliers.memorize(
() -> asRemoteFile(absolutePath,
absoluteFilePath, file, getEndpoint().getCharset()));
Supplier<String> relativePath =
getRelativeFilePath(endpointPath, null, absolutePath, file);
- if (isValidFile(remote, file.getFilename(),
absoluteFilePath, relativePath, false, files)) {
+ if (isValidFile(dynamic, remote, file.getFilename(),
absoluteFilePath, relativePath, false, files)) {
// matched file so add
fileList.add(remote.get());
}
diff --git
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpPollDynamicAware.java
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpPollDynamicAware.java
new file mode 100644
index 00000000000..14117e61383
--- /dev/null
+++
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpPollDynamicAware.java
@@ -0,0 +1,24 @@
+/*
+ * 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.remote;
+
+import org.apache.camel.component.file.GenericFilePollDynamicAware;
+import org.apache.camel.spi.annotations.PollDynamic;
+
+@PollDynamic("sftp")
+public class SftpPollDynamicAware extends GenericFilePollDynamicAware {
+}
diff --git
a/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/RemoteFileIgnoreDoPollErrorTest.java
b/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/RemoteFileIgnoreDoPollErrorTest.java
index 780d3d5f1a6..bd6e173712f 100644
---
a/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/RemoteFileIgnoreDoPollErrorTest.java
+++
b/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/RemoteFileIgnoreDoPollErrorTest.java
@@ -72,14 +72,14 @@ public class RemoteFileIgnoreDoPollErrorTest {
@Test
public void testReadDirErrorIsHandled() {
RemoteFileConsumer<Object> consumer = getRemoteFileConsumer("true",
true);
- boolean result = consumer.doSafePollSubDirectory("anyPath", "adir",
new ArrayList<>(), 0);
+ boolean result = consumer.doSafePollSubDirectory(null, "anyPath",
"adir", new ArrayList<>(), 0);
assertTrue(result);
}
@Test
public void testReadDirErrorIsHandledWithNoMorePoll() {
RemoteFileConsumer<Object> consumer = getRemoteFileConsumer("false",
true);
- boolean result = consumer.doSafePollSubDirectory("anyPath", "adir",
new ArrayList<>(), 0);
+ boolean result = consumer.doSafePollSubDirectory(null, "anyPath",
"adir", new ArrayList<>(), 0);
assertFalse(result);
}
@@ -89,7 +89,7 @@ public class RemoteFileIgnoreDoPollErrorTest {
List<GenericFile<Object>> list = Collections.emptyList();
Exception ex = assertThrows(GenericFileOperationFailedException.class,
- () -> consumer.doSafePollSubDirectory("anyPath", "adir", list,
0));
+ () -> consumer.doSafePollSubDirectory(null, "anyPath", "adir",
list, 0));
assertInstanceOf(IllegalStateException.class, ex.getCause());
}
@@ -100,7 +100,7 @@ public class RemoteFileIgnoreDoPollErrorTest {
List<GenericFile<Object>> list = Collections.emptyList();
Exception ex = assertThrows(GenericFileOperationFailedException.class,
- () -> consumer.doSafePollSubDirectory("anyPath", "adir", list,
0));
+ () -> consumer.doSafePollSubDirectory(null, "anyPath", "adir",
list, 0));
assertNull(ex.getCause());
}
@@ -113,6 +113,7 @@ public class RemoteFileIgnoreDoPollErrorTest {
return new RemoteFileConsumer<>(remoteFileEndpoint, null, null, null) {
@Override
protected boolean doPollDirectory(
+ Exchange dynamic,
String absolutePath, String dirName,
List<GenericFile<Object>> genericFiles, int depth) {
if ("IllegalStateException".equals(doPollResult)) {
throw new IllegalStateException("Problem");
@@ -124,7 +125,8 @@ public class RemoteFileIgnoreDoPollErrorTest {
}
@Override
- protected boolean pollDirectory(String fileName,
List<GenericFile<Object>> genericFiles, int depth) {
+ protected boolean pollDirectory(
+ Exchange dynamic, String fileName,
List<GenericFile<Object>> genericFiles, int depth) {
return false;
}
diff --git
a/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/integration/PollDynamicFileNameTest.java
b/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/integration/PollDynamicFileNameTest.java
new file mode 100644
index 00000000000..ff8e5abf473
--- /dev/null
+++
b/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/integration/PollDynamicFileNameTest.java
@@ -0,0 +1,105 @@
+/*
+ * 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.remote.integration;
+
+import java.io.File;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class PollDynamicFileNameTest extends FtpServerTestSupport {
+
+ protected String getFtpUrl() {
+ return
"ftp://admin@localhost:{{ftp.server.port}}/pollfile?password=admin&binary=false&noop=true";;
+ }
+
+ @Override
+ public void doPostSetup() throws Exception {
+ prepareFtpServer();
+ }
+
+ @Test
+ public void testPollEnrichFileOne() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(2);
+ getMockEndpoint("mock:result").message(0).body().isEqualTo("Hello
World");
+ getMockEndpoint("mock:result").message(1).body().isNull();
+
+ template.sendBodyAndHeader("direct:start", "Foo", "target",
"myfile.txt");
+ template.sendBodyAndHeader("direct:start", "Bar", "target",
"unknown.txt");
+
+ MockEndpoint.assertIsSatisfied(context);
+
+ // there should only be 1 file endpoint
+ long c = context.getEndpoints().stream()
+ .filter(e -> e.getEndpointKey().startsWith("ftp") &&
e.getEndpointUri().contains("&fileName=")).count();
+ Assertions.assertEquals(1, c, "There should only be 1 ftp endpoint");
+ }
+
+ @Test
+ public void testPollEnrichFileTwo() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceivedInAnyOrder("Hello
World", "Bye World");
+
+ template.sendBodyAndHeader(getFtpUrl(), "Bye World",
Exchange.FILE_NAME, "myfile2.txt");
+
+ template.sendBodyAndHeader("direct:start", "Foo", "target",
"myfile.txt");
+ template.sendBodyAndHeader("direct:start", "Bar", "target",
"myfile2.txt");
+
+ MockEndpoint.assertIsSatisfied(context);
+
+ // there should only be 1 file endpoint
+ long c = context.getEndpoints().stream()
+ .filter(e -> e.getEndpointKey().startsWith("ftp") &&
e.getEndpointUri().contains("&fileName=")).count();
+ Assertions.assertEquals(1, c, "There should only be 1 ftp endpoint");
+ }
+
+ private void prepareFtpServer() throws Exception {
+ // prepares the FTP Server by creating a file on the server that we
want
+ // to unit test that we can pool and store as a local file
+ Endpoint endpoint = context.getEndpoint(getFtpUrl());
+
+ Exchange exchange = endpoint.createExchange();
+ exchange.getIn().setBody("Hello World");
+ exchange.getIn().setHeader(Exchange.FILE_NAME, "myfile.txt");
+
+ Producer producer = endpoint.createProducer();
+ producer.start();
+ producer.process(exchange);
+ producer.stop();
+
+ // assert files is created
+ File file = service.ftpFile("pollfile/myfile.txt").toFile();
+ assertTrue(file.exists(), "The file should exists");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:start")
+ .poll(getFtpUrl() + "&fileName=${header.target}", 2000)
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/components/camel-jsch/src/generated/resources/META-INF/services/org/apache/camel/send-dynamic/scp
b/components/camel-jsch/src/generated/resources/META-INF/services/org/apache/camel/send-dynamic/scp
new file mode 100644
index 00000000000..5509e2820d1
--- /dev/null
+++
b/components/camel-jsch/src/generated/resources/META-INF/services/org/apache/camel/send-dynamic/scp
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.scp.ScpSendDynamicAware
diff --git
a/components/camel-jsch/src/main/java/org/apache/camel/component/scp/ScpSendDynamicAware.java
b/components/camel-jsch/src/main/java/org/apache/camel/component/scp/ScpSendDynamicAware.java
new file mode 100644
index 00000000000..e0293b480eb
--- /dev/null
+++
b/components/camel-jsch/src/main/java/org/apache/camel/component/scp/ScpSendDynamicAware.java
@@ -0,0 +1,24 @@
+/*
+ * 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.scp;
+
+import org.apache.camel.component.file.GenericFileSendDynamicAware;
+import org.apache.camel.spi.annotations.SendDynamic;
+
+@SendDynamic("scp")
+public class ScpSendDynamicAware extends GenericFileSendDynamicAware {
+}
diff --git
a/components/camel-smb/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/smb
b/components/camel-smb/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/smb
new file mode 100644
index 00000000000..913461c58fd
--- /dev/null
+++
b/components/camel-smb/src/generated/resources/META-INF/services/org/apache/camel/poll-dynamic/smb
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.smb.SmbPollDynamicAware
diff --git
a/components/camel-smb/src/main/java/org/apache/camel/component/smb/SmbConsumer.java
b/components/camel-smb/src/main/java/org/apache/camel/component/smb/SmbConsumer.java
index d7131ff6cbe..2ae5bab54ff 100644
---
a/components/camel-smb/src/main/java/org/apache/camel/component/smb/SmbConsumer.java
+++
b/components/camel-smb/src/main/java/org/apache/camel/component/smb/SmbConsumer.java
@@ -68,7 +68,8 @@ public class SmbConsumer extends
GenericFileConsumer<FileIdBothDirectoryInformat
}
@Override
- protected boolean pollDirectory(String path,
List<GenericFile<FileIdBothDirectoryInformation>> fileList, int depth) {
+ protected boolean pollDirectory(
+ Exchange dynamic, String path,
List<GenericFile<FileIdBothDirectoryInformation>> fileList, int depth) {
depth++;
path = (path == null) ? "" : path;
FileIdBothDirectoryInformation[] files = getSmbFiles(path);
@@ -96,7 +97,7 @@ public class SmbConsumer extends
GenericFileConsumer<FileIdBothDirectoryInformat
= path + (path.endsWith("/") ? "" : "/") +
file.getFileName();
}
- if (handleSmbEntries(fullFilePath, fileList, depth, files, file)) {
+ if (handleSmbEntries(dynamic, fullFilePath, fileList, depth,
files, file)) {
return false;
}
}
@@ -109,20 +110,22 @@ public class SmbConsumer extends
GenericFileConsumer<FileIdBothDirectoryInformat
}
private boolean handleSmbEntries(
+ Exchange dynamic,
String fullFilePath,
List<GenericFile<FileIdBothDirectoryInformation>> fileList, int depth,
FileIdBothDirectoryInformation[] files,
FileIdBothDirectoryInformation file) {
if (isDirectory(file)) {
LOG.trace("SmbFile[name={}, dir=true]", file.getFileName());
- return handleDirectory(fullFilePath, fileList, depth, files, file);
+ return handleDirectory(dynamic, fullFilePath, fileList, depth,
files, file);
} else {
LOG.trace("SmbFile[name={}, file=true]", file.getFileName());
- handleFile(fullFilePath, fileList, depth, files, file);
+ handleFile(dynamic, fullFilePath, fileList, depth, files, file);
}
return false;
}
private boolean handleDirectory(
+ Exchange dynamic,
String fullFilePath,
List<GenericFile<FileIdBothDirectoryInformation>> fileList, int depth,
FileIdBothDirectoryInformation[] files,
FileIdBothDirectoryInformation file) {
@@ -130,10 +133,10 @@ public class SmbConsumer extends
GenericFileConsumer<FileIdBothDirectoryInformat
SmbFile smbFile = asGenericFile(fullFilePath, file,
getEndpoint().getCharset());
Supplier<GenericFile<FileIdBothDirectoryInformation>>
genericFileSupplier = Suppliers.memorize(() -> smbFile);
Supplier<String> relativePath = smbFile::getRelativeFilePath;
- if (isValidFile(genericFileSupplier, file.getFileName(),
+ if (isValidFile(dynamic, genericFileSupplier, file.getFileName(),
smbFile.getAbsoluteFilePath(), relativePath, true, files))
{
// recursive scan and add the sub files and folders
- boolean canPollMore = pollDirectory(fullFilePath, fileList,
depth);
+ boolean canPollMore = pollDirectory(dynamic, fullFilePath,
fileList, depth);
return !canPollMore;
}
}
@@ -141,6 +144,7 @@ public class SmbConsumer extends
GenericFileConsumer<FileIdBothDirectoryInformat
}
private void handleFile(
+ Exchange dynamic,
String fullFilePath,
List<GenericFile<FileIdBothDirectoryInformation>> fileList, int depth,
FileIdBothDirectoryInformation[] files,
FileIdBothDirectoryInformation file) {
@@ -149,7 +153,7 @@ public class SmbConsumer extends
GenericFileConsumer<FileIdBothDirectoryInformat
Supplier<GenericFile<FileIdBothDirectoryInformation>>
genericFileSupplier = Suppliers.memorize(() -> smbFile);
Supplier<String> relativePath = smbFile::getRelativeFilePath;
- if (isValidFile(genericFileSupplier, file.getFileName(),
+ if (isValidFile(dynamic, genericFileSupplier, file.getFileName(),
smbFile.getAbsoluteFilePath(), relativePath, false,
files)) {
fileList.add(smbFile);
}
diff --git
a/components/camel-smb/src/main/java/org/apache/camel/component/smb/SmbPollDynamicAware.java
b/components/camel-smb/src/main/java/org/apache/camel/component/smb/SmbPollDynamicAware.java
new file mode 100644
index 00000000000..9a6dd8b186c
--- /dev/null
+++
b/components/camel-smb/src/main/java/org/apache/camel/component/smb/SmbPollDynamicAware.java
@@ -0,0 +1,24 @@
+/*
+ * 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.smb;
+
+import org.apache.camel.component.file.GenericFilePollDynamicAware;
+import org.apache.camel.spi.annotations.PollDynamic;
+
+@PollDynamic("smb")
+public class SmbPollDynamicAware extends GenericFilePollDynamicAware {
+}
diff --git
a/components/camel-smb/src/test/java/org/apache/camel/component/smb/PollDynamicFileNameTest.java
b/components/camel-smb/src/test/java/org/apache/camel/component/smb/PollDynamicFileNameTest.java
new file mode 100644
index 00000000000..2fd7f496e74
--- /dev/null
+++
b/components/camel-smb/src/test/java/org/apache/camel/component/smb/PollDynamicFileNameTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.smb;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class PollDynamicFileNameTest extends SmbServerTestSupport {
+
+ protected String getSmbUrl() {
+ return String.format(
+ "smb:%s/%s?username=%s&password=%s&path=/noop&noop=true",
+ service.address(), service.shareName(), service.userName(),
service.password());
+ }
+
+ @Override
+ public void doPostSetup() throws Exception {
+ prepareSmbServer();
+ }
+
+ @Test
+ public void testPollEnrichFileOne() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(2);
+ getMockEndpoint("mock:result").message(0).body().isEqualTo("Hello
World");
+ getMockEndpoint("mock:result").message(1).body().isNull();
+
+ template.sendBodyAndHeader("direct:start", "Foo", "target",
"myfile.txt");
+ template.sendBodyAndHeader("direct:start", "Bar", "target",
"unknown.txt");
+
+ MockEndpoint.assertIsSatisfied(context);
+
+ // there should only be 1 file endpoint
+ long c = context.getEndpoints().stream()
+ .filter(e -> e.getEndpointKey().startsWith("smb") &&
e.getEndpointUri().contains("?fileName=")).count();
+ Assertions.assertEquals(1, c, "There should only be 1 smb endpoint");
+ }
+
+ @Test
+ public void testPollEnrichFileTwo() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceivedInAnyOrder("Hello
World", "Bye World");
+
+ template.sendBodyAndHeader(getSmbUrl(), "Bye World",
Exchange.FILE_NAME, "myfile2.txt");
+
+ template.sendBodyAndHeader("direct:start", "Foo", "target",
"myfile.txt");
+ template.sendBodyAndHeader("direct:start", "Bar", "target",
"myfile2.txt");
+
+ MockEndpoint.assertIsSatisfied(context);
+
+ // there should only be 1 file endpoint
+ long c = context.getEndpoints().stream()
+ .filter(e -> e.getEndpointKey().startsWith("smb") &&
e.getEndpointUri().contains("?fileName=")).count();
+ Assertions.assertEquals(1, c, "There should only be 1 smb endpoint");
+ }
+
+ private void prepareSmbServer() throws Exception {
+ // prepares the smb Server by creating a file on the server that we
want
+ // to unit test that we can pool and store as a local file
+ Endpoint endpoint = context.getEndpoint(getSmbUrl());
+ Exchange exchange = endpoint.createExchange();
+ exchange.getIn().setBody("Hello World");
+ exchange.getIn().setHeader(Exchange.FILE_NAME, "myfile.txt");
+
+ Producer producer = endpoint.createProducer();
+ producer.start();
+ producer.process(exchange);
+ producer.stop();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:start")
+ .poll(getSmbUrl() + "&fileName=${header.target}", 2000)
+ .to("mock:result");
+ }
+ };
+ }
+
+}
diff --git
a/core/camel-api/src/generated/java/org/apache/camel/spi/annotations/PollDynamic.java
b/core/camel-api/src/generated/java/org/apache/camel/spi/annotations/PollDynamic.java
new file mode 100644
index 00000000000..fd5325c72e4
--- /dev/null
+++
b/core/camel-api/src/generated/java/org/apache/camel/spi/annotations/PollDynamic.java
@@ -0,0 +1,33 @@
+/*
+ * 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.spi.annotations;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Target({ ElementType.TYPE })
+@ServiceFactory("poll-dynamic")
+public @interface PollDynamic {
+
+ String value();
+
+}
diff --git
a/core/camel-api/src/main/java/org/apache/camel/DynamicPollingConsumer.java
b/core/camel-api/src/main/java/org/apache/camel/DynamicPollingConsumer.java
new file mode 100644
index 00000000000..c939c607375
--- /dev/null
+++ b/core/camel-api/src/main/java/org/apache/camel/DynamicPollingConsumer.java
@@ -0,0 +1,65 @@
+/*
+ * 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;
+
+/**
+ * A {@link PollingConsumer} that are used by dynamic Poll and PollEnrich EIPs
to facilitate components that can use
+ * information from the current {@link Exchange} during the poll.
+ */
+public interface DynamicPollingConsumer extends PollingConsumer {
+
+ /**
+ * Waits until a message is available and then returns it. Warning that
this method could block indefinitely if no
+ * messages are available.
+ * <p/>
+ * Will return <tt>null</tt> if the consumer is not started
+ * <p/>
+ * <b>Important: </b> See the class javadoc about the need for done the
{@link org.apache.camel.spi.UnitOfWork} on
+ * the returned {@link Exchange}
+ *
+ * @param exchange the current exchange
+ *
+ * @return the message exchange received.
+ */
+ Exchange receive(Exchange exchange);
+
+ /**
+ * Attempts to receive a message exchange immediately without waiting and
returning <tt>null</tt> if a message
+ * exchange is not available yet.
+ * <p/>
+ * <b>Important: </b> See the class javadoc about the need for done the
{@link org.apache.camel.spi.UnitOfWork} on
+ * the returned {@link Exchange}
+ *
+ * @return the message exchange if one is immediately available otherwise
<tt>null</tt>
+ */
+ Exchange receiveNoWait(Exchange exchange);
+
+ /**
+ * Attempts to receive a message exchange, waiting up to the given timeout
to expire if a message is not yet
+ * available.
+ * <p/>
+ * <b>Important: </b> See the class javadoc about the need for done the
{@link org.apache.camel.spi.UnitOfWork} on
+ * the returned {@link Exchange}
+ *
+ * @param timeout the amount of time in milliseconds to wait for a
message before timing out and returning
+ * <tt>null</tt>
+ *
+ * @return the message exchange if one was available within the
timeout period, or <tt>null</tt> if the
+ * timeout expired
+ */
+ Exchange receive(Exchange exchange, long timeout);
+}
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/SendDynamicAware.java
b/core/camel-api/src/main/java/org/apache/camel/spi/PollDynamicAware.java
similarity index 63%
copy from
core/camel-api/src/main/java/org/apache/camel/spi/SendDynamicAware.java
copy to core/camel-api/src/main/java/org/apache/camel/spi/PollDynamicAware.java
index 09e374d076e..9e6b22d6cd3 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/SendDynamicAware.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/PollDynamicAware.java
@@ -21,16 +21,14 @@ import java.util.Map;
import org.apache.camel.CamelContextAware;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.Producer;
import org.apache.camel.Service;
/**
- * Used for components that can optimise the usage of {@link
org.apache.camel.processor.SendDynamicProcessor} (toD) to
- * reuse a static {@link org.apache.camel.Endpoint} and {@link Producer} that
supports using headers to provide the
- * dynamic parts. For example many of the HTTP components supports this.
+ * Used for components that can optimise the usage of {@link
org.apache.camel.processor.PollProcessor} (poll/pollEnrich)
+ * to reuse a static {@link Endpoint} and {@link
org.apache.camel.DynamicPollingConsumer} that supports using headers to
+ * provide the dynamic parts. For example many of the file based components
supports this.
*/
-public interface SendDynamicAware extends Service, CamelContextAware {
+public interface PollDynamicAware extends Service, CamelContextAware {
/**
* Sets the component name.
@@ -45,7 +43,7 @@ public interface SendDynamicAware extends Service,
CamelContextAware {
String getScheme();
/**
- * Whether to traverses the given parameters, and resolve any parameter
values which uses the RAW token syntax:
+ * Whether to traverse the given parameters, and resolve any parameter
values which uses the RAW token syntax:
* <tt>key=RAW(value)</tt>. And then remove the RAW tokens, and replace
the content of the value, with just the
* value.
*/
@@ -61,8 +59,8 @@ public interface SendDynamicAware extends Service,
CamelContextAware {
boolean isLenientProperties();
/**
- * An entry of detailed information from the recipient uri, which allows
the {@link SendDynamicAware} implementation
- * to prepare pre- and post- processor and the static uri to be used for
the optimised dynamic to.
+ * An entry of detailed information from the recipient uri, which allows
the {@link PollDynamicAware} implementation
+ * to prepare the static uri to be used for the optimised poll.
*/
class DynamicAwareEntry {
@@ -97,8 +95,7 @@ public interface SendDynamicAware extends Service,
CamelContextAware {
}
/**
- * Prepares for using optimised dynamic to by parsing the uri and
returning an entry of details that are used for
- * creating the pre and post processors, and the static uri.
+ * Prepares for using optimised dynamic polling consumer by parsing the
uri and returning an entry of details.
*
* @param exchange the exchange
* @param uri the resolved uri which is intended to be used
@@ -109,36 +106,14 @@ public interface SendDynamicAware extends Service,
CamelContextAware {
DynamicAwareEntry prepare(Exchange exchange, String uri, String
originalUri) throws Exception;
/**
- * Resolves the static part of the uri that are used for creating a single
{@link org.apache.camel.Endpoint} and
- * {@link Producer} that will be reused for processing the optimised toD.
+ * Resolves the static part of the uri that are used for creating a single
{@link Endpoint} and
+ * {@link org.apache.camel.DynamicPollingConsumer} that will be reused for
processing the optimised poll/pollEnrich.
*
* @param exchange the exchange
* @param entry prepared information about the dynamic endpoint to use
- * @return the static uri, or <tt>null</tt> to not let toD use
this optimisation.
+ * @return the static uri, or <tt>null</tt> to not let
poll/pollEnrich use this optimisation.
* @throws Exception is thrown if error resolving the static uri.
*/
String resolveStaticUri(Exchange exchange, DynamicAwareEntry entry) throws
Exception;
- /**
- * Creates the pre {@link Processor} that will prepare the {@link
Exchange} with dynamic details from the given
- * recipient.
- *
- * @param exchange the exchange
- * @param entry prepared information about the dynamic endpoint to use
- * @return the processor, or <tt>null</tt> to not let toD use
this optimisation.
- * @throws Exception is thrown if error creating the pre processor.
- */
- Processor createPreProcessor(Exchange exchange, DynamicAwareEntry entry)
throws Exception;
-
- /**
- * Creates an optional post {@link Processor} that will be executed
afterwards when the message has been sent
- * dynamic.
- *
- * @param exchange the exchange
- * @param entry prepared information about the dynamic endpoint to use
- * @return the post processor, or <tt>null</tt> if no post
processor is needed.
- * @throws Exception is thrown if error creating the post processor.
- */
- Processor createPostProcessor(Exchange exchange, DynamicAwareEntry entry)
throws Exception;
-
}
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/SendDynamicAware.java
b/core/camel-api/src/main/java/org/apache/camel/spi/SendDynamicAware.java
index 09e374d076e..5c48df877ab 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/SendDynamicAware.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/SendDynamicAware.java
@@ -45,7 +45,7 @@ public interface SendDynamicAware extends Service,
CamelContextAware {
String getScheme();
/**
- * Whether to traverses the given parameters, and resolve any parameter
values which uses the RAW token syntax:
+ * Whether to traverse the given parameters, and resolve any parameter
values which uses the RAW token syntax:
* <tt>key=RAW(value)</tt>. And then remove the RAW tokens, and replace
the content of the value, with just the
* value.
*/
@@ -98,7 +98,7 @@ public interface SendDynamicAware extends Service,
CamelContextAware {
/**
* Prepares for using optimised dynamic to by parsing the uri and
returning an entry of details that are used for
- * creating the pre and post processors, and the static uri.
+ * creating the pre- and post-processors, and the static uri.
*
* @param exchange the exchange
* @param uri the resolved uri which is intended to be used
diff --git
a/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/pollEnrich.json
b/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/pollEnrich.json
index 85409de3ace..5644c4a8a54 100644
---
a/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/pollEnrich.json
+++
b/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/pollEnrich.json
@@ -24,7 +24,8 @@
"timeout": { "index": 9, "kind": "attribute", "displayName": "Timeout",
"group": "common", "required": false, "type": "duration", "javaType":
"java.lang.String", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "-1", "description": "Timeout in millis when polling from the
external service. The timeout has influence about the poll enrich behavior. It
basically operations in three different modes: negative value - Waits until a
message is available and then ret [...]
"cacheSize": { "index": 10, "kind": "attribute", "displayName": "Cache
Size", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired":
false, "secret": false, "description": "Sets the maximum size used by the
org.apache.camel.spi.ConsumerCache which is used to cache and reuse consumers
when uris are reused. Beware that when using dynamic endpoints then it affects
how well the cache can be utiliz [...]
"ignoreInvalidEndpoint": { "index": 11, "kind": "attribute",
"displayName": "Ignore Invalid Endpoint", "group": "advanced", "label":
"advanced", "required": false, "type": "boolean", "javaType":
"java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Ignore the invalidate endpoint exception
when try to create a producer with that endpoint" },
- "autoStartComponents": { "index": 12, "kind": "attribute", "displayName":
"Auto Start Components", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Whether to auto startup components when poll enricher is starting up." }
+ "allowOptimisedComponents": { "index": 12, "kind": "attribute",
"displayName": "Allow Optimised Components", "group": "advanced", "label":
"advanced", "required": false, "type": "boolean", "javaType":
"java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": true, "description": "Whether to allow components to optimise
if they are org.apache.camel.spi.SendDynamicAware ." },
+ "autoStartComponents": { "index": 13, "kind": "attribute", "displayName":
"Auto Start Components", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Whether to auto startup components when poll enricher is starting up." }
},
"exchangeProperties": {
"CamelToEndpoint": { "index": 0, "kind": "exchangeProperty",
"displayName": "To Endpoint", "label": "producer", "required": false,
"javaType": "String", "deprecated": false, "autowired": false, "secret": false,
"description": "Endpoint URI where this Exchange is being sent to" }
diff --git
a/core/camel-core-model/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
b/core/camel-core-model/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
index db1cb2ac36a..8ba0f887054 100644
---
a/core/camel-core-model/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
+++
b/core/camel-core-model/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
@@ -65,6 +65,9 @@ public class PollEnrichDefinition extends ExpressionNode
private String ignoreInvalidEndpoint;
@XmlAttribute
@Metadata(label = "advanced", defaultValue = "true", javaType =
"java.lang.Boolean")
+ private String allowOptimisedComponents;
+ @XmlAttribute
+ @Metadata(label = "advanced", defaultValue = "true", javaType =
"java.lang.Boolean")
private String autoStartComponents;
public PollEnrichDefinition() {
@@ -86,6 +89,7 @@ public class PollEnrichDefinition extends ExpressionNode
this.timeout = source.timeout;
this.cacheSize = source.cacheSize;
this.ignoreInvalidEndpoint = source.ignoreInvalidEndpoint;
+ this.allowOptimisedComponents = source.allowOptimisedComponents;
this.autoStartComponents = source.autoStartComponents;
}
@@ -276,6 +280,36 @@ public class PollEnrichDefinition extends ExpressionNode
return this;
}
+ /**
+ * Whether to auto startup components when poll enricher is starting up.
+ *
+ * @return the builder
+ */
+ public PollEnrichDefinition autoStartComponents(boolean
autoStartComponents) {
+ setAutoStartComponents(Boolean.toString(autoStartComponents));
+ return this;
+ }
+
+ /**
+ * Whether to allow components to optimise if they are {@link
org.apache.camel.spi.SendDynamicAware}.
+ *
+ * @return the builder
+ */
+ public PollEnrichDefinition allowOptimisedComponents(String
allowOptimisedComponents) {
+ setAllowOptimisedComponents(allowOptimisedComponents);
+ return this;
+ }
+
+ /**
+ * Whether to allow components to optimise if they are {@link
org.apache.camel.spi.SendDynamicAware}.
+ *
+ * @return the builder
+ */
+ public PollEnrichDefinition allowOptimisedComponents(boolean
allowOptimisedComponents) {
+
setAllowOptimisedComponents(Boolean.toString(allowOptimisedComponents));
+ return this;
+ }
+
// Properties
//
-------------------------------------------------------------------------
@@ -374,6 +408,14 @@ public class PollEnrichDefinition extends ExpressionNode
this.autoStartComponents = autoStartComponents;
}
+ public String getAllowOptimisedComponents() {
+ return allowOptimisedComponents;
+ }
+
+ public void setAllowOptimisedComponents(String allowOptimisedComponents) {
+ this.allowOptimisedComponents = allowOptimisedComponents;
+ }
+
@Override
public PollEnrichDefinition copyDefinition() {
return new PollEnrichDefinition(this);
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollDynamicAwareResolver.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollDynamicAwareResolver.java
new file mode 100644
index 00000000000..dc743ca4901
--- /dev/null
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollDynamicAwareResolver.java
@@ -0,0 +1,67 @@
+/*
+ * 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.processor;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.FactoryFinder;
+import org.apache.camel.spi.PollDynamicAware;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PollDynamicAwareResolver {
+
+ public static final String RESOURCE_PATH =
"META-INF/services/org/apache/camel/poll-dynamic/";
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PollDynamicAwareResolver.class);
+
+ private FactoryFinder factoryFinder;
+
+ public PollDynamicAware resolve(CamelContext context, String scheme) {
+
+ // use factory finder to find a custom implementations
+ Class<?> type = null;
+ try {
+ type = findFactory(scheme, context);
+ } catch (Exception e) {
+ // ignore
+ }
+
+ if (type != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Found PollDynamicAware: {} via: {}{}",
type.getName(), factoryFinder.getResourcePath(), scheme);
+ }
+ if (PollDynamicAware.class.isAssignableFrom(type)) {
+ PollDynamicAware answer = (PollDynamicAware)
context.getInjector().newInstance(type, false);
+ answer.setScheme(scheme);
+ answer.setCamelContext(context);
+ return answer;
+ } else {
+ throw new IllegalArgumentException("Type is not a
PollDynamicAware implementation. Found: " + type.getName());
+ }
+ }
+
+ return null;
+ }
+
+ private Class<?> findFactory(String name, CamelContext context) {
+ if (factoryFinder == null) {
+ factoryFinder =
context.getCamelContextExtension().getFactoryFinder(RESOURCE_PATH);
+ }
+ return factoryFinder.findClass(name).orElse(null);
+ }
+
+}
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
index a5de92aa02e..12cfd2db423 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
@@ -23,18 +23,22 @@ import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Component;
import org.apache.camel.Consumer;
+import org.apache.camel.DynamicPollingConsumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.Expression;
import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.PollingConsumer;
+import org.apache.camel.ResolveEndpointFailedException;
import org.apache.camel.spi.ConsumerCache;
import org.apache.camel.spi.EndpointUtilizationStatistics;
import org.apache.camel.spi.ExceptionHandler;
import org.apache.camel.spi.HeadersMapFactory;
import org.apache.camel.spi.IdAware;
+import org.apache.camel.spi.PollDynamicAware;
import org.apache.camel.spi.RouteIdAware;
import org.apache.camel.support.AsyncProcessorSupport;
import org.apache.camel.support.BridgeExceptionHandlerToErrorHandler;
@@ -44,6 +48,7 @@ import org.apache.camel.support.EventDrivenPollingConsumer;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.cache.DefaultConsumerCache;
import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -64,10 +69,11 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
private static final Logger LOG =
LoggerFactory.getLogger(PollEnricher.class);
+ private PollDynamicAware dynamicAware;
+ private volatile String scheme;
private CamelContext camelContext;
private ConsumerCache consumerCache;
private HeadersMapFactory headersMapFactory;
- private volatile String scheme;
private String id;
private String routeId;
private String variableReceive;
@@ -79,6 +85,7 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
private int cacheSize;
private boolean ignoreInvalidEndpoint;
private boolean autoStartupComponents = true;
+ private boolean allowOptimisedComponents = true;
/**
* Creates a new {@link PollEnricher}.
@@ -135,6 +142,10 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
this.routeId = routeId;
}
+ public PollDynamicAware getDynamicAware() {
+ return dynamicAware;
+ }
+
public String getUri() {
return uri;
}
@@ -215,6 +226,14 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
this.autoStartupComponents = autoStartupComponents;
}
+ public boolean isAllowOptimisedComponents() {
+ return allowOptimisedComponents;
+ }
+
+ public void setAllowOptimisedComponents(boolean allowOptimisedComponents) {
+ this.allowOptimisedComponents = allowOptimisedComponents;
+ }
+
/**
* Enriches the input data (<code>exchange</code>) by first obtaining
additional data from an endpoint represented
* by an endpoint <code>producer</code> and second by aggregating input
data and additional data. Aggregation of
@@ -240,24 +259,54 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
// use dynamic endpoint so calculate the endpoint to use
Object recipient = null;
+ String staticUri = null;
boolean prototype = cacheSize < 0;
try {
- // favour using expression to compute the recipient endpoint
- recipient = expression != null ? expression.evaluate(exchange,
Object.class) : uri;
- recipient = prepareRecipient(exchange, recipient);
- Endpoint existing = getExistingEndpoint(camelContext, recipient);
+ recipient = expression.evaluate(exchange, Object.class);
+ if (dynamicAware != null) {
+ // if its the same scheme as the pre-resolved dynamic aware
then we can optimise to use it
+ String originalUri = uri;
+ String uri = resolveUri(exchange, recipient);
+ String scheme = resolveScheme(exchange, uri);
+ if (dynamicAware.getScheme().equals(scheme)) {
+ PollDynamicAware.DynamicAwareEntry entry =
dynamicAware.prepare(exchange, uri, originalUri);
+ if (entry != null) {
+ staticUri = dynamicAware.resolveStaticUri(exchange,
entry);
+ if (staticUri != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Optimising poll via
PollDynamicAware component: {} to use static uri: {}", scheme,
+ URISupport.sanitizeUri(staticUri));
+ }
+ }
+ }
+ }
+ }
+ Object targetRecipient = staticUri != null ? staticUri : recipient;
+ targetRecipient = prepareRecipient(exchange, targetRecipient);
+ if (targetRecipient == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Poll dynamic evaluated as null so cannot poll
from any endpoint");
+ }
+ // no endpoint to send to, so ignore
+ callback.done(true);
+ return true;
+ }
+ Endpoint existing = getExistingEndpoint(exchange, targetRecipient);
if (existing == null) {
- endpoint = resolveEndpoint(camelContext, recipient, prototype);
+ endpoint = resolveEndpoint(exchange, targetRecipient,
prototype);
} else {
endpoint = existing;
// we have an existing endpoint then its not a prototype scope
prototype = false;
}
+
// acquire the consumer from the cache
consumer = consumerCache.acquirePollingConsumer(endpoint);
} catch (Exception e) {
if (isIgnoreInvalidEndpoint()) {
- LOG.debug("Endpoint uri is invalid: {}. This exception will be
ignored.", recipient, e);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Endpoint uri is invalid: {}. This exception
will be ignored.", recipient, e);
+ }
} else {
exchange.setException(e);
}
@@ -268,19 +317,25 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
// grab the real delegate consumer that performs the actual polling
final boolean bridgeErrorHandler = isBridgeErrorHandler(consumer);
+ DynamicPollingConsumer dynamicConsumer = null;
+ if (consumer instanceof DynamicPollingConsumer dyn) {
+ dynamicConsumer = dyn;
+ }
+
Exchange resourceExchange;
try {
if (timeout < 0) {
LOG.debug("Consumer receive: {}", consumer);
- resourceExchange = consumer.receive();
+ resourceExchange = dynamicConsumer != null ?
dynamicConsumer.receive(exchange) : consumer.receive();
} else if (timeout == 0) {
LOG.debug("Consumer receiveNoWait: {}", consumer);
- resourceExchange = consumer.receiveNoWait();
+ resourceExchange = dynamicConsumer != null ?
dynamicConsumer.receiveNoWait(exchange) : consumer.receiveNoWait();
} else {
if (LOG.isDebugEnabled()) {
LOG.debug("Consumer receive with timeout: {} ms. {}",
timeout, consumer);
}
- resourceExchange = consumer.receive(timeout);
+ resourceExchange
+ = dynamicConsumer != null ?
dynamicConsumer.receive(exchange, timeout) : consumer.receive(timeout);
}
if (resourceExchange == null) {
@@ -302,7 +357,7 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
}
// remember current redelivery stats
- Object redeliveried = exchange.getIn().getHeader(Exchange.REDELIVERED);
+ Object redelivered = exchange.getIn().getHeader(Exchange.REDELIVERED);
Object redeliveryCounter =
exchange.getIn().getHeader(Exchange.REDELIVERY_COUNTER);
Object redeliveryMaxCounter =
exchange.getIn().getHeader(Exchange.REDELIVERY_MAX_COUNTER);
@@ -365,8 +420,8 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
exchange.getExchangeExtension().setRedeliveryExhausted(false);
// preserve the redelivery stats
- if (redeliveried != null) {
- exchange.getMessage().setHeader(Exchange.REDELIVERED,
redeliveried);
+ if (redelivered != null) {
+ exchange.getMessage().setHeader(Exchange.REDELIVERED,
redelivered);
}
if (redeliveryCounter != null) {
exchange.getMessage().setHeader(Exchange.REDELIVERY_COUNTER, redeliveryCounter);
@@ -410,14 +465,14 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
return ProcessorHelper.prepareRecipient(exchange, recipient);
}
- protected static Endpoint getExistingEndpoint(CamelContext context, Object
recipient) {
- return ProcessorHelper.getExistingEndpoint(context, recipient);
+ protected static Endpoint getExistingEndpoint(Exchange exchange, Object
recipient) {
+ return ProcessorHelper.getExistingEndpoint(exchange, recipient);
}
- protected static Endpoint resolveEndpoint(CamelContext camelContext,
Object recipient, boolean prototype) {
+ protected static Endpoint resolveEndpoint(Exchange exchange, Object
recipient, boolean prototype) {
return prototype
- ? ExchangeHelper.resolvePrototypeEndpoint(camelContext,
recipient)
- : ExchangeHelper.resolveEndpoint(camelContext, recipient);
+ ? ExchangeHelper.resolvePrototypeEndpoint(exchange, recipient)
+ : ExchangeHelper.resolveEndpoint(exchange, recipient);
}
/**
@@ -443,6 +498,36 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
return id;
}
+ protected static String resolveUri(Exchange exchange, Object recipient)
throws NoTypeConversionAvailableException {
+ if (recipient == null) {
+ return null;
+ }
+
+ String uri;
+ // trim strings as end users might have added spaces between separators
+ if (recipient instanceof String string) {
+ uri = string.trim();
+ } else if (recipient instanceof Endpoint endpoint) {
+ uri = endpoint.getEndpointKey();
+ } else {
+ // convert to a string type we can work with
+ uri =
exchange.getContext().getTypeConverter().mandatoryConvertTo(String.class,
exchange, recipient);
+ }
+
+ // in case path has property placeholders then try to let property
component resolve those
+ try {
+ uri =
EndpointHelper.resolveEndpointUriPropertyPlaceholders(exchange.getContext(),
uri);
+ } catch (Exception e) {
+ throw new ResolveEndpointFailedException(uri, e);
+ }
+
+ return uri;
+ }
+
+ protected static String resolveScheme(Exchange exchange, String uri) {
+ return ExchangeHelper.resolveScheme(uri);
+ }
+
@Override
protected void doBuild() throws Exception {
if (consumerCache == null) {
@@ -470,9 +555,43 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
scheme = ExchangeHelper.resolveScheme(u);
}
+ if (isAllowOptimisedComponents() && uri != null) {
+ try {
+ if (scheme != null) {
+ // find out if the component can be optimised for
send-dynamic
+ PollDynamicAwareResolver resolver = new
PollDynamicAwareResolver();
+ dynamicAware = resolver.resolve(camelContext, scheme);
+ if (dynamicAware == null) {
+ // okay fallback and try with default component name
+ Component comp = camelContext.getComponent(scheme,
false, isAutoStartupComponents());
+ if (comp != null) {
+ String defaultScheme = comp.getDefaultName();
+ if (!scheme.equals(defaultScheme)) {
+ dynamicAware = resolver.resolve(camelContext,
defaultScheme);
+ dynamicAware.setScheme(scheme);
+ }
+ }
+ }
+ if (dynamicAware != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Detected PollDynamicAware component: {}
optimising poll: {}", scheme,
+ URISupport.sanitizeUri(uri));
+ }
+ }
+ }
+ } catch (Exception e) {
+ // ignore
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Error creating optimised PollDynamicAwareResolver
for uri: {} due to {}. This exception is ignored",
+ URISupport.sanitizeUri(uri), e.getMessage(), e);
+ }
+ }
+ }
+
headersMapFactory =
camelContext.getCamelContextExtension().getHeadersMapFactory();
- ServiceHelper.initService(consumerCache, aggregationStrategy);
+ ServiceHelper.initService(consumerCache, aggregationStrategy,
dynamicAware);
}
@Override
@@ -482,12 +601,12 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
camelContext.getComponent(scheme);
}
- ServiceHelper.startService(consumerCache, aggregationStrategy);
+ ServiceHelper.startService(consumerCache, aggregationStrategy,
dynamicAware);
}
@Override
protected void doStop() throws Exception {
- ServiceHelper.stopService(aggregationStrategy, consumerCache);
+ ServiceHelper.stopService(aggregationStrategy, consumerCache,
dynamicAware);
}
@Override
@@ -502,7 +621,7 @@ public class PollEnricher extends AsyncProcessorSupport
implements IdAware, Rout
if (newExchange != null) {
copyResultsPreservePattern(oldExchange, newExchange);
} else {
- // if no newExchange then there was no message from the
external resource
+ // if no newExchange then there was no message from the
external resource,
// and therefore we should set an empty body to indicate this
fact
// but keep headers/attachments as we want to propagate those
oldExchange.getIn().setBody(null);
diff --git
a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/PollEnrichReifier.java
b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/PollEnrichReifier.java
index 2fb2f72a281..10616641780 100644
---
a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/PollEnrichReifier.java
+++
b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/PollEnrichReifier.java
@@ -73,6 +73,9 @@ public class PollEnrichReifier extends
ProcessorReifier<PollEnrichDefinition> {
if (definition.getAutoStartComponents() != null) {
enricher.setAutoStartupComponents(parseBoolean(definition.getAutoStartComponents(),
true));
}
+ if (definition.getAllowOptimisedComponents() != null) {
+
enricher.setAllowOptimisedComponents(parseBoolean(definition.getAllowOptimisedComponents(),
true));
+ }
return enricher;
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/enricher/PollDynamicFileNameTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/enricher/PollDynamicFileNameOptimizeDisabledTest.java
similarity index 54%
copy from
core/camel-core/src/test/java/org/apache/camel/processor/enricher/PollDynamicFileNameTest.java
copy to
core/camel-core/src/test/java/org/apache/camel/processor/enricher/PollDynamicFileNameOptimizeDisabledTest.java
index 7eaf817514b..fcf2c588d51 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/enricher/PollDynamicFileNameTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/enricher/PollDynamicFileNameOptimizeDisabledTest.java
@@ -19,12 +19,13 @@ package org.apache.camel.processor.enricher;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-public class PollDynamicFileNameTest extends ContextTestSupport {
+public class PollDynamicFileNameOptimizeDisabledTest extends
ContextTestSupport {
@Test
- public void testPollEnrichFile() throws Exception {
+ public void testPollEnrichFileOne() throws Exception {
getMockEndpoint("mock:result").expectedMessageCount(2);
getMockEndpoint("mock:result").message(0).body().isEqualTo("Hello
World");
getMockEndpoint("mock:result").message(1).body().isNull();
@@ -35,6 +36,29 @@ public class PollDynamicFileNameTest extends
ContextTestSupport {
template.sendBodyAndHeader("direct:start", "Bar", "target",
"unknown.txt");
assertMockEndpointsSatisfied();
+
+ // there should only be 1 file endpoint
+ long c = context.getEndpoints().stream()
+ .filter(e -> e.getEndpointKey().startsWith("file") &&
e.getEndpointUri().contains("?fileName=")).count();
+ Assertions.assertEquals(2, c, "There should only be 2 file endpoints");
+ }
+
+ @Test
+ public void testPollEnrichFileTwo() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceivedInAnyOrder("Hello
World", "Bye World");
+
+ template.sendBodyAndHeader(fileUri(), "Hello World",
Exchange.FILE_NAME, "myfile.txt");
+ template.sendBodyAndHeader(fileUri(), "Bye World", Exchange.FILE_NAME,
"myfile2.txt");
+
+ template.sendBodyAndHeader("direct:start", "Foo", "target",
"myfile.txt");
+ template.sendBodyAndHeader("direct:start", "Bar", "target",
"myfile2.txt");
+
+ assertMockEndpointsSatisfied();
+
+ // there should only be 1 file endpoint
+ long c = context.getEndpoints().stream()
+ .filter(e -> e.getEndpointKey().startsWith("file") &&
e.getEndpointUri().contains("?fileName=")).count();
+ Assertions.assertEquals(2, c, "There should only be 2 file endpoints");
}
@Override
@@ -43,7 +67,8 @@ public class PollDynamicFileNameTest extends
ContextTestSupport {
@Override
public void configure() {
from("direct:start")
- .poll(fileUri() +
"?noop=true&fileName=${header.target}", 500)
+ .pollEnrich().simple(fileUri() +
"?noop=true&fileName=${header.target}")
+ .allowOptimisedComponents(false).timeout(500)
.to("mock:result");
}
};
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/enricher/PollDynamicFileNameTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/enricher/PollDynamicFileNameTest.java
index 7eaf817514b..cd80d65f042 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/enricher/PollDynamicFileNameTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/enricher/PollDynamicFileNameTest.java
@@ -19,12 +19,13 @@ package org.apache.camel.processor.enricher;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
public class PollDynamicFileNameTest extends ContextTestSupport {
@Test
- public void testPollEnrichFile() throws Exception {
+ public void testPollEnrichFileOne() throws Exception {
getMockEndpoint("mock:result").expectedMessageCount(2);
getMockEndpoint("mock:result").message(0).body().isEqualTo("Hello
World");
getMockEndpoint("mock:result").message(1).body().isNull();
@@ -35,6 +36,29 @@ public class PollDynamicFileNameTest extends
ContextTestSupport {
template.sendBodyAndHeader("direct:start", "Bar", "target",
"unknown.txt");
assertMockEndpointsSatisfied();
+
+ // there should only be 1 file endpoint
+ long c = context.getEndpoints().stream()
+ .filter(e -> e.getEndpointKey().startsWith("file") &&
e.getEndpointUri().contains("?fileName=")).count();
+ Assertions.assertEquals(1, c, "There should only be 1 file endpoint");
+ }
+
+ @Test
+ public void testPollEnrichFileTwo() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceivedInAnyOrder("Hello
World", "Bye World");
+
+ template.sendBodyAndHeader(fileUri(), "Hello World",
Exchange.FILE_NAME, "myfile.txt");
+ template.sendBodyAndHeader(fileUri(), "Bye World", Exchange.FILE_NAME,
"myfile2.txt");
+
+ template.sendBodyAndHeader("direct:start", "Foo", "target",
"myfile.txt");
+ template.sendBodyAndHeader("direct:start", "Bar", "target",
"myfile2.txt");
+
+ assertMockEndpointsSatisfied();
+
+ // there should only be 1 file endpoint
+ long c = context.getEndpoints().stream()
+ .filter(e -> e.getEndpointKey().startsWith("file") &&
e.getEndpointUri().contains("?fileName=")).count();
+ Assertions.assertEquals(1, c, "There should only be 1 file endpoint");
}
@Override
diff --git
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPollEnricherMBean.java
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPollEnricherMBean.java
index 9e677be2500..439a8675033 100644
---
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPollEnricherMBean.java
+++
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPollEnricherMBean.java
@@ -45,6 +45,12 @@ public interface ManagedPollEnricherMBean extends
ManagedProcessorMBean, Managed
@ManagedAttribute(description = "Whether to aggregate when there was an
exception thrown during calling the resource endpoint")
Boolean isAggregateOnException();
+ @ManagedAttribute(description = "Whether to allow components to optimise
poll if they are PollDynamicAware")
+ Boolean isAllowOptimisedComponents();
+
+ @ManagedAttribute(description = "Whether an optimised component
(PollDynamicAware) is in use")
+ Boolean isOptimised();
+
@Override
@ManagedOperation(description = "Statistics of the endpoints that has been
poll enriched from")
TabularData extendedInformation();
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPollEnricher.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPollEnricher.java
index ab6b19f602d..db939964dc0 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPollEnricher.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPollEnricher.java
@@ -114,6 +114,16 @@ public class ManagedPollEnricher extends ManagedProcessor
implements ManagedPoll
return processor.isAggregateOnException();
}
+ @Override
+ public Boolean isAllowOptimisedComponents() {
+ return processor.isAllowOptimisedComponents();
+ }
+
+ @Override
+ public Boolean isOptimised() {
+ return processor.getDynamicAware() != null;
+ }
+
@Override
public TabularData extendedInformation() {
try {
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
b/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
index 658f7896d25..c6f852bbec2 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
@@ -568,6 +568,18 @@ public abstract class ScheduledPollConsumer extends
DefaultConsumer
*/
protected abstract int poll() throws Exception;
+ /**
+ * The polling method which is invoked periodically to poll this consumer,
for components that support
+ * {@link org.apache.camel.DynamicPollingConsumer} such as camel-file.
+ *
+ * @param dynamic the current exchange when being used from Poll and
PollEnrich EIPs in dynamic mode,
+ * @return number of messages polled, will be <tt>0</tt> if no
message was polled at all.
+ * @throws Exception can be thrown if an exception occurred during polling
+ */
+ protected int poll(Exchange dynamic) throws Exception {
+ return poll();
+ }
+
@Override
protected void doBuild() throws Exception {
if (getHealthCheck() == null) {
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/component/PollDynamicAwareSupport.java
b/core/camel-support/src/main/java/org/apache/camel/support/component/PollDynamicAwareSupport.java
new file mode 100644
index 00000000000..6b8dfe9dda4
--- /dev/null
+++
b/core/camel-support/src/main/java/org/apache/camel/support/component/PollDynamicAwareSupport.java
@@ -0,0 +1,144 @@
+/*
+ * 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.support.component;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.EndpointUriFactory;
+import org.apache.camel.spi.PollDynamicAware;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.util.StringHelper;
+import org.apache.camel.util.URISupport;
+
+/**
+ * Support class for {@link PollDynamicAware} implementations.
+ */
+public abstract class PollDynamicAwareSupport extends ServiceSupport
implements PollDynamicAware {
+
+ private CamelContext camelContext;
+ private Set<String> knownProperties;
+ private Set<String> knownPrefixes;
+ private String scheme;
+
+ @Override
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ @Override
+ public void setScheme(String scheme) {
+ this.scheme = scheme;
+ }
+
+ @Override
+ public String getScheme() {
+ return scheme;
+ }
+
+ @Override
+ public boolean resolveRawParameterValues() {
+ return true;
+ }
+
+ @Override
+ protected void doInit() throws Exception {
+ if (knownProperties == null || knownPrefixes == null) {
+ // optimize to eager load the list of known properties/prefixes
+ EndpointUriFactory factory =
getCamelContext().getCamelContextExtension().getEndpointUriFactory(getScheme());
+ if (factory == null) {
+ throw new IllegalStateException("Cannot find
EndpointUriFactory for component: " + getScheme());
+ }
+ knownProperties = factory.propertyNames();
+ knownPrefixes = factory.multiValuePrefixes();
+ }
+ }
+
+ public Map<String, Object> endpointProperties(Exchange exchange, String
uri) throws Exception {
+ Map<String, Object> properties;
+ // optimize as we know its only query parameters that can be dynamic
+ Map<String, Object> map =
URISupport.parseQuery(URISupport.extractQuery(uri));
+ if (map != null && !map.isEmpty() && isLenientProperties()) {
+ if (resolveRawParameterValues()) {
+ // parameters using raw syntax: RAW(value)
+ // should have the token removed, so its only the value we
have in parameters, as we are about to create
+ // an endpoint and want to have the parameter values without
the RAW tokens
+
RawParameterHelper.resolveRawParameterValues(exchange.getContext(), map);
+ }
+ // okay so only add the known properties as they are the non
lenient properties
+ properties = new LinkedHashMap<>();
+ map.forEach((k, v) -> {
+ boolean accept = knownProperties.contains(k);
+ // we should put the key from a multi-value (prefix) in the
+ // properties too, or the property may be lost
+ if (!accept && !knownPrefixes.isEmpty()) {
+ accept = knownPrefixes.stream().anyMatch(k::startsWith);
+ }
+ if (accept) {
+ properties.put(k, v);
+ }
+ });
+ } else {
+ properties = map;
+ }
+
+ return properties;
+ }
+
+ public Map<String, Object> endpointLenientProperties(Exchange exchange,
String uri) throws Exception {
+ Map<String, Object> properties;
+ // optimize as we know its only query parameters that can be dynamic
+ Map<String, Object> map =
URISupport.parseQuery(URISupport.extractQuery(uri));
+ if (map != null && !map.isEmpty()) {
+ if (resolveRawParameterValues()) {
+ // parameters using raw syntax: RAW(value)
+ // should have the token removed, so its only the value we
have in parameters, as we are about to create
+ // an endpoint and want to have the parameter values without
the RAW tokens
+
RawParameterHelper.resolveRawParameterValues(exchange.getContext(), map);
+ }
+ properties = new LinkedHashMap<>();
+ map.forEach((k, v) -> {
+ // we only accept if the key is not an existing known property
+ // or that the key is not from a multi-value (prefix)
+ boolean accept = !knownProperties.contains(k);
+ if (accept && !knownPrefixes.isEmpty()) {
+ accept = knownPrefixes.stream().noneMatch(k::startsWith);
+ }
+ if (accept) {
+ properties.put(k, v.toString());
+ }
+ });
+ } else {
+ properties = map;
+ }
+ return properties;
+ }
+
+ public String asEndpointUri(Exchange exchange, String uri, Map<String,
Object> properties) throws Exception {
+ String query = URISupport.createQueryString(properties, false);
+ return StringHelper.before(uri, "?", uri) + "?" + query;
+ }
+
+}
diff --git
a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
index e0aa413639d..c0b72173aa4 100644
---
a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
+++
b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
@@ -672,6 +672,7 @@ public class ModelParser extends BaseParser {
case "aggregationStrategy": def.setAggregationStrategy(val);
yield true;
case "aggregationStrategyMethodAllowNull":
def.setAggregationStrategyMethodAllowNull(val); yield true;
case "aggregationStrategyMethodName":
def.setAggregationStrategyMethodName(val); yield true;
+ case "allowOptimisedComponents":
def.setAllowOptimisedComponents(val); yield true;
case "autoStartComponents": def.setAutoStartComponents(val);
yield true;
case "cacheSize": def.setCacheSize(val); yield true;
case "ignoreInvalidEndpoint":
def.setIgnoreInvalidEndpoint(val); yield true;
diff --git
a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
index 08d86619ceb..a0b81f396b9 100644
---
a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
+++
b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
@@ -1371,6 +1371,7 @@ public class ModelWriter extends BaseWriter {
doWriteAttribute("aggregationStrategy", def.getAggregationStrategy(),
null);
doWriteAttribute("ignoreInvalidEndpoint",
def.getIgnoreInvalidEndpoint(), null);
doWriteAttribute("autoStartComponents", def.getAutoStartComponents(),
"true");
+ doWriteAttribute("allowOptimisedComponents",
def.getAllowOptimisedComponents(), "true");
doWriteAttribute("aggregateOnException",
def.getAggregateOnException(), null);
doWriteAttribute("aggregationStrategyMethodName",
def.getAggregationStrategyMethodName(), null);
doWriteAttribute("timeout", def.getTimeout(), "-1");
diff --git
a/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/ModelWriter.java
b/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/ModelWriter.java
index 5175a869386..89d590b00c7 100644
---
a/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/ModelWriter.java
+++
b/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/ModelWriter.java
@@ -1371,6 +1371,7 @@ public class ModelWriter extends BaseWriter {
doWriteAttribute("aggregationStrategy", def.getAggregationStrategy(),
null);
doWriteAttribute("ignoreInvalidEndpoint",
def.getIgnoreInvalidEndpoint(), null);
doWriteAttribute("autoStartComponents", def.getAutoStartComponents(),
"true");
+ doWriteAttribute("allowOptimisedComponents",
def.getAllowOptimisedComponents(), "true");
doWriteAttribute("aggregateOnException",
def.getAggregateOnException(), null);
doWriteAttribute("aggregationStrategyMethodName",
def.getAggregationStrategyMethodName(), null);
doWriteAttribute("timeout", def.getTimeout(), "-1");
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_11.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_11.adoc
new file mode 100644
index 00000000000..2696dd68c5e
--- /dev/null
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_11.adoc
@@ -0,0 +1,18 @@
+= Apache Camel 4.x Upgrade Guide
+
+This document is for helping you upgrade your Apache Camel application
+from Camel 4.x to 4.y. For example, if you are upgrading Camel 4.0 to 4.2,
then you should follow the guides
+from both 4.0 to 4.1 and 4.1 to 4.2.
+
+== Upgrading Camel 4.10 to 4.11
+
+=== file based components
+
+The file based component such as `camel-file`, `camel-ftp`, `camel-smb`, and
`camel-azure-files` has
+been improved to allow optimized dynamic poll when using dynamic `fileName`
header.
+
+This change means that there is a new `DynamicPollingConsumer` API the
consumer implements.
+And as such some APIs inside these components has been changed.
+
+This will only affect if you have built your own custom Camel component on top
of `camel-file`.
+And if so, your custom code may need to be changed slightly as well.
diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide.adoc
index 0645d3df09f..4e5b432ef2a 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide.adoc
@@ -19,3 +19,4 @@ You can find the upgrade guide for each release in the
following pages:
- xref:camel-4x-upgrade-guide-4_8.adoc[Upgrade guide 4.7 -> 4.8]
- xref:camel-4x-upgrade-guide-4_9.adoc[Upgrade guide 4.8 -> 4.9]
- xref:camel-4x-upgrade-guide-4_10.adoc[Upgrade guide 4.9 -> 4.10]
+- xref:camel-4x-upgrade-guide-4_11.adoc[Upgrade guide 4.10 -> 4.11]
diff --git
a/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
b/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
index 5fbdc56df6f..53e8c08fcdb 100644
---
a/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
+++
b/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
@@ -12096,6 +12096,7 @@ public final class ModelDeserializers extends
YamlDeserializerSupport {
@YamlProperty(name = "aggregationStrategy", type =
"string", description = "Sets the AggregationStrategy to be used to merge the
reply from the external service, into a single outgoing message. By default
Camel will use the reply from the external service as outgoing message.",
displayName = "Aggregation Strategy"),
@YamlProperty(name = "aggregationStrategyMethodAllowNull",
type = "string", description = "If this option is false then the aggregate
method is not used if there was no data to enrich. If this option is true then
null values is used as the oldExchange (when no data to enrich), when using
POJOs as the AggregationStrategy.", displayName = "Aggregation Strategy Method
Allow Null"),
@YamlProperty(name = "aggregationStrategyMethodName", type
= "string", description = "This option can be used to explicit declare the
method name to use, when using POJOs as the AggregationStrategy.", displayName
= "Aggregation Strategy Method Name"),
+ @YamlProperty(name = "allowOptimisedComponents", type =
"boolean", description = "Whether to allow components to optimise if they are
org.apache.camel.spi.SendDynamicAware .", displayName = "Allow Optimised
Components"),
@YamlProperty(name = "autoStartComponents", type =
"boolean", description = "Whether to auto startup components when poll enricher
is starting up.", displayName = "Auto Start Components"),
@YamlProperty(name = "cacheSize", type = "number",
description = "Sets the maximum size used by the
org.apache.camel.spi.ConsumerCache which is used to cache and reuse consumers
when uris are reused. Beware that when using dynamic endpoints then it affects
how well the cache can be utilized. If each dynamic endpoint is unique then its
best to turn off caching by setting this to -1, which allows Camel to not cache
both the producers and endpoints; they are regarded as [...]
@YamlProperty(name = "description", type = "string",
description = "Sets the description of this node", displayName = "Description"),
@@ -12142,6 +12143,11 @@ public final class ModelDeserializers extends
YamlDeserializerSupport {
target.setAggregationStrategyMethodName(val);
break;
}
+ case "allowOptimisedComponents": {
+ String val = asText(node);
+ target.setAllowOptimisedComponents(val);
+ break;
+ }
case "autoStartComponents": {
String val = asText(node);
target.setAutoStartComponents(val);
diff --git
a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
index 7deecd56aaf..c69e1e12d76 100644
---
a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
+++
b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
@@ -4108,6 +4108,11 @@
"title" : "Aggregation Strategy Method Name",
"description" : "This option can be used to explicit declare the
method name to use, when using POJOs as the AggregationStrategy."
},
+ "allowOptimisedComponents" : {
+ "type" : "boolean",
+ "title" : "Allow Optimised Components",
+ "description" : "Whether to allow components to optimise if they
are org.apache.camel.spi.SendDynamicAware ."
+ },
"autoStartComponents" : {
"type" : "boolean",
"title" : "Auto Start Components",
diff --git
a/tooling/spi-annotations/src/main/java/org/apache/camel/spi/annotations/PollDynamic.java
b/tooling/spi-annotations/src/main/java/org/apache/camel/spi/annotations/PollDynamic.java
new file mode 100644
index 00000000000..fd5325c72e4
--- /dev/null
+++
b/tooling/spi-annotations/src/main/java/org/apache/camel/spi/annotations/PollDynamic.java
@@ -0,0 +1,33 @@
+/*
+ * 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.spi.annotations;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Target({ ElementType.TYPE })
+@ServiceFactory("poll-dynamic")
+public @interface PollDynamic {
+
+ String value();
+
+}
