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 2021/04/06 10:05:27 UTC
[camel] branch master updated (a9fd03f -> 0b0e388)
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 a9fd03f CAMEL-16454: camel-core - Optimize CircuitBreaker constants for fast exchange properties
new 9e6f960 CAMEL-16451: camel-core - ExchangePooling for EIPs. Enricher EIP
new 633c7ab CAMEL-16457: camel-core - Optimize ExchangeHelper.copyResults - No need to copy message id not in use
new d739e25 CAMEL-16457: camel-core - Optimize ExchangeHelper.copyResults - No need to copy message id not in use
new 0b0e388 CAMEL-16451: camel-core - ExchangePooling for EIPs. Enricher EIP
The 4 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:
.../src/main/java/org/apache/camel/Message.java | 5 ++++
.../java/org/apache/camel/processor/Enricher.java | 32 +++++++++++++++-------
.../FileConsumerThreadsInProgressIssueTest.java | 8 +++---
.../org/apache/camel/support/MessageSupport.java | 10 +++++--
4 files changed, 39 insertions(+), 16 deletions(-)
[camel] 04/04: CAMEL-16451: camel-core - ExchangePooling for EIPs.
Enricher 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 0b0e3882fac9fc81d22563ffc9e05845639061b7
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue Apr 6 11:48:35 2021 +0200
CAMEL-16451: camel-core - ExchangePooling for EIPs. Enricher EIP
---
.../src/main/java/org/apache/camel/processor/Enricher.java | 2 ++
1 file changed, 2 insertions(+)
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java
index 8c5095a..847fe72 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java
@@ -421,6 +421,8 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA
// create a per processor exchange factory
this.processorExchangeFactory = getCamelContext().adapt(ExtendedCamelContext.class)
.getProcessorExchangeFactory().newProcessorExchangeFactory(this);
+ this.processorExchangeFactory.setRouteId(getRouteId());
+ this.processorExchangeFactory.setId(getId());
if (aggregationStrategy == null) {
aggregationStrategy = defaultAggregationStrategy();
[camel] 03/04: CAMEL-16457: camel-core - Optimize
ExchangeHelper.copyResults - No need to copy message id not in use
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 d739e2544c3ec820eef1801aed86bcaad1c16d39
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue Apr 6 11:48:20 2021 +0200
CAMEL-16457: camel-core - Optimize ExchangeHelper.copyResults - No need to copy message id not in use
---
.../component/file/FileConsumerThreadsInProgressIssueTest.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerThreadsInProgressIssueTest.java b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerThreadsInProgressIssueTest.java
index e9286a0..5b53f9f 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerThreadsInProgressIssueTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerThreadsInProgressIssueTest.java
@@ -101,14 +101,14 @@ public class FileConsumerThreadsInProgressIssueTest extends ContextTestSupport {
@Override
public void process(Exchange exchange) throws Exception {
- Integer integer = duplicate.get(exchange.toString());
+ Integer integer = duplicate.get(exchange.getExchangeId());
if (integer == null) {
- duplicate.put(exchange.toString(), 1);
+ duplicate.put(exchange.getExchangeId(), 1);
} else {
integer++;
- duplicate.put(exchange.toString(), integer);
+ duplicate.put(exchange.getExchangeId(), integer);
}
- log.info("Process called for-" + exchange);
+ log.info("Process called for-" + exchange.getExchangeId());
Thread.sleep(20);
}
[camel] 02/04: CAMEL-16457: camel-core - Optimize
ExchangeHelper.copyResults - No need to copy message id not in use
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 633c7ab53113a75ce89994f5b8e29bfbc9824497
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue Apr 6 11:46:17 2021 +0200
CAMEL-16457: camel-core - Optimize ExchangeHelper.copyResults - No need to copy message id not in use
---
core/camel-api/src/main/java/org/apache/camel/Message.java | 5 +++++
.../src/main/java/org/apache/camel/support/MessageSupport.java | 10 ++++++++--
2 files changed, 13 insertions(+), 2 deletions(-)
diff --git a/core/camel-api/src/main/java/org/apache/camel/Message.java b/core/camel-api/src/main/java/org/apache/camel/Message.java
index df8dcbc..a2db06f 100644
--- a/core/camel-api/src/main/java/org/apache/camel/Message.java
+++ b/core/camel-api/src/main/java/org/apache/camel/Message.java
@@ -60,6 +60,11 @@ public interface Message {
void setMessageId(String messageId);
/**
+ * Whether the message has any message ID assigned.
+ */
+ boolean hasMessageId();
+
+ /**
* Returns the exchange this message is related to
*
* @return the exchange
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/MessageSupport.java b/core/camel-support/src/main/java/org/apache/camel/support/MessageSupport.java
index 48fcd71..f8f3e8b 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/MessageSupport.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/MessageSupport.java
@@ -218,9 +218,10 @@ public abstract class MessageSupport implements Message, CamelContextAware, Data
setExchange(that.getExchange());
}
+ if (that.hasMessageId()) {
+ setMessageId(that.getMessageId());
+ }
// should likely not set DataType as the new body may be a different type than the original body
-
- setMessageId(that.getMessageId());
setBody(newBody);
// the headers may be the same instance if the end user has made some mistake
@@ -288,6 +289,11 @@ public abstract class MessageSupport implements Message, CamelContextAware, Data
this.messageId = messageId;
}
+ @Override
+ public boolean hasMessageId() {
+ return messageId != null;
+ }
+
/**
* Allow implementations to auto-create a messageId
*/
[camel] 01/04: CAMEL-16451: camel-core - ExchangePooling for EIPs.
Enricher 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 9e6f960a9f764b8ed7a5bd20a648e718b89e1fd4
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue Apr 6 11:35:40 2021 +0200
CAMEL-16451: camel-core - ExchangePooling for EIPs. Enricher EIP
---
.../java/org/apache/camel/processor/Enricher.java | 30 ++++++++++++++--------
1 file changed, 20 insertions(+), 10 deletions(-)
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java
index 41232dd..8c5095a 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java
@@ -34,6 +34,7 @@ import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.spi.EndpointUtilizationStatistics;
import org.apache.camel.spi.IdAware;
import org.apache.camel.spi.NormalizedEndpointUri;
+import org.apache.camel.spi.ProcessorExchangeFactory;
import org.apache.camel.spi.ProducerCache;
import org.apache.camel.spi.RouteIdAware;
import org.apache.camel.support.AsyncProcessorConverterHelper;
@@ -75,6 +76,7 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA
private boolean shareUnitOfWork;
private int cacheSize;
private boolean ignoreInvalidEndpoint;
+ private ProcessorExchangeFactory processorExchangeFactory;
public Enricher(Expression expression) {
this.expression = expression;
@@ -246,9 +248,6 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA
} catch (Throwable e) {
// if the aggregationStrategy threw an exception, set it on the original exchange
exchange.setException(new CamelExchangeException("Error occurred during aggregation", exchange, e));
- callback.done(false);
- // we failed so break out now
- return;
}
}
@@ -266,6 +265,9 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA
ServiceHelper.stopAndShutdownService(endpoint);
}
+ // and release resource exchange back in pool
+ processorExchangeFactory.release(resourceExchange);
+
callback.done(false);
}
});
@@ -308,9 +310,6 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA
} catch (Throwable e) {
// if the aggregationStrategy threw an exception, set it on the original exchange
exchange.setException(new CamelExchangeException("Error occurred during aggregation", exchange, e));
- callback.done(true);
- // we failed so break out now
- return true;
}
}
@@ -328,6 +327,9 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA
ServiceHelper.stopAndShutdownService(endpoint);
}
+ // and release resource exchange back in pool
+ processorExchangeFactory.release(resourceExchange);
+
callback.done(true);
return true;
}
@@ -387,7 +389,7 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA
*/
protected Exchange createResourceExchange(Exchange source, ExchangePattern pattern) {
// copy exchange, and do not share the unit of work
- Exchange target = ExchangeHelper.createCorrelatedCopy(source, false);
+ Exchange target = processorExchangeFactory.createCorrelatedCopy(source, false);
target.setPattern(pattern);
// if we share unit of work, we need to prepare the resource exchange
@@ -415,14 +417,22 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA
}
@Override
- protected void doStart() throws Exception {
+ protected void doBuild() throws Exception {
+ // create a per processor exchange factory
+ this.processorExchangeFactory = getCamelContext().adapt(ExtendedCamelContext.class)
+ .getProcessorExchangeFactory().newProcessorExchangeFactory(this);
+
if (aggregationStrategy == null) {
aggregationStrategy = defaultAggregationStrategy();
}
if (aggregationStrategy instanceof CamelContextAware) {
((CamelContextAware) aggregationStrategy).setCamelContext(camelContext);
}
+ ServiceHelper.buildService(processorExchangeFactory);
+ }
+ @Override
+ protected void doStart() throws Exception {
if (producerCache == null) {
if (cacheSize < 0) {
producerCache = new EmptyProducerCache(this, camelContext);
@@ -433,12 +443,12 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA
}
}
- ServiceHelper.startService(producerCache, aggregationStrategy);
+ ServiceHelper.startService(processorExchangeFactory, producerCache, aggregationStrategy);
}
@Override
protected void doStop() throws Exception {
- ServiceHelper.stopService(aggregationStrategy, producerCache);
+ ServiceHelper.stopService(aggregationStrategy, producerCache, processorExchangeFactory);
}
private static class CopyAggregationStrategy implements AggregationStrategy {