davsclaus commented on code in PR #27671:
URL: https://github.com/apache/camel/pull/27671#discussion_r4240436925
##########
components/camel-google/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumer.java:
##########
@@ -91,17 +95,41 @@ protected int poll() throws Exception {
forceConsumerAsReady();
if (c.getMessages() != null) {
- for (Message message : c.getMessages()) {
- Message mess
- = getClient().users().messages().get("me",
message.getId()).setFormat(messageFormat()).execute();
- Exchange exchange =
createExchange(getEndpoint().getExchangePattern(), mess);
- answer.add(exchange);
+ try {
+ for (Message message : c.getMessages()) {
+ addExchange(message.getId(), answer);
+ }
+ } catch (Exception e) {
+ // none of the messages of this poll is processed
+ releaseInProgress(answer);
+ throw e;
}
}
return processBatch(CastUtils.cast(answer));
}
+ private void addExchange(String id, Queue<Exchange> answer) throws
Exception {
+ // skip the message if an exchange of a previous poll is still
processing it
+ if (!inProgress.add(id)) {
Review Comment:
Suggest snapshotting the in-progress ids before listing and skipping
anything that was in flight at listing time:
```java
Set<String> busy = Set.copyOf(inProgress); // before request.execute()
...
if (busy.contains(id) || !inProgress.add(id)) {
// skip
}
```
--
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]
