[camel] branch master updated: camel-core - Optimize Idempotent Consumer EIP

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/master by this push:
     new af188bd  camel-core - Optimize Idempotent Consumer EIP
af188bd is described below

commit af188bd294129dd50a7f9b0008fe5c79ecad75f3
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Mar 15 07:44:09 2021 +0100

    camel-core - Optimize Idempotent Consumer EIP
---
 .../processor/idempotent/IdempotentConsumer.java   | 25 +++++++++++-----------
 1 file changed, 12 insertions(+), 13 deletions(-)

diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java
index 04a20a5..a425c4f 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java
@@ -155,8 +155,13 @@ public class IdempotentConsumer extends AsyncProcessorSupport
 
             final Synchronization onCompletion
                     = new IdempotentOnCompletion(idempotentRepository, messageId, eager, removeOnFailure);
-            target = new IdempotentConsumerCallback(exchange, onCompletion, callback, completionEager);
-            if (!completionEager) {
+
+            if (completionEager) {
+                // the callback will eager complete
+                target = new IdempotentConsumerCallback(exchange, onCompletion, callback);
+            } else {
+                // we can use existing callback as target
+                target = callback;
                 // the scope is to do the idempotent completion work as an unit of work on the exchange when its done being routed
                 exchange.adapt(ExtendedExchange.class).addOnCompletion(onCompletion);
             }
@@ -279,27 +284,21 @@ public class IdempotentConsumer extends AsyncProcessorSupport
         private final Exchange exchange;
         private final Synchronization onCompletion;
         private final AsyncCallback callback;
-        private final boolean completionEager;
 
-        IdempotentConsumerCallback(Exchange exchange, Synchronization onCompletion, AsyncCallback callback,
-                                   boolean completionEager) {
+        IdempotentConsumerCallback(Exchange exchange, Synchronization onCompletion, AsyncCallback callback) {
             this.exchange = exchange;
             this.onCompletion = onCompletion;
             this.callback = callback;
-            this.completionEager = completionEager;
         }
 
         @Override
         public void done(boolean doneSync) {
             try {
-                if (completionEager) {
-                    if (exchange.isFailed()) {
-                        onCompletion.onFailure(exchange);
-                    } else {
-                        onCompletion.onComplete(exchange);
-                    }
+                if (exchange.isFailed()) {
+                    onCompletion.onFailure(exchange);
+                } else {
+                    onCompletion.onComplete(exchange);
                 }
-                // if completion is not eager then the onCompletion is invoked as part of the UoW of the Exchange
             } finally {
                 callback.done(doneSync);
             }