davsclaus commented on code in PR #28035:
URL: https://github.com/apache/camel/pull/28035#discussion_r4306101172
##########
components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java:
##########
@@ -323,15 +353,51 @@ protected void
removeExcessiveInProgressFiles(List<GenericFile<T>> files) {
}
}
+ /**
+ * The files left by the last poll while a done file with a fixed name was
there, or <tt>null</tt> if not used (no
+ * such done file, or a custom processor that does not complete the files).
+ */
+ GenericFileDoneFileAdmission getDoneFileAdmission() {
+ return customProcessor == null ? doneFileAdmission : null;
+ }
+
+ private void startDoneFileBatch(GenericFileDoneFileAdmission admission,
List<GenericFile<T>> files) {
+ Set<String> doneFiles = new HashSet<>();
+ for (GenericFile<T> file : files) {
+
doneFiles.add(endpoint.createDoneFileName(file.getAbsoluteFilePath()));
+ }
+ List<String> gone = admission.startBatch(doneFiles);
+ if (gone != null) {
+ // the done file was kept for files that are no longer there, so
it is no longer needed
+ for (String doneFileName : gone) {
+ try {
+ boolean deleted = operations.deleteFile(doneFileName);
+ LOG.trace("Done file: {} kept for files that are gone was
deleted: {}", doneFileName, deleted);
+ } catch (Exception e) {
+ LOG.debug("Cannot delete done file: {} due to: {}",
doneFileName, e.getMessage(), e);
+ }
+ }
+ }
+ }
+
+ private void doneFileLeft(GenericFile<?> file) {
+ GenericFileDoneFileAdmission admission = getDoneFileAdmission();
+ if (admission != null) {
+ String absoluteFileName = file.getAbsoluteFilePath();
+ admission.left(endpoint.createDoneFileName(absoluteFileName),
absoluteFileName);
+ }
+ }
+
/**
* Whether or not we can continue polling for more files
*
* @param fileList the current list of gathered files
* @return <tt>true</tt> to continue, <tt>false</tt> to stop due
hitting maxMessagesPerPoll limit
*/
public boolean canPollMoreFiles(List<?> fileList) {
- // at this point we should not limit if we are not eager
- if (!eagerLimitMaxMessagesPerPoll) {
+ // at this point we should not limit if we are not eager, or when a
done file with a fixed name is used, as
+ // the files beyond the limit must be known to keep the done file for
them (the list is cut after polling)
+ if (!eagerLimitMaxMessagesPerPoll || getDoneFileAdmission() != null) {
Review Comment:
`GenericFilePollingConsumer` sets `maxMessagesPerPoll=1` on its consumer, so
pollEnrich / `ConsumerTemplate.receive` with a fixed `doneFileName` now take
this path on every receive: a full directory listing, the gate, and a done file
kept across receives (on main the first receive deletes it). Worth a test and a
line in the upgrade guide.
##########
components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java:
##########
@@ -914,6 +982,13 @@ protected boolean isMatched(
if (!isMatched(file, doneFileName, files)) {
return false;
}
+
+ // a done file kept for files left by the last poll only admits
these files, not the ones written after it
+ GenericFileDoneFileAdmission admission = getDoneFileAdmission();
+ if (admission != null && !admission.isAdmitted(doneFileName,
absoluteFilePath)) {
Review Comment:
This is the part I am unsure about. `GenericFileProducer.writeDoneFile`
deletes and rewrites a fixed done file after every written file, so with a
Camel producer on the other side, the files it writes while the left files
drain are rejected here. The done file it rewrote for them is then deleted in
`handleDoneFile` once `left` is empty, and they wait for the next write. On
main they would be picked up by the next poll. Could the gate be lifted when
the done file was rewritten (timestamp), or dropped entirely?
##########
components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java:
##########
@@ -259,6 +286,8 @@ public int processBatch(Queue<Object> exchanges) {
// if we did not start process the file then decrement the counter
if (!started) {
+ // the done file must stay for this file, also if it is not
started as the last file of the batch
+ doneFileLeft(getExchangeFileProperty(exchange));
Review Comment:
Small one: when the download fails after `preMove`, `processExchange`
returns false, but the exchange file has already been rebound to the pre-moved
file (`GenericFileRenameProcessStrategy.begin` -> `bindToExchange`). The left
entry is then keyed by the done file of the preMove directory, and the next
poll's `startDoneFileBatch` would try to delete `<preMoveDir>/<doneFileName>`.
Capturing `getExchangeFileProperty(exchange)` before calling
`processExchange`/`customProcessExchange`, and using that here, avoids it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
