You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2020/08/03 06:31:44 UTC
[camel] branch master updated (9d0d2ab -> c46f7d7)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 9d0d2ab Sync Properties
new a8bdd19 camel-core - Optimize choice eip
new c46f7d7 Removed todo
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/camel/processor/ChoiceProcessor.java | 65 +++++++++++-----------
.../camel/processor/DelayProcessorSupport.java | 2 -
2 files changed, 31 insertions(+), 36 deletions(-)
[camel] 01/02: camel-core - Optimize choice eip
Posted by 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
commit a8bdd194e325dffd879303e02df7c2226e152883
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Aug 3 07:28:06 2020 +0200
camel-core - Optimize choice eip
---
.../apache/camel/processor/ChoiceProcessor.java | 65 +++++++++++-----------
1 file changed, 31 insertions(+), 34 deletions(-)
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/ChoiceProcessor.java b/core/camel-base/src/main/java/org/apache/camel/processor/ChoiceProcessor.java
index b2705ef..ac86616 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/ChoiceProcessor.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/ChoiceProcessor.java
@@ -17,7 +17,6 @@
package org.apache.camel.processor;
import java.util.ArrayList;
-import java.util.Iterator;
import java.util.List;
import org.apache.camel.AsyncCallback;
@@ -48,18 +47,16 @@ public class ChoiceProcessor extends AsyncProcessorSupport implements Navigate<P
private String id;
private String routeId;
private final List<FilterProcessor> filters;
- private final Processor otherwise;
+ private final AsyncProcessor otherwise;
private transient long notFiltered;
public ChoiceProcessor(List<FilterProcessor> filters, Processor otherwise) {
this.filters = filters;
- this.otherwise = otherwise;
+ this.otherwise = AsyncProcessorConverterHelper.convert(otherwise);;
}
@Override
public boolean process(final Exchange exchange, final AsyncCallback callback) {
- Iterator<Processor> processors = next().iterator();
-
// callback to restore existing FILTER_MATCHED property on the Exchange
final Object existing = exchange.getProperty(Exchange.FILTER_MATCHED);
final AsyncCallback choiceCallback = new AsyncCallback() {
@@ -74,30 +71,16 @@ public class ChoiceProcessor extends AsyncProcessorSupport implements Navigate<P
}
};
- // as we only pick one processor to process, then no need to have async callback that has a while loop as well
- // as this should not happen, eg we pick the first filter processor that matches, or the otherwise (if present)
- // and if not, we just continue without using any processor
- while (processors.hasNext()) {
- // get the next processor
- Processor processor = processors.next();
-
+ // find the first matching filter and process the exchange using it
+ for (FilterProcessor filter : filters) {
// evaluate the predicate on filter predicate early to be faster
// and avoid issues when having nested choices
// as we should only pick one processor
boolean matches = false;
- if (processor instanceof FilterProcessor) {
- FilterProcessor filter = (FilterProcessor) processor;
- try {
- matches = filter.matches(exchange);
- // as we have pre evaluated the predicate then use its processor directly when routing
- processor = filter.getProcessor();
- } catch (Throwable e) {
- exchange.setException(e);
- }
- } else {
- // its the otherwise processor, so its a match
- notFiltered++;
- matches = true;
+ try {
+ matches = filter.matches(exchange);
+ } catch (Throwable e) {
+ exchange.setException(e);
}
// check for error if so we should break out
@@ -110,14 +93,19 @@ public class ChoiceProcessor extends AsyncProcessorSupport implements Navigate<P
continue;
}
- // okay we found a filter or its the otherwise we are processing
- AsyncProcessor async = AsyncProcessorConverterHelper.convert(processor);
- return async.process(exchange, choiceCallback);
+ // okay we found a filter then process it directly via its processor as we have already done the matching
+ return filter.getProcessor().process(exchange, choiceCallback);
}
- // when no filter matches and there is no otherwise, then just continue
- choiceCallback.done(true);
- return true;
+ if (otherwise != null) {
+ // no filter matched then use otherwise
+ notFiltered++;
+ return otherwise.process(exchange, choiceCallback);
+ } else {
+ // when no filter matches and there is no otherwise, then just continue
+ choiceCallback.done(true);
+ return true;
+ }
}
@Override
@@ -149,7 +137,7 @@ public class ChoiceProcessor extends AsyncProcessorSupport implements Navigate<P
* Reset counters.
*/
public void reset() {
- for (FilterProcessor filter : getFilters()) {
+ for (FilterProcessor filter : filters) {
filter.reset();
}
notFiltered = 0;
@@ -161,7 +149,7 @@ public class ChoiceProcessor extends AsyncProcessorSupport implements Navigate<P
return null;
}
List<Processor> answer = new ArrayList<>();
- if (filters != null) {
+ if (!filters.isEmpty()) {
answer.addAll(filters);
}
if (otherwise != null) {
@@ -172,7 +160,7 @@ public class ChoiceProcessor extends AsyncProcessorSupport implements Navigate<P
@Override
public boolean hasNext() {
- return otherwise != null || (filters != null && !filters.isEmpty());
+ return otherwise != null || !filters.isEmpty();
}
@Override
@@ -196,6 +184,11 @@ public class ChoiceProcessor extends AsyncProcessorSupport implements Navigate<P
}
@Override
+ protected void doInit() throws Exception {
+ ServiceHelper.initService(filters, otherwise);
+ }
+
+ @Override
protected void doStart() throws Exception {
ServiceHelper.startService(filters, otherwise);
}
@@ -205,4 +198,8 @@ public class ChoiceProcessor extends AsyncProcessorSupport implements Navigate<P
ServiceHelper.stopService(otherwise, filters);
}
+ @Override
+ protected void doShutdown() throws Exception {
+ ServiceHelper.stopAndShutdownServices(otherwise, filters);
+ }
}
[camel] 02/02: Removed todo
Posted by 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
commit c46f7d707d007a0941b31f46387b4bb5835b33db
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Aug 3 07:43:14 2020 +0200
Removed todo
---
.../src/main/java/org/apache/camel/processor/DelayProcessorSupport.java | 2 --
1 file changed, 2 deletions(-)
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java b/core/camel-base/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java
index 5e7aea9..069c3cf 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java
@@ -47,8 +47,6 @@ public abstract class DelayProcessorSupport extends DelegateAsyncProcessor {
private boolean callerRunsWhenRejected = true;
private final AtomicInteger delayedCount = new AtomicInteger(0);
- // TODO: Add option to cancel tasks on shutdown so we can stop fast
-
private final class ProcessCall implements Runnable {
private final Exchange exchange;
private final AsyncCallback callback;