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 b0d94141a1ae CAMEL-26229: camel-mongodb-gridfs - the consumer offers
the file of a failed exchange again, closes the download stream and sets
CamelFileName, an unknown producer operation fails, and listAll gives the hex
id (#29071)
b0d94141a1ae is described below
commit b0d94141a1aefc86394294cc37c1c1d8a1e10168
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Nov 3 18:04:56 2026 +0100
CAMEL-26229: camel-mongodb-gridfs - the consumer offers the file of a
failed exchange again, closes the download stream and sets CamelFileName, an
unknown producer operation fails, and listAll gives the hex id (#29071)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../camel/catalog/components/mongodb-gridfs.json | 8 +-
.../catalog/docs/mongodb-gridfs-component.adoc | 34 ++-
.../component/mongodb/gridfs/mongodb-gridfs.json | 8 +-
.../src/main/docs/mongodb-gridfs-component.adoc | 34 ++-
.../component/mongodb/gridfs/GridFsConstants.java | 6 +-
.../component/mongodb/gridfs/GridFsConsumer.java | 137 ++++++++--
.../component/mongodb/gridfs/GridFsEndpoint.java | 15 +-
.../component/mongodb/gridfs/GridFsProducer.java | 22 +-
.../gridfs/GridFsConsumerTimestampTest.java | 291 ++++++++++++++++++++-
.../mongodb/gridfs/GridFsProducerTest.java | 129 +++++++++
.../gridfs/integration/GridFsConsumerIT.java | 1 +
.../integration/GridFsProducerOperationsIT.java | 19 ++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 27 +-
.../endpoint/dsl/GridFsEndpointBuilderFactory.java | 8 +-
14 files changed, 679 insertions(+), 60 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mongodb-gridfs.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mongodb-gridfs.json
index 7a2d571f88ea..6ea9b6144d29 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mongodb-gridfs.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mongodb-gridfs.json
@@ -31,8 +31,8 @@
"headers": {
"CamelFileContentType": { "index": 0, "kind": "header", "displayName": "",
"group": "consumer", "label": "consumer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The content type of the file.",
"constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_CONTENT_TYPE" },
"CamelFileLength": { "index": 1, "kind": "header", "displayName": "",
"group": "consumer", "label": "consumer", "required": false, "javaType":
"long", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The size of the file.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_LENGTH" },
- "CamelFileLastModified": { "index": 2, "kind": "header", "displayName":
"", "group": "consumer", "label": "consumer", "required": false, "javaType":
"Date", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The size of the file.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_LAST_MODIFIED"
},
- "CamelFileName": { "index": 3, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The name of the file.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_NAME" },
+ "CamelFileLastModified": { "index": 2, "kind": "header", "displayName":
"", "group": "consumer", "label": "consumer", "required": false, "javaType":
"Date", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The upload date of the file.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_LAST_MODIFIED"
},
+ "CamelFileName": { "index": 3, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "(producer) The name of the file: the name that create
gives the file, and the file that findOne, remove, listAll and count look for.
(consumer) The name of the file that was found.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsCo [...]
"Content-Type": { "index": 4, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The content type of the file.",
"constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#CONTENT_TYPE" },
"CamelFileNameProduced": { "index": 5, "kind": "header", "displayName":
"", "group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The file name produced.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_NAME_PRODUCED"
},
"CamelGridFsMetadata": { "index": 6, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "Any additional metadata stored along with the file in
JSON format.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#GRIDFS_METADATA" },
@@ -42,9 +42,9 @@
"CamelGridFsObjectId": { "index": 10, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"org.bson.types.ObjectId", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The ObjectId of the
file.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#GRIDFS_OBJECT_ID" }
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "stream" ], "description": "One message per
new file: a stream of the content of the file (the stream cache when stream
caching is on), with its length, upload date, content type and metadata in
headers" },
+ "consumerOut": { "carriers": [ "stream" ], "description": "One message per
new file: a stream of the content of the file (the stream cache when stream
caching is on), closed when the exchange is done, with its name, length, upload
date, content type and metadata in headers" },
"producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"byOption": [ { "option": "operation", "value": "findOne", "carriers": [ "any"
] }, { "option": "operation", "value": "listAll", "carriers": [ "any" ] }, {
"option": "operation", "value": "count", "carriers": [ "any" ] }, { "option":
"operation", "value": "remove", "carriers": [ "any" ] } ], "description": "The
content of the file that create, the default operation, uploads under the name
in the CamelFileName header: [...]
- "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "findOne", "carriers": [ "stream" ] }, { "option":
"operation", "value": "listAll", "carriers": [ "stream" ] }, { "option":
"operation", "value": "count", "carriers": [ "value" ], "javaType":
"java.lang.Long" }, { "option": "operation", "value": "remove", "carriers": [
"unchanged" ] } ], "description": "create, the default, leaves the body
unchanged, with the id of the file in the CamelGridF [...]
+ "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "findOne", "carriers": [ "stream" ] }, { "option":
"operation", "value": "listAll", "carriers": [ "stream" ] }, { "option":
"operation", "value": "count", "carriers": [ "value" ], "javaType":
"java.lang.Long" }, { "option": "operation", "value": "remove", "carriers": [
"unchanged" ] } ], "description": "create, the default, leaves the body
unchanged, with the id of the file in the CamelGridF [...]
},
"properties": {
"connectionBean": { "index": 0, "kind": "path", "displayName": "Connection
Bean", "group": "common", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "Name of
com.mongodb.client.MongoClient to use." },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mongodb-gridfs-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mongodb-gridfs-component.adoc
index 0b7208e9a083..c4569730e750 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mongodb-gridfs-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mongodb-gridfs-component.adoc
@@ -42,6 +42,10 @@ include::partial$component-endpoint-headers.adoc[]
=== GridFS operations - producer endpoint
+The `operation` option, or the `CamelGridFsOperation` header when the option
is not set, chooses the operation:
+`create` (the default), `remove`, `findOne`, `listAll` or `count`. Another
operation fails the exchange with an
+`IllegalArgumentException` that lists the supported operations.
+
==== count
Returns the total number of files in the collection, returning a Long
@@ -68,14 +72,14 @@ Long count = template.requestBodyAndHeaders("direct:count",
"irrelevantBody", he
==== listAll
Returns a Reader that lists all the filenames and their IDs in a tab
-separated stream.
+separated stream. The ID of a file is the hex string of its ObjectId.
----
//
from("direct:listAll").to("mongodb-gridfs?database=tickets&operation=listAll");
Reader result = template.requestBodyAndHeader("direct:listAll",
"irrelevantBody");
-filename1.txt 1252314321
-filename2.txt 2897651254
+filename1.txt 665f1c2e8b3e4a5d6c7b8a90
+filename2.txt 665f1c2e8b3e4a5d6c7b8a91
----
==== findOne
@@ -122,6 +126,30 @@ headers.put(Exchange.FILE_NAME, "filename.txt");
template.requestBodyAndHeaders("direct:remove", "", headers);
----
+=== GridFS consumer
+
+The consumer polls the bucket for new files, as the `queryStrategy` option
chooses, and routes one message per file.
+The body is a stream of the content of the file, which is closed when the
exchange is done. The `CamelFileName`,
+`CamelFileLength`, `CamelFileLastModified` (the upload date),
`CamelFileContentType` and `CamelGridFsMetadata` headers
+describe the file.
+
+When the exchange of a file fails, the consumer reports the failure to its
exception handler (which logs it, unless
+the `exceptionHandler` or `bridgeErrorHandler` option is set) and leaves the
file as it was, so that the next poll
+offers it again:
+
+* With a file attribute (`FileAttribute`, `TimeStampAndFileAttribute` and
`PersistentTimestampAndFileAttribute`), the
+ file is not marked as processed, and the poll goes on with the other files.
+* With `TimeStamp` and `PersistentTimestamp`, the files are routed in the
order of their upload date, and the poll stops
+ at the file that failed: the timestamp stays before that file, so the next
poll offers it again, and then the files
+ uploaded after it. A file that always fails holds back the files after it.
+
+Only the first failure of a file is reported to the exception handler: when
the file fails again at the next poll, the
+consumer logs the failure at `DEBUG`. A file is reported again when it fails
after a poll that routed it or did not
+offer it.
+
+A failure that the route handles, such as with a dead letter channel or
`onException` with `handled(true)`, does not
+fail the exchange: the file is then done.
+
== Examples
=== Example route
diff --git
a/components/camel-mongodb-gridfs/src/generated/resources/META-INF/org/apache/camel/component/mongodb/gridfs/mongodb-gridfs.json
b/components/camel-mongodb-gridfs/src/generated/resources/META-INF/org/apache/camel/component/mongodb/gridfs/mongodb-gridfs.json
index 7a2d571f88ea..6ea9b6144d29 100644
---
a/components/camel-mongodb-gridfs/src/generated/resources/META-INF/org/apache/camel/component/mongodb/gridfs/mongodb-gridfs.json
+++
b/components/camel-mongodb-gridfs/src/generated/resources/META-INF/org/apache/camel/component/mongodb/gridfs/mongodb-gridfs.json
@@ -31,8 +31,8 @@
"headers": {
"CamelFileContentType": { "index": 0, "kind": "header", "displayName": "",
"group": "consumer", "label": "consumer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The content type of the file.",
"constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_CONTENT_TYPE" },
"CamelFileLength": { "index": 1, "kind": "header", "displayName": "",
"group": "consumer", "label": "consumer", "required": false, "javaType":
"long", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The size of the file.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_LENGTH" },
- "CamelFileLastModified": { "index": 2, "kind": "header", "displayName":
"", "group": "consumer", "label": "consumer", "required": false, "javaType":
"Date", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The size of the file.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_LAST_MODIFIED"
},
- "CamelFileName": { "index": 3, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The name of the file.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_NAME" },
+ "CamelFileLastModified": { "index": 2, "kind": "header", "displayName":
"", "group": "consumer", "label": "consumer", "required": false, "javaType":
"Date", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The upload date of the file.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_LAST_MODIFIED"
},
+ "CamelFileName": { "index": 3, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "(producer) The name of the file: the name that create
gives the file, and the file that findOne, remove, listAll and count look for.
(consumer) The name of the file that was found.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsCo [...]
"Content-Type": { "index": 4, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The content type of the file.",
"constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#CONTENT_TYPE" },
"CamelFileNameProduced": { "index": 5, "kind": "header", "displayName":
"", "group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The file name produced.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#FILE_NAME_PRODUCED"
},
"CamelGridFsMetadata": { "index": 6, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "Any additional metadata stored along with the file in
JSON format.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#GRIDFS_METADATA" },
@@ -42,9 +42,9 @@
"CamelGridFsObjectId": { "index": 10, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"org.bson.types.ObjectId", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The ObjectId of the
file.", "constantName":
"org.apache.camel.component.mongodb.gridfs.GridFsConstants#GRIDFS_OBJECT_ID" }
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "stream" ], "description": "One message per
new file: a stream of the content of the file (the stream cache when stream
caching is on), with its length, upload date, content type and metadata in
headers" },
+ "consumerOut": { "carriers": [ "stream" ], "description": "One message per
new file: a stream of the content of the file (the stream cache when stream
caching is on), closed when the exchange is done, with its name, length, upload
date, content type and metadata in headers" },
"producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"byOption": [ { "option": "operation", "value": "findOne", "carriers": [ "any"
] }, { "option": "operation", "value": "listAll", "carriers": [ "any" ] }, {
"option": "operation", "value": "count", "carriers": [ "any" ] }, { "option":
"operation", "value": "remove", "carriers": [ "any" ] } ], "description": "The
content of the file that create, the default operation, uploads under the name
in the CamelFileName header: [...]
- "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "findOne", "carriers": [ "stream" ] }, { "option":
"operation", "value": "listAll", "carriers": [ "stream" ] }, { "option":
"operation", "value": "count", "carriers": [ "value" ], "javaType":
"java.lang.Long" }, { "option": "operation", "value": "remove", "carriers": [
"unchanged" ] } ], "description": "create, the default, leaves the body
unchanged, with the id of the file in the CamelGridF [...]
+ "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "findOne", "carriers": [ "stream" ] }, { "option":
"operation", "value": "listAll", "carriers": [ "stream" ] }, { "option":
"operation", "value": "count", "carriers": [ "value" ], "javaType":
"java.lang.Long" }, { "option": "operation", "value": "remove", "carriers": [
"unchanged" ] } ], "description": "create, the default, leaves the body
unchanged, with the id of the file in the CamelGridF [...]
},
"properties": {
"connectionBean": { "index": 0, "kind": "path", "displayName": "Connection
Bean", "group": "common", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "Name of
com.mongodb.client.MongoClient to use." },
diff --git
a/components/camel-mongodb-gridfs/src/main/docs/mongodb-gridfs-component.adoc
b/components/camel-mongodb-gridfs/src/main/docs/mongodb-gridfs-component.adoc
index 0b7208e9a083..c4569730e750 100644
---
a/components/camel-mongodb-gridfs/src/main/docs/mongodb-gridfs-component.adoc
+++
b/components/camel-mongodb-gridfs/src/main/docs/mongodb-gridfs-component.adoc
@@ -42,6 +42,10 @@ include::partial$component-endpoint-headers.adoc[]
=== GridFS operations - producer endpoint
+The `operation` option, or the `CamelGridFsOperation` header when the option
is not set, chooses the operation:
+`create` (the default), `remove`, `findOne`, `listAll` or `count`. Another
operation fails the exchange with an
+`IllegalArgumentException` that lists the supported operations.
+
==== count
Returns the total number of files in the collection, returning a Long
@@ -68,14 +72,14 @@ Long count = template.requestBodyAndHeaders("direct:count",
"irrelevantBody", he
==== listAll
Returns a Reader that lists all the filenames and their IDs in a tab
-separated stream.
+separated stream. The ID of a file is the hex string of its ObjectId.
----
//
from("direct:listAll").to("mongodb-gridfs?database=tickets&operation=listAll");
Reader result = template.requestBodyAndHeader("direct:listAll",
"irrelevantBody");
-filename1.txt 1252314321
-filename2.txt 2897651254
+filename1.txt 665f1c2e8b3e4a5d6c7b8a90
+filename2.txt 665f1c2e8b3e4a5d6c7b8a91
----
==== findOne
@@ -122,6 +126,30 @@ headers.put(Exchange.FILE_NAME, "filename.txt");
template.requestBodyAndHeaders("direct:remove", "", headers);
----
+=== GridFS consumer
+
+The consumer polls the bucket for new files, as the `queryStrategy` option
chooses, and routes one message per file.
+The body is a stream of the content of the file, which is closed when the
exchange is done. The `CamelFileName`,
+`CamelFileLength`, `CamelFileLastModified` (the upload date),
`CamelFileContentType` and `CamelGridFsMetadata` headers
+describe the file.
+
+When the exchange of a file fails, the consumer reports the failure to its
exception handler (which logs it, unless
+the `exceptionHandler` or `bridgeErrorHandler` option is set) and leaves the
file as it was, so that the next poll
+offers it again:
+
+* With a file attribute (`FileAttribute`, `TimeStampAndFileAttribute` and
`PersistentTimestampAndFileAttribute`), the
+ file is not marked as processed, and the poll goes on with the other files.
+* With `TimeStamp` and `PersistentTimestamp`, the files are routed in the
order of their upload date, and the poll stops
+ at the file that failed: the timestamp stays before that file, so the next
poll offers it again, and then the files
+ uploaded after it. A file that always fails holds back the files after it.
+
+Only the first failure of a file is reported to the exception handler: when
the file fails again at the next poll, the
+consumer logs the failure at `DEBUG`. A file is reported again when it fails
after a poll that routed it or did not
+offer it.
+
+A failure that the route handles, such as with a dead letter channel or
`onException` with `handled(true)`, does not
+fail the exchange: the file is then done.
+
== Examples
=== Example route
diff --git
a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConstants.java
b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConstants.java
index 674fa58a6207..da1d7121f1f2 100644
---
a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConstants.java
+++
b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConstants.java
@@ -25,9 +25,11 @@ public final class GridFsConstants {
public static final String FILE_CONTENT_TYPE = Exchange.FILE_CONTENT_TYPE;
@Metadata(label = "consumer", description = "The size of the file.",
javaType = "long")
public static final String FILE_LENGTH = Exchange.FILE_LENGTH;
- @Metadata(label = "consumer", description = "The size of the file.",
javaType = "Date")
+ @Metadata(label = "consumer", description = "The upload date of the
file.", javaType = "Date")
public static final String FILE_LAST_MODIFIED =
Exchange.FILE_LAST_MODIFIED;
- @Metadata(label = "producer", description = "The name of the file.",
javaType = "String")
+ @Metadata(description = "(producer) The name of the file: the name that
create gives the file, and the file that"
+ + " findOne, remove, listAll and count look for.
(consumer) The name of the file that was found.",
+ javaType = "String")
public static final String FILE_NAME = Exchange.FILE_NAME;
@Metadata(label = "producer", description = "The content type of the
file.", javaType = "String")
public static final String CONTENT_TYPE = Exchange.CONTENT_TYPE;
diff --git
a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
index 439fc15b4664..896c6366324b 100644
---
a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
+++
b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
@@ -19,6 +19,8 @@ package org.apache.camel.component.mongodb.gridfs;
import java.io.InputStream;
import java.time.Duration;
import java.util.Date;
+import java.util.HashSet;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
@@ -26,17 +28,23 @@ import com.mongodb.BasicDBObject;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.gridfs.GridFSDownloadStream;
+import com.mongodb.client.gridfs.GridFSFindIterable;
import com.mongodb.client.gridfs.model.GridFSFile;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.FindOneAndUpdateOptions;
import com.mongodb.client.model.ReturnDocument;
+import com.mongodb.client.model.Sorts;
import com.mongodb.client.model.Updates;
import org.apache.camel.Exchange;
+import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.SynchronizationAdapter;
import org.apache.camel.support.task.BlockingTask;
import org.apache.camel.support.task.Tasks;
import org.apache.camel.support.task.budget.Budgets;
+import org.apache.camel.util.IOHelper;
+import org.bson.BsonValue;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.slf4j.Logger;
@@ -68,6 +76,10 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
private volatile Document persistentTimestamp;
// the number of polls in a row that failed, so that only the first
failure of an outage is reported
private volatile int consecutiveFailures;
+ // the ids of the files whose exchange failed in the last poll: when such
a file fails again in the next poll, the
+ // failure is logged at DEBUG instead of being reported again, so a file
that fails at every poll is reported once,
+ // as for the failures of the polls. It holds the files of one poll only,
so it does not grow with the bucket
+ private volatile Set<BsonValue> lastFailedFiles = Set.of();
public GridFsConsumer(GridFsEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -115,6 +127,7 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
ptsCollection = null;
persistentTimestamp = null;
consecutiveFailures = 0;
+ lastFailedFiles = Set.of();
if (usesTimestamp && !persistsTimestamp) {
fromDate = startDate;
}
@@ -193,9 +206,18 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
return false;
}
- Date pollFromDate = fromDate;
- try (MongoCursor<GridFSFile> cursor =
getGridFSFileMongoCursor(pollFromDate, usesTimestamp, usesAttribute)) {
- boolean dateModified = false;
+ // without a file attribute, the timestamp is the only record of the
files that were routed: the files are
+ // routed in the order of their upload date, and the poll stops at a
file whose exchange failed, so that the
+ // timestamp stays before that file and the next poll offers it again
(and the files after it)
+ boolean stopsAtFailure = usesTimestamp && !usesAttribute;
+ Date pollStartDate = fromDate;
+ Date pollFromDate = pollStartDate;
+ // the upload date of the oldest file of this poll whose exchange
failed
+ Date failedDate = null;
+ // the ids of the files of this poll whose exchange failed
+ Set<BsonValue> failedFiles = new HashSet<>();
+ try (MongoCursor<GridFSFile> cursor
+ = getGridFSFileMongoCursor(pollStartDate, usesTimestamp,
usesAttribute, stopsAtFailure)) {
while (cursor.hasNext() && isStarted()) {
GridFSFile file = cursor.next();
@@ -208,55 +230,122 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
fOrig =
endpoint.getFilesCollection().findOneAndUpdate(filter, update, options);
}
if (fOrig != null) {
- Exchange exchange = createExchange(true);
// by id: several files can have the same name
(revisions), and by name GridFS opens the newest
GridFSDownloadStream downloadStream =
endpoint.getGridFsBucket().openDownloadStream(file.getId());
file = downloadStream.getGridFSFile();
- Document metadata = file.getMetadata();
- if (metadata != null) {
- String contentType =
metadata.get(GRIDFS_FILE_KEY_CONTENT_TYPE, String.class);
- if (contentType != null) {
-
exchange.getIn().setHeader(GridFsConstants.FILE_CONTENT_TYPE, contentType);
- }
-
exchange.getIn().setHeader(GridFsConstants.GRIDFS_METADATA, metadata.toJson());
- }
-
- exchange.getIn().setHeader(GridFsConstants.FILE_LENGTH,
file.getLength());
-
exchange.getIn().setHeader(GridFsConstants.FILE_LAST_MODIFIED,
file.getUploadDate());
- exchange.getIn().setBody(downloadStream,
InputStream.class);
- try {
- getProcessor().process(exchange);
+ if (processFile(file, downloadStream,
lastFailedFiles.contains(file.getId()))) {
if (usesAttribute) {
Bson update =
Updates.set(endpoint.getFileAttributeName(), GRIDFS_FILE_ATTRIBUTE_DONE);
endpoint.getFilesCollection().findOneAndUpdate(eq("_id", fOrig.getId()),
update);
}
if (usesTimestamp &&
file.getUploadDate().compareTo(pollFromDate) > 0) {
pollFromDate = file.getUploadDate();
- dateModified = true;
if (!usesAttribute) {
fromDate = pollFromDate;
}
}
- } catch (Exception e) {
- // ignore
+ } else {
+ failedFiles.add(file.getId());
+ // leave the file as it was, so that the next poll
offers it again
+ if (usesAttribute) {
+ Bson update =
Updates.unset(endpoint.getFileAttributeName());
+
endpoint.getFilesCollection().findOneAndUpdate(eq("_id", fOrig.getId()),
update);
+ }
+ if (usesTimestamp && (failedDate == null ||
file.getUploadDate().before(failedDate))) {
+ failedDate = file.getUploadDate();
+ }
+ if (stopsAtFailure) {
+ break;
+ }
}
}
}
- if (persistsTimestamp && dateModified) {
+ if (failedDate != null && !pollFromDate.before(failedDate)) {
+ // a file routed in this poll is not older than a file that
failed (it has the same upload date or, with
+ // a file attribute, it is newer): keep the timestamp before
the failed file. With a file attribute the
+ // routed files are not routed again; without it, the next
poll also offers again the routed files with
+ // the same upload date as the failed file
+ pollFromDate = new Date(failedDate.getTime() - 1);
+ if (!usesAttribute) {
+ fromDate = pollFromDate;
+ }
+ }
+ if (persistsTimestamp && !pollFromDate.equals(pollStartDate)) {
Bson update = Updates.set(PERSISTENT_TIMESTAMP_KEY,
pollFromDate);
ptsCollection.findOneAndUpdate(eq("_id",
persistentTimestamp.getObjectId("_id")), update);
}
+ } finally {
+ lastFailedFiles = failedFiles;
}
return false;
}
- private MongoCursor<GridFSFile> getGridFSFileMongoCursor(Date fromDate,
boolean usesTimestamp, boolean usesAttribute) {
+ /**
+ * Routes the exchange of one file. The download stream is closed when the
exchange is done. Returns false when the
+ * exchange failed: the failure is reported to the exception handler of
the consumer, or logged at DEBUG when the
+ * file failed in the last poll too, as that failure was reported already.
+ */
+ private boolean processFile(GridFSFile file, GridFSDownloadStream
downloadStream, boolean failedInLastPoll) {
+ Exchange exchange = createExchange(false);
+ try {
+ exchange.getExchangeExtension().addOnCompletion(new
SynchronizationAdapter() {
+ @Override
+ public void onDone(Exchange exchange) {
+ IOHelper.close(downloadStream);
+ }
+ });
+
+ Message in = exchange.getIn();
+ in.setHeader(GridFsConstants.FILE_NAME, file.getFilename());
+ Document metadata = file.getMetadata();
+ if (metadata != null) {
+ String contentType =
metadata.get(GRIDFS_FILE_KEY_CONTENT_TYPE, String.class);
+ if (contentType != null) {
+ in.setHeader(GridFsConstants.FILE_CONTENT_TYPE,
contentType);
+ }
+ in.setHeader(GridFsConstants.GRIDFS_METADATA,
metadata.toJson());
+ }
+ in.setHeader(GridFsConstants.FILE_LENGTH, file.getLength());
+ in.setHeader(GridFsConstants.FILE_LAST_MODIFIED,
file.getUploadDate());
+ in.setBody(downloadStream, InputStream.class);
+
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ // the exchange may not have reached its unit of work, which
runs the on completion
+ IOHelper.close(downloadStream);
+ }
+ Exception cause = exchange.getException();
+ if (cause != null) {
+ if (failedInLastPoll) {
+ LOG.debug("Error processing the GridFS file {} again,
which the next poll offers again: {}",
+ file.getFilename(), cause.getMessage());
+ } else {
+ getExceptionHandler().handleException(
+ "Error processing the GridFS file " +
file.getFilename() + ", which the next poll offers again",
+ exchange, cause);
+ }
+ return false;
+ }
+ return true;
+ } finally {
+ releaseExchange(exchange, false);
+ }
+ }
+
+ private MongoCursor<GridFSFile> getGridFSFileMongoCursor(
+ Date fromDate, boolean usesTimestamp, boolean usesAttribute,
boolean byUploadDate) {
String queryString = endpoint.getQuery();
Bson query = getBsonDocument(fromDate, usesTimestamp, usesAttribute,
queryString);
- return endpoint.getGridFsBucket().find(query).cursor();
+ GridFSFindIterable files = endpoint.getGridFsBucket().find(query);
+ if (byUploadDate) {
+ files = files.sort(Sorts.ascending(GRIDFS_FILE_KEY_UPLOAD_DATE));
+ }
+ return files.cursor();
}
private Bson getBsonDocument(Date fromDate, boolean usesTimestamp, boolean
usesAttribute, String queryString) {
diff --git
a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsEndpoint.java
b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsEndpoint.java
index 35c33bc3d8e6..ef07bcf9eb2c 100644
---
a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsEndpoint.java
+++
b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsEndpoint.java
@@ -48,8 +48,9 @@ import org.slf4j.LoggerFactory;
headersClass = GridFsConstants.class)
@BodyTypes(consumerOut = @BodyType(value = Carrier.STREAM,
description = "One message per new file: a
stream of the content of the file (the"
- + " stream cache when stream
caching is on), with its length, upload date,"
- + " content type and metadata
in headers"),
+ + " stream cache when stream
caching is on), closed when the exchange is"
+ + " done, with its name,
length, upload date, content type and metadata in"
+ + " headers"),
producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE },
byOption = {
@BodyType.When(option = "operation",
value = "findOne", carriers = Carrier.ANY),
@@ -74,8 +75,9 @@ import org.slf4j.LoggerFactory;
description = "create, the default, leaves
the body unchanged, with the id of the file in"
+ " the CamelGridFsFileId
header. findOne gives a stream of the content of"
+ " the file named by
CamelFileName; listAll a Reader of text, one line per"
- + " file with its name and id
separated by a tab; count the number of files"
- + " as a Long; remove leaves
the body unchanged"))
+ + " file with its name and id
(the hex string of an ObjectId) separated by a"
+ + " tab; count the number of
files as a Long; remove leaves the body"
+ + " unchanged"))
public class GridFsEndpoint extends DefaultEndpoint {
private static final Logger LOG =
LoggerFactory.getLogger(GridFsEndpoint.class);
@@ -129,7 +131,10 @@ public class GridFsEndpoint extends DefaultEndpoint {
@Override
public Consumer createConsumer(Processor processor) throws Exception {
initializeConnection();
- return new GridFsConsumer(this, processor);
+ GridFsConsumer consumer = new GridFsConsumer(this, processor);
+ // the exceptionHandler and bridgeErrorHandler options
+ configureConsumer(consumer);
+ return consumer;
}
public void initializeConnection() throws Exception {
diff --git
a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsProducer.java
b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsProducer.java
index 1d68bcf68ee0..4958047c6435 100644
---
a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsProducer.java
+++
b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsProducer.java
@@ -26,6 +26,7 @@ import com.mongodb.client.gridfs.model.GridFSFile;
import com.mongodb.client.gridfs.model.GridFSUploadOptions;
import org.apache.camel.Exchange;
import org.apache.camel.support.DefaultProducer;
+import org.bson.BsonValue;
import org.bson.Document;
import org.bson.types.ObjectId;
@@ -126,6 +127,10 @@ public class GridFsProducer extends DefaultProducer {
count =
endpoint.getFilesCollection().countDocuments(eq(GRIDFS_FILE_KEY_FILENAME,
filename));
}
exchange.getIn().setBody(count, Long.class);
+ } else {
+ throw new IllegalArgumentException(
+ "Unsupported operation: " + operation + ". The supported
operations are: create, remove, findOne,"
+ + " listAll and count");
}
}
@@ -148,10 +153,25 @@ public class GridFsProducer extends DefaultProducer {
}
while (cursor.hasNext() && current.length() < 4000) {
GridFSFile file = cursor.next();
-
current.append(file.getFilename()).append("\t").append(file.getId()).append("\n");
+
current.append(file.getFilename()).append("\t").append(idOf(file)).append("\n");
}
}
+ /**
+ * The id of the file as text: the hex string of an ObjectId (the id
GridFS gives a file), the text of a String
+ * id, and the BSON value of another custom id.
+ */
+ private static String idOf(GridFSFile file) {
+ BsonValue id = file.getId();
+ if (id.isObjectId()) {
+ return id.asObjectId().getValue().toHexString();
+ }
+ if (id.isString()) {
+ return id.asString().getValue();
+ }
+ return String.valueOf(id);
+ }
+
@Override
public int read(char[] cbuf, int off, int len) throws IOException {
if (pos == current.length()) {
diff --git
a/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumerTimestampTest.java
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumerTimestampTest.java
index 8f6cd8fd17cf..58f4300dc6a3 100644
---
a/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumerTimestampTest.java
+++
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumerTimestampTest.java
@@ -21,6 +21,7 @@ import java.util.Comparator;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -32,6 +33,7 @@ import java.util.concurrent.atomic.AtomicReference;
import com.mongodb.MongoClientSettings;
import com.mongodb.MongoTimeoutException;
import com.mongodb.client.FindIterable;
+import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
@@ -43,6 +45,7 @@ import com.mongodb.client.model.FindOneAndUpdateOptions;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.spi.ExceptionHandler;
import org.bson.BsonArray;
@@ -62,13 +65,16 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* The consumer with a timestamp query strategy against a fake bucket that
answers the upload date query of each poll
* the way MongoDB does: a file that was routed must not be routed again by
the following polls. With a file attribute,
- * the fake files collection marks a file as the consumer asks, and the bucket
no longer returns a marked file.
+ * the fake files collection marks a file as the consumer asks, and the bucket
no longer returns a marked file. The
+ * tests that start the consumer in a route also cover a file whose exchange
fails, the headers and the download stream.
*/
class GridFsConsumerTimestampTest {
@@ -81,6 +87,14 @@ class GridFsConsumerTimestampTest {
private final List<Date> persisted = new CopyOnWriteArrayList<>();
// the ids of the files that the consumer has marked with its file
attribute
private final Set<BsonValue> marked = ConcurrentHashMap.newKeySet();
+ // the ids of the files that the consumer has marked as done
+ private final Set<BsonValue> done = ConcurrentHashMap.newKeySet();
+ // the CamelFileName header of each exchange of the route
+ private final List<String> fileNames = new CopyOnWriteArrayList<>();
+ // the download streams that the consumer opened
+ private final List<GridFSDownloadStream> streams = new
CopyOnWriteArrayList<>();
+ // the number of times the route fails the exchange of a file, by file name
+ private final Map<String, AtomicInteger> failures = new
ConcurrentHashMap<>();
// the number of polls (queries of the bucket) that fail, as when MongoDB
cannot be reached
private final AtomicInteger failingPolls = new AtomicInteger();
// the number of reads of the persistent timestamp that fail
@@ -92,6 +106,22 @@ class GridFsConsumerTimestampTest {
private final List<Date> created = new CopyOnWriteArrayList<>();
// the failures that the consumer reported through its ExceptionHandler
private final List<Throwable> reported = new CopyOnWriteArrayList<>();
+ private final ExceptionHandler exceptionHandler = new ExceptionHandler() {
+ @Override
+ public void handleException(Throwable exception) {
+ reported.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ reported.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ reported.add(exception);
+ }
+ };
// called by a query of the bucket, before the files are returned (or the
failure is thrown)
private volatile Runnable onPoll = () -> {
};
@@ -136,6 +166,16 @@ class GridFsConsumerTimestampTest {
return marked.add(id)
? stored.stream().filter(f ->
f.getId().equals(id)).findFirst().orElseThrow() : null;
});
+ // the consumer marks a routed file as done, and removes the mark of a
file whose exchange failed
+ when(files.findOneAndUpdate(any(Bson.class),
any(Bson.class))).thenAnswer(invocation -> {
+ BsonValue id = find(toDocument(invocation.getArgument(0)), "_id");
+ if (toDocument(invocation.getArgument(1)).containsKey("$unset")) {
+ marked.remove(id);
+ } else {
+ done.add(id);
+ }
+ return null;
+ });
MongoCollection<Document> timestamps = timestampCollection();
when(database.getCollection("camel-timestamps")).thenReturn(timestamps);
}
@@ -299,6 +339,188 @@ class GridFsConsumerTimestampTest {
assertEquals(List.of(), reported, "no poll follows, the failure must
not be reported as one that is retried");
}
+ @ParameterizedTest
+ @EnumSource(value = QueryStrategy.class,
+ names = { "FileAttribute", "TimeStampAndFileAttribute",
"PersistentTimestampAndFileAttribute" })
+ void
strategyWithFileAttributeOffersTheFileOfAFailedExchangeAgain(QueryStrategy
strategy) throws Exception {
+ failures.put("bad", new AtomicInteger(1));
+ startRoute(strategy);
+
+ GridFSFile bad = file("bad", 60);
+ GridFSFile good = file("good", 120);
+ stored.addAll(List.of(bad, good));
+ awaitFullPolls(2);
+
+ assertEquals(List.of("bad", "good", "bad"), routed,
+ "the file whose exchange failed must not be marked as done:
the poll goes on with the other files, and the"
+ + " next poll
offers it again");
+ assertEquals(Set.of(bad.getId(), good.getId()), done);
+ assertEquals(1, reported.size(), "the failure must be reported to the
exception handler");
+ assertEquals("Forced failure of bad (test)",
reported.get(0).getMessage());
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = QueryStrategy.class, names = { "TimeStamp",
"PersistentTimestamp" })
+ void
timestampStrategyStopsAtTheFileOfAFailedExchangeAndOffersItAgain(QueryStrategy
strategy) throws Exception {
+ failures.put("first", new AtomicInteger(1));
+ startRoute(strategy);
+
+ GridFSFile first = file("first", 60);
+ GridFSFile second = file("second", 120);
+ stored.addAll(List.of(first, second));
+ awaitFullPolls(2);
+
+ assertEquals(List.of("first", "first", "second"), routed,
+ "the poll must stop at the file whose exchange failed, so that
the timestamp stays before it and the next"
+ + " poll
offers it again");
+ assertEquals(1, reported.size(), "the failure must be reported to the
exception handler");
+ }
+
+ @Test
+ void timestampStrategyRoutesTheFilesInTheOrderOfTheirUploadDate() throws
Exception {
+ failures.put("older", new AtomicInteger(1));
+ startRoute(QueryStrategy.TimeStamp);
+
+ // the newer file comes first in the bucket (another uploader, or a
client with another clock)
+ stored.addAll(List.of(file("newer", 120), file("older", 60)));
+ awaitFullPolls(2);
+
+ assertEquals(List.of("older", "older", "newer"), routed,
+ "a newer file routed first would move the timestamp past the
older file whose exchange failed");
+ }
+
+ @Test
+ void
timestampStrategyOffersAgainTheFilesWithTheUploadDateOfTheFileOfAFailedExchange()
throws Exception {
+ failures.put("b", new AtomicInteger(1));
+ startRoute(QueryStrategy.TimeStamp);
+
+ // two files with the same upload date: the timestamp must stay before
both, so a is routed again
+ Date uploadDate = new Date(System.currentTimeMillis() +
TimeUnit.SECONDS.toMillis(60));
+ stored.addAll(List.of(new GridFSFile(new BsonObjectId(), "a", 1, 1,
uploadDate, null),
+ new GridFSFile(new BsonObjectId(), "b", 1, 1, uploadDate,
null)));
+ awaitFullPolls(2);
+
+ assertEquals(List.of("a", "b", "a", "b"), routed);
+ }
+
+ @Test
+ void
persistentTimestampAndFileAttributeStrategyStoresATimestampBeforeTheFileOfAFailedExchange()
throws Exception {
+ failures.put("bad", new AtomicInteger(1));
+ startRoute(QueryStrategy.PersistentTimestampAndFileAttribute);
+
+ GridFSFile bad = file("bad", 60);
+ stored.addAll(List.of(bad, file("good", 120)));
+ awaitFullPolls(2);
+
+ assertEquals(List.of("bad", "good", "bad"), routed);
+ assertTrue(persisted.get(0).before(bad.getUploadDate()),
+ "a consumer started again with the stored timestamp must find
the file whose exchange failed");
+ }
+
+ @ParameterizedTest
+ @EnumSource(QueryStrategy.class)
+ void consumerSetsTheFileNameHeader(QueryStrategy strategy) throws
Exception {
+ startRoute(strategy);
+
+ stored.add(file("report.csv", 60));
+ awaitFullPolls(2);
+
+ assertEquals(List.of("report.csv"), fileNames);
+ }
+
+ @Test
+ void downloadStreamIsClosedWhenTheExchangeIsDone() throws Exception {
+ startRoute(QueryStrategy.TimeStamp);
+
+ stored.addAll(List.of(file("first", 60), file("second", 120)));
+ awaitFullPolls(2);
+
+ assertEquals(2, streams.size());
+ for (GridFSDownloadStream stream : streams) {
+ verify(stream, atLeastOnce()).close();
+ }
+ }
+
+ @Test
+ void downloadStreamOfAFailedExchangeIsClosed() throws Exception {
+ failures.put("bad", new AtomicInteger(1));
+ startRoute(QueryStrategy.FileAttribute);
+
+ stored.add(file("bad", 60));
+ awaitFullPolls(2);
+
+ assertFalse(streams.isEmpty());
+ for (GridFSDownloadStream stream : streams) {
+ verify(stream, atLeastOnce()).close();
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(QueryStrategy.class)
+ void fileThatFailsAtEveryPollIsReportedOnce(QueryStrategy strategy) throws
Exception {
+ failures.put("bad", new AtomicInteger(Integer.MAX_VALUE));
+ startRoute(strategy);
+
+ stored.add(file("bad", 60));
+ awaitFullPolls(3);
+
+ assertTrue(routed.size() >= 3, "each poll must offer the file again: "
+ routed);
+ assertEquals(List.of("Forced failure of bad (test)"),
messagesOf(reported),
+ "the failures of the file at the next polls must not be
reported again");
+ }
+
+ @Test
+ void fileThatFailsAgainAfterItWasRoutedIsReportedAgain() throws Exception {
+ failures.put("bad", new AtomicInteger(2));
+ startRoute(QueryStrategy.FileAttribute);
+
+ GridFSFile bad = file("bad", 60);
+ stored.add(bad);
+ // two polls fail the file, the third one routes it
+ awaitFullPolls(3);
+ assertEquals(List.of("bad", "bad", "bad"), routed);
+ assertEquals(1, reported.size(), "the second failure must not be
reported again");
+
+ // the file is offered again (its mark is removed), and fails once more
+ failures.get("bad").set(1);
+ marked.remove(bad.getId());
+ awaitFullPolls(3);
+
+ assertEquals(List.of("bad", "bad", "bad", "bad", "bad"), routed);
+ assertEquals(2, reported.size(), "a file that fails again after it was
routed must be reported again");
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = QueryStrategy.class,
+ names = { "FileAttribute", "TimeStampAndFileAttribute",
"PersistentTimestampAndFileAttribute" })
+ void filesThatFailAtEveryPollAreEachReportedOnce(QueryStrategy strategy)
throws Exception {
+ failures.put("first", new AtomicInteger(Integer.MAX_VALUE));
+ failures.put("second", new AtomicInteger(Integer.MAX_VALUE));
+ startRoute(strategy);
+
+ stored.addAll(List.of(file("first", 60), file("second", 120)));
+ awaitFullPolls(3);
+
+ assertEquals(List.of("Forced failure of first (test)", "Forced failure
of second (test)"), messagesOf(reported));
+ }
+
+ @Test
+ void timestampStrategyReportsEachFileThatFailsOnce() throws Exception {
+ // the poll stops at the first file while it fails, then the second
file fails at every poll
+ failures.put("first", new AtomicInteger(3));
+ failures.put("second", new AtomicInteger(Integer.MAX_VALUE));
+ startRoute(QueryStrategy.TimeStamp);
+
+ stored.addAll(List.of(file("first", 60), file("second", 120)));
+ awaitFullPolls(6);
+
+ assertEquals(List.of("Forced failure of first (test)", "Forced failure
of second (test)"), messagesOf(reported));
+ }
+
+ private static List<String> messagesOf(List<Throwable> exceptions) {
+ return exceptions.stream().map(Throwable::getMessage).toList();
+ }
+
private void startConsumer(QueryStrategy strategy) {
startConsumer(strategy, 1);
}
@@ -333,23 +555,62 @@ class GridFsConsumerTimestampTest {
routedIds.add(file.getId());
};
consumer = new GridFsConsumer(endpoint, processor);
- consumer.setExceptionHandler(new ExceptionHandler() {
+ consumer.setExceptionHandler(exceptionHandler);
+ consumer.start();
+ }
+
+ /**
+ * Starts the consumer in a route, which runs the exchanges in a unit of
work, as in an application. The route fails
+ * the exchange of a file as many times as the failures map says, and the
exceptionHandler option of the endpoint
+ * gets the failures that the consumer reports.
+ */
+ private void startRoute(QueryStrategy strategy) throws Exception {
+ context.getRegistry().bind("fakeMongo", mock(MongoClient.class));
+ GridFsComponent component = new GridFsComponent();
+ component.setCamelContext(context);
+ GridFsEndpoint endpoint = new
GridFsEndpoint("mongodb-gridfs:fakeMongo", component) {
+ @Override
+ public void initializeConnection() {
+ // the fake bucket and collections
+ }
+
+ @Override
+ public GridFSBucket getGridFsBucket() {
+ return bucket;
+ }
+
@Override
- public void handleException(Throwable exception) {
- reported.add(exception);
+ public MongoDatabase getDB() {
+ return database;
}
@Override
- public void handleException(String message, Throwable exception) {
- reported.add(exception);
+ public MongoCollection<GridFSFile> getFilesCollection() {
+ return files;
}
+ };
+ endpoint.setCamelContext(context);
+ endpoint.setConnectionBean("fakeMongo");
+ endpoint.setQueryStrategy(strategy);
+ endpoint.setInitialDelay(1);
+ endpoint.setDelay(1);
+ endpoint.setExceptionHandler(exceptionHandler);
+ context.addRoutes(new RouteBuilder() {
@Override
- public void handleException(String message, Exchange exchange,
Throwable exception) {
- reported.add(exception);
+ public void configure() {
+ // the fake download streams have no content to cache
+ from(endpoint).noStreamCaching().process(exchange -> {
+ GridFSFile file =
exchange.getMessage().getBody(GridFSDownloadStream.class).getGridFSFile();
+ routed.add(file.getFilename());
+
fileNames.add(exchange.getMessage().getHeader(Exchange.FILE_NAME,
String.class));
+ AtomicInteger left = failures.get(file.getFilename());
+ if (left != null && left.getAndDecrement() > 0) {
+ throw new IllegalStateException("Forced failure of " +
file.getFilename() + " (test)");
+ }
+ });
}
});
- consumer.start();
}
/**
@@ -362,9 +623,10 @@ class GridFsConsumerTimestampTest {
assertTrue(latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), "the
consumer did not poll " + (count + 1) + " times");
}
- private static GridFSDownloadStream downloadStreamOf(GridFSFile file) {
+ private GridFSDownloadStream downloadStreamOf(GridFSFile file) {
GridFSDownloadStream stream = mock(GridFSDownloadStream.class);
when(stream.getGridFSFile()).thenReturn(file);
+ streams.add(stream);
return stream;
}
@@ -424,6 +686,15 @@ class GridFsConsumerTimestampTest {
when(cursor.next()).thenAnswer(invocation -> iterator.next());
GridFSFindIterable iterable = mock(GridFSFindIterable.class);
when(iterable.cursor()).thenReturn(cursor);
+ // as MongoDB: the files sorted by their upload date, in the direction
that the sort document gives
+ when(iterable.sort(any(Bson.class))).thenAnswer(invocation -> {
+ int direction =
toDocument(invocation.getArgument(0)).getInt32(GridFsConstants.GRIDFS_FILE_KEY_UPLOAD_DATE)
+ .getValue();
+ Comparator<GridFSFile> byUploadDate =
Comparator.comparing(GridFSFile::getUploadDate);
+ List<GridFSFile> sorted = new ArrayList<>(files);
+ sorted.sort(direction < 0 ? byUploadDate.reversed() :
byUploadDate);
+ return iterableOf(sorted);
+ });
return iterable;
}
diff --git
a/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsProducerTest.java
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsProducerTest.java
new file mode 100644
index 000000000000..4d84743034b8
--- /dev/null
+++
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsProducerTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.mongodb.gridfs;
+
+import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
+
+import com.mongodb.client.MongoCursor;
+import com.mongodb.client.gridfs.GridFSBucket;
+import com.mongodb.client.gridfs.GridFSFindIterable;
+import com.mongodb.client.gridfs.model.GridFSFile;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.bson.BsonObjectId;
+import org.bson.BsonString;
+import org.bson.types.ObjectId;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+/**
+ * The producer against a mocked bucket: an operation it does not know fails,
and listAll lists the ids as text.
+ */
+class GridFsProducerTest {
+
+ private final GridFSBucket bucket = mock(GridFSBucket.class);
+ private CamelContext context;
+
+ @BeforeEach
+ void setUp() {
+ context = new DefaultCamelContext();
+ context.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ context.stop();
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "findone", "delete", "list" })
+ void operationOfTheHeaderThatIsNotSupportedFails(String operation) {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setHeader(GridFsConstants.GRIDFS_OPERATION,
operation);
+ exchange.getMessage().setHeader(GridFsConstants.FILE_NAME,
"report.csv");
+
+ IllegalArgumentException e
+ = assertThrows(IllegalArgumentException.class, () ->
producer(null).process(exchange));
+ assertEquals("Unsupported operation: " + operation
+ + ". The supported operations are: create, remove,
findOne, listAll and count",
+ e.getMessage());
+ verifyNoInteractions(bucket);
+ }
+
+ @Test
+ void operationOfTheOptionThatIsNotSupportedFails() {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setHeader(GridFsConstants.FILE_NAME,
"report.csv");
+
+ assertThrows(IllegalArgumentException.class, () ->
producer("findone").process(exchange));
+ verifyNoInteractions(bucket);
+ }
+
+ @Test
+ void listAllGivesTheNameAndTheIdOfEachFile() throws Exception {
+ ObjectId id = new ObjectId();
+ GridFSFindIterable found = iterableOf(List.of(
+ new GridFSFile(new BsonObjectId(id), "report.csv", 1, 1, new
Date(), null),
+ // a file that another client uploaded with an id of its own
+ new GridFSFile(new BsonString("custom-id"), "custom.csv", 1,
1, new Date(), null)));
+ when(bucket.find()).thenReturn(found);
+
+ Exchange exchange = new DefaultExchange(context);
+ producer("listAll").process(exchange);
+
+ assertEquals("report.csv\t" + id.toHexString() +
"\ncustom.csv\tcustom-id\n",
+ exchange.getMessage().getBody(String.class));
+ }
+
+ private GridFsProducer producer(String operation) {
+ GridFsComponent component = new GridFsComponent();
+ component.setCamelContext(context);
+ GridFsEndpoint endpoint = new GridFsEndpoint("mongodb-gridfs:fake",
component) {
+ @Override
+ public GridFSBucket getGridFsBucket() {
+ return bucket;
+ }
+ };
+ endpoint.setCamelContext(context);
+ endpoint.setOperation(operation);
+ return new GridFsProducer(endpoint);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static GridFSFindIterable iterableOf(List<GridFSFile> files) {
+ Iterator<GridFSFile> iterator = files.iterator();
+ MongoCursor<GridFSFile> cursor = mock(MongoCursor.class);
+ when(cursor.hasNext()).thenAnswer(invocation -> iterator.hasNext());
+ when(cursor.next()).thenAnswer(invocation -> iterator.next());
+ GridFSFindIterable iterable = mock(GridFSFindIterable.class);
+ when(iterable.cursor()).thenReturn(cursor);
+ return iterable;
+ }
+}
diff --git
a/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/integration/GridFsConsumerIT.java
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/integration/GridFsConsumerIT.java
index dc9d3bb9f4b9..ee6851c57d0c 100644
---
a/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/integration/GridFsConsumerIT.java
+++
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/integration/GridFsConsumerIT.java
@@ -113,6 +113,7 @@ class GridFsConsumerIT extends AbstractMongoDbITSupport {
MockEndpoint mock = getMockEndpoint(mockUri);
mock.expectedBodiesReceived(FILE_DATA);
mock.expectedHeaderReceived(GridFsConstants.GRIDFS_METADATA,
"{\"contentType\": \"text/plain\"}");
+ mock.expectedHeaderReceived(Exchange.FILE_NAME, FILE_NAME);
Map<String, Object> headers = new HashMap<>();
assertFalse(gridfs.find(eq(GRIDFS_FILE_KEY_FILENAME,
FILE_NAME)).cursor().hasNext());
diff --git
a/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/integration/GridFsProducerOperationsIT.java
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/integration/GridFsProducerOperationsIT.java
index 87303e28c07c..0b2159230076 100644
---
a/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/integration/GridFsProducerOperationsIT.java
+++
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/integration/GridFsProducerOperationsIT.java
@@ -31,6 +31,7 @@ import org.junit.jupiter.api.Test;
import static com.mongodb.client.model.Filters.eq;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -91,6 +92,24 @@ public class GridFsProducerOperationsIT extends
AbstractMongoDbITSupport {
assertFalse(s.contains("2-" + FILE_NAME));
}
+ @Test
+ void testListAllGivesTheIdAndAnUnknownOperationFails() {
+ Exchange created = template.request("direct:create", exchange -> {
+ exchange.getMessage().setBody(FILE_DATA);
+ exchange.getMessage().setHeader(Exchange.FILE_NAME, FILE_NAME);
+ });
+ ObjectId objectId =
created.getMessage().getHeader(GridFsConstants.GRIDFS_OBJECT_ID,
ObjectId.class);
+ assertNotNull(objectId);
+
+ String listing = template.requestBody("direct:listAll", null,
String.class);
+ assertTrue(listing.contains(FILE_NAME + "\t" + objectId.toHexString()
+ "\n"), listing);
+
+ Exchange unknown = template.request(
+
"mongodb-gridfs:myDb?database={{mongodb.testDb}}&operation=findone&bucket=" +
getBucket(),
+ exchange ->
exchange.getMessage().setHeader(Exchange.FILE_NAME, FILE_NAME));
+ assertInstanceOf(IllegalArgumentException.class,
unknown.getException());
+ }
+
@Test
public void testRemoveByObjectId() {
Map<String, Object> headers = new HashMap<>();
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 89212f6aeb35..8061e87cc40e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -4325,7 +4325,7 @@ poll. The failure is given to the exception handler of
the consumer, with the ob
whole bucket, one such object stopped the consumer from consuming any object.
The poll still fails when none of the
listed objects could be got.
-=== camel-mongodb-gridfs - the consumer routes a file once
+=== camel-mongodb-gridfs - the consumer routes a file once, offers the file of
a failed exchange again and sets CamelFileName, and an unknown operation fails
With the `TimeStamp` (default) and `PersistentTimestamp` query strategies, the
`mongodb-gridfs` consumer routed
every file uploaded since the consumer started again at each poll (every
`delay`, 500 ms by default). The upload
@@ -4337,6 +4337,31 @@ visible later: GridFS sets the upload date on the
client. Use `TimeStampAndFileA
`PersistentTimestampAndFileAttribute` if such files must be routed: these
strategies keep the start date in their
query, as before, and mark each file so that it is routed once.
+When the exchange of a file fails, the consumer no longer marks the file as
processed or moves its timestamp past the
+file: it reports the failure to its exception handler (which logs it by
default), and the next poll offers the file
+again. Before, the failure was ignored and the file was never offered again. A
failure that the route handles, such as
+with a dead letter channel, still completes the file. With the `TimeStamp` and
`PersistentTimestamp` query strategies,
+the files of a poll are now routed in the order of their upload date, and the
poll stops at a file whose exchange
+fails, so a file that always fails holds back the files uploaded after it:
handle such failures in the route, or use a
+strategy with a file attribute, which goes on with the other files. Only the
first failure of a file is reported to
+the exception handler: when the file fails again at the next poll, the
consumer logs the failure at `DEBUG`, so a file
+that always fails is reported once (it is reported again when it fails after a
poll that routed it or did not offer
+it). The `exceptionHandler` and `bridgeErrorHandler` options now apply to the
consumer; before, they had no effect.
+
+The consumer now sets the `CamelFileName` header to the name of the file. A
route that sends the message on to an
+endpoint that uses this header, such as a `file` endpoint, now writes the file
under that name instead of a generated
+one. Remove the header before that endpoint to keep the generated name.
+
+The consumer now closes the stream of the content of the file when the
exchange is done. With stream caching turned off
+(it is on by default), read the stream before the exchange is done.
+
+The producer now fails with an `IllegalArgumentException` that lists the
supported operations when the `operation`
+option or the `CamelGridFsOperation` header is not `create`, `remove`,
`findOne`, `listAll` or `count`. Before, a
+misspelled operation, such as `findone`, did nothing and the exchange
succeeded.
+
+`listAll` now gives the id of each file as the hex string of its ObjectId,
such as `665f1c2e8b3e4a5d6c7b8a90`. Before,
+each line held `BsonObjectId{value=665f1c2e8b3e4a5d6c7b8a90}`.
+
=== camel-paho-mqtt5, camel-paho - consumers that share a client get the
messages of their own topic
A client has a single callback, and each consumer set it on its client. The
consumers of several endpoints that use
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GridFsEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GridFsEndpointBuilderFactory.java
index e9085cb32562..6a162c465846 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GridFsEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GridFsEndpointBuilderFactory.java
@@ -842,7 +842,7 @@ public interface GridFsEndpointBuilderFactory {
return "CamelFileLength";
}
/**
- * The size of the file.
+ * The upload date of the file.
*
* The option is a: {@code Date} type.
*
@@ -854,11 +854,13 @@ public interface GridFsEndpointBuilderFactory {
return "CamelFileLastModified";
}
/**
- * The name of the file.
+ * (producer) The name of the file: the name that create gives the
file,
+ * and the file that findOne, remove, listAll and count look for.
+ * (consumer) The name of the file that was found.
*
* The option is a: {@code String} type.
*
- * Group: producer
+ * Group: common
*
* @return the name of the header {@code FileName}.
*/
