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/01/19 10:59:03 UTC
[camel] branch master updated (c892e86 -> f3eca85)
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 c892e86 CAMEL-14354: Optimize core and fixed test
new 3fbd153 CAMEL-14354: Optimize core
new 42a045b CAMEL-14354: Optimize core
new f3eca85 CAMEL-14354: Optimize core
The 3 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:
.../java/org/apache/camel/processor/Pipeline.java | 18 +++++------
.../org/apache/camel/processor/SendProcessor.java | 35 ++++++++++++----------
.../processor/SharedCamelInternalProcessor.java | 31 ++++++++++++++-----
3 files changed, 51 insertions(+), 33 deletions(-)
[camel] 02/03: CAMEL-14354: Optimize core
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 42a045be77a182d95be47cdf62f4e6e0c55bdb3c
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Sun Jan 19 11:10:35 2020 +0100
CAMEL-14354: Optimize core
---
.../processor/SharedCamelInternalProcessor.java | 31 ++++++++++++++++------
1 file changed, 23 insertions(+), 8 deletions(-)
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/SharedCamelInternalProcessor.java b/core/camel-base/src/main/java/org/apache/camel/processor/SharedCamelInternalProcessor.java
index a278187..61f3177 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/SharedCamelInternalProcessor.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/SharedCamelInternalProcessor.java
@@ -69,13 +69,23 @@ import org.slf4j.LoggerFactory;
public class SharedCamelInternalProcessor {
private static final Logger LOG = LoggerFactory.getLogger(SharedCamelInternalProcessor.class);
- private final List<CamelInternalProcessorAdvice> advices = new ArrayList<>();
+ private static final Object[] EMPTY_STATES = new Object[0];
+ private final List<CamelInternalProcessorAdvice> advices;
+ private byte statefulAdvices;
public SharedCamelInternalProcessor(CamelInternalProcessorAdvice... advices) {
if (advices != null) {
- this.advices.addAll(Arrays.asList(advices));
+ this.advices = new ArrayList<>(advices.length);
+ for (CamelInternalProcessorAdvice advice : advices) {
+ this.advices.add(advice);
+ if (advice.hasState()) {
+ statefulAdvices++;
+ }
+ }
// ensure advices are sorted so they are in the order we want
this.advices.sort(OrderedComparator.get());
+ } else {
+ this.advices = null;
}
}
@@ -128,14 +138,16 @@ public class SharedCamelInternalProcessor {
return true;
}
- // optimise to use object array for states
- final Object[] states = new Object[advices.size()];
+ // optimise to use object array for states, and only for the number of advices that keep state
+ final Object[] states = statefulAdvices > 0 ? new Object[statefulAdvices] : EMPTY_STATES;
// optimise for loop using index access to avoid creating iterator object
- for (int i = 0; i < advices.size(); i++) {
+ for (int i = 0, j = 0; i < advices.size(); i++) {
CamelInternalProcessorAdvice task = advices.get(i);
try {
Object state = task.before(exchange);
- states[i] = state;
+ if (task.hasState()) {
+ states[j++] = state;
+ }
} catch (Throwable e) {
exchange.setException(e);
originalCallback.done(true);
@@ -239,9 +251,12 @@ public class SharedCamelInternalProcessor {
// we should call after in reverse order
try {
- for (int i = advices.size() - 1; i >= 0; i--) {
+ for (int i = advices.size() - 1, j = states.length - 1; i >= 0; i--) {
CamelInternalProcessorAdvice task = advices.get(i);
- Object state = states[i];
+ Object state = null;
+ if (task.hasState()) {
+ state = states[j--];
+ }
try {
task.after(exchange, state);
} catch (Throwable e) {
[camel] 03/03: CAMEL-14354: Optimize core
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 f3eca85d24736dae6afd0c6cf1b57bbe85affb71
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Sun Jan 19 11:50:10 2020 +0100
CAMEL-14354: Optimize core
---
.../main/java/org/apache/camel/processor/Pipeline.java | 18 +++++++++---------
1 file changed, 9 insertions(+), 9 deletions(-)
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/Pipeline.java b/core/camel-base/src/main/java/org/apache/camel/processor/Pipeline.java
index e646a45..8e2f5f3 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/Pipeline.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/Pipeline.java
@@ -18,7 +18,6 @@ package org.apache.camel.processor;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Iterator;
import java.util.List;
import java.util.stream.Collectors;
@@ -83,25 +82,26 @@ public class Pipeline extends AsyncProcessorSupport implements Navigate<Processo
@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
if (exchange.isTransacted()) {
- camelContext.getReactiveExecutor().scheduleSync(() -> Pipeline.this.doProcess(exchange, callback, processors.iterator(), true));
+ camelContext.getReactiveExecutor().scheduleSync(() -> Pipeline.this.doProcess(exchange, callback, processors, 0, true));
} else {
- camelContext.getReactiveExecutor().scheduleMain(() -> Pipeline.this.doProcess(exchange, callback, processors.iterator(), true));
+ camelContext.getReactiveExecutor().scheduleMain(() -> Pipeline.this.doProcess(exchange, callback, processors, 0, true));
}
return false;
}
- protected void doProcess(Exchange exchange, AsyncCallback callback, Iterator<AsyncProcessor> processors, boolean first) {
- if (continueRouting(processors, exchange)
+ protected void doProcess(Exchange exchange, AsyncCallback callback, List<AsyncProcessor> processors, int index, boolean first) {
+ if (continueRouting(processors, index, exchange)
&& (first || continueProcessing(exchange, "so breaking out of pipeline", log))) {
// prepare for next run
ExchangeHelper.prepareOutToIn(exchange);
// get the next processor
- AsyncProcessor processor = processors.next();
+ AsyncProcessor processor = processors.get(index);
+ final Integer idx = index + 1;
processor.process(exchange, doneSync ->
- camelContext.getReactiveExecutor().schedule(() -> doProcess(exchange, callback, processors, false)));
+ camelContext.getReactiveExecutor().schedule(() -> doProcess(exchange, callback, processors, idx, false)));
} else {
ExchangeHelper.copyResults(exchange, exchange);
@@ -114,7 +114,7 @@ public class Pipeline extends AsyncProcessorSupport implements Navigate<Processo
}
}
- protected boolean continueRouting(Iterator<AsyncProcessor> it, Exchange exchange) {
+ protected boolean continueRouting(List<AsyncProcessor> processors, int index, Exchange exchange) {
Object stop = exchange.getProperty(Exchange.ROUTE_STOP);
if (stop != null) {
boolean doStop = exchange.getContext().getTypeConverter().convertTo(Boolean.class, stop);
@@ -124,7 +124,7 @@ public class Pipeline extends AsyncProcessorSupport implements Navigate<Processo
}
}
// continue if there are more processors to route
- boolean answer = it.hasNext();
+ boolean answer = index < processors.size();
log.trace("ExchangeId: {} should continue routing: {}", exchange.getExchangeId(), answer);
return answer;
}
[camel] 01/03: CAMEL-14354: Optimize core
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 3fbd153c41c88ad496232c5471511b2f8948400c
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Sun Jan 19 10:32:42 2020 +0100
CAMEL-14354: Optimize core
---
.../org/apache/camel/processor/SendProcessor.java | 35 ++++++++++++----------
1 file changed, 19 insertions(+), 16 deletions(-)
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/SendProcessor.java b/core/camel-base/src/main/java/org/apache/camel/processor/SendProcessor.java
index 1aaa98f..6c20e6c 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/SendProcessor.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/SendProcessor.java
@@ -140,24 +140,27 @@ public class SendProcessor extends AsyncProcessorSupport implements Traceable, E
watch = null;
}
- try {
- log.debug(">>>> {} {}", destination, exchange);
- return producer.process(exchange, new AsyncCallback() {
- @Override
- public void done(boolean doneSync) {
- try {
- // restore previous MEP
- target.setPattern(existingPattern);
- // emit event that the exchange was sent to the endpoint
- if (watch != null) {
- long timeTaken = watch.taken();
- EventHelper.notifyExchangeSent(target.getContext(), target, destination, timeTaken);
- }
- } finally {
- callback.done(doneSync);
+ // optimize to only create a new callback if really needed, otherwise we can use the provided callback as-is
+ AsyncCallback ac = callback;
+ boolean newCallback = watch != null || existingPattern != target.getPattern();
+ if (newCallback) {
+ ac = doneSync -> {
+ try {
+ // restore previous MEP
+ target.setPattern(existingPattern);
+ // emit event that the exchange was sent to the endpoint
+ if (watch != null) {
+ long timeTaken = watch.taken();
+ EventHelper.notifyExchangeSent(target.getContext(), target, destination, timeTaken);
}
+ } finally {
+ callback.done(doneSync);
}
- });
+ };
+ }
+ try {
+ log.debug(">>>> {} {}", destination, exchange);
+ return producer.process(exchange, ac);
} catch (Throwable throwable) {
exchange.setException(throwable);
callback.done(true);