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 9141da28a7da CAMEL-25859: camel-mongodb-gridfs - the consumer routes
each found file with its own content (#28381)
9141da28a7da is described below
commit 9141da28a7da09a4c1a1a781c05ee5a83342a466
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 24 19:13:26 2026 +0530
CAMEL-25859: camel-mongodb-gridfs - the consumer routes each found file
with its own content (#28381)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/mongodb/gridfs/GridFsConsumer.java | 3 +-
.../gridfs/GridFsConsumerTimestampTest.java | 44 +++++++++++++++++++---
2 files changed, 40 insertions(+), 7 deletions(-)
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 2e6717382c23..439fc15b4664 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
@@ -209,7 +209,8 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
}
if (fOrig != null) {
Exchange exchange = createExchange(true);
- GridFSDownloadStream downloadStream =
endpoint.getGridFsBucket().openDownloadStream(file.getFilename());
+ // 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();
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 4994049fb485..8f6cd8fd17cf 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
@@ -17,6 +17,7 @@
package org.apache.camel.component.mongodb.gridfs;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
@@ -75,6 +76,8 @@ class GridFsConsumerTimestampTest {
private final List<GridFSFile> stored = new CopyOnWriteArrayList<>();
private final List<String> routed = new CopyOnWriteArrayList<>();
+ // the ids of the files whose content was routed
+ private final List<BsonValue> routedIds = new CopyOnWriteArrayList<>();
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();
@@ -116,12 +119,15 @@ class GridFsConsumerTimestampTest {
polls.countDown();
return iterableOf(found);
});
+ // as MongoDB: by file name, the newest revision of the files with
that name; by id, that file
when(bucket.openDownloadStream(anyString())).thenAnswer(invocation -> {
String name = invocation.getArgument(0);
- GridFSDownloadStream stream = mock(GridFSDownloadStream.class);
- when(stream.getGridFSFile())
- .thenReturn(stored.stream().filter(f ->
f.getFilename().equals(name)).findFirst().orElseThrow());
- return stream;
+ return downloadStreamOf(stored.stream().filter(f ->
f.getFilename().equals(name))
+
.max(Comparator.comparing(GridFSFile::getUploadDate)).orElseThrow());
+ });
+
when(bucket.openDownloadStream(any(BsonValue.class))).thenAnswer(invocation -> {
+ BsonValue id = invocation.getArgument(0);
+ return downloadStreamOf(stored.stream().filter(f ->
f.getId().equals(id)).findFirst().orElseThrow());
});
// the consumer marks a file before it routes it: only a file that is
not marked yet is returned
when(files.findOneAndUpdate(any(Bson.class), any(Bson.class),
any(FindOneAndUpdateOptions.class)))
@@ -176,6 +182,23 @@ class GridFsConsumerTimestampTest {
assertEquals(List.of(first.getUploadDate()), persisted, "the timestamp
is stored when it moves, not at every poll");
}
+ @ParameterizedTest
+ @EnumSource(QueryStrategy.class)
+ void filesWithTheSameNameAreEachRoutedWithTheirOwnContent(QueryStrategy
strategy) throws Exception {
+ startConsumer(strategy);
+
+ // two revisions of a file name, both new for the same poll (uploaded
within one delay, or while the
+ // consumer was stopped): GridFS keeps both, and each is a file of its
own
+ GridFSFile first = file("report.csv", 60);
+ GridFSFile second = file("report.csv", 61);
+ stored.add(first);
+ stored.add(second);
+ awaitFullPolls(2);
+
+ assertEquals(List.of(first.getId(), second.getId()), routedIds,
+ "each file found by the poll must be routed with its own
content, not with the newest file of that name");
+ }
+
@Test
void timestampStrategySkipsAFileWithAnOlderUploadDateThanARoutedFile()
throws Exception {
startConsumer(QueryStrategy.TimeStamp);
@@ -304,8 +327,11 @@ class GridFsConsumerTimestampTest {
endpoint.setInitialDelay(1);
endpoint.setDelay(delay);
- Processor processor = exchange ->
routed.add(exchange.getMessage().getBody(GridFSDownloadStream.class)
- .getGridFSFile().getFilename());
+ Processor processor = exchange -> {
+ GridFSFile file =
exchange.getMessage().getBody(GridFSDownloadStream.class).getGridFSFile();
+ routed.add(file.getFilename());
+ routedIds.add(file.getId());
+ };
consumer = new GridFsConsumer(endpoint, processor);
consumer.setExceptionHandler(new ExceptionHandler() {
@Override
@@ -336,6 +362,12 @@ class GridFsConsumerTimestampTest {
assertTrue(latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), "the
consumer did not poll " + (count + 1) + " times");
}
+ private static GridFSDownloadStream downloadStreamOf(GridFSFile file) {
+ GridFSDownloadStream stream = mock(GridFSDownloadStream.class);
+ when(stream.getGridFSFile()).thenReturn(file);
+ return stream;
+ }
+
/**
* A file uploaded the given number of seconds after the test (and so the
consumer) started.
*/
