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 2016/05/04 12:08:14 UTC
[5/8] camel git commit: CAMEL-9940: ProducerTemplate - Make extract
result set part of UoW
CAMEL-9940: ProducerTemplate - Make extract result set part of UoW
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/1cca6b7c
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/1cca6b7c
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/1cca6b7c
Branch: refs/heads/master
Commit: 1cca6b7c1d5445c3d75ae6b1d3abbf6c1568eafa
Parents: b4a6ad4
Author: Claus Ibsen <da...@apache.org>
Authored: Wed May 4 09:04:17 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed May 4 14:07:56 2016 +0200
----------------------------------------------------------------------
.../camel/impl/DefaultProducerTemplate.java | 40 ++++++++++++---
.../org/apache/camel/impl/ProducerCache.java | 52 ++++++++++++++++----
2 files changed, 76 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/1cca6b7c/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
index aee4973..a5c8867 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
@@ -32,6 +32,7 @@ import org.apache.camel.Message;
import org.apache.camel.NoSuchEndpointException;
import org.apache.camel.Processor;
import org.apache.camel.ProducerTemplate;
+import org.apache.camel.processor.ConvertBodyProcessor;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.support.ServiceSupport;
import org.apache.camel.util.CamelContextHelper;
@@ -324,37 +325,44 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
}
public <T> T requestBody(Object body, Class<T> type) {
- Object answer = requestBody(body);
+ Exchange exchange = producerCache.send(getMandatoryDefaultEndpoint(), ExchangePattern.InOut, createSetBodyProcessor(body), createConvertBodyProcessor(type));
+ Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
public <T> T requestBody(Endpoint endpoint, Object body, Class<T> type) {
- Object answer = requestBody(endpoint, body);
+ Exchange exchange = producerCache.send(endpoint, ExchangePattern.InOut, createSetBodyProcessor(body), createConvertBodyProcessor(type));
+ Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
public <T> T requestBody(String endpointUri, Object body, Class<T> type) {
- Object answer = requestBody(endpointUri, body);
+ Exchange exchange = producerCache.send(resolveMandatoryEndpoint(endpointUri), ExchangePattern.InOut, createSetBodyProcessor(body), createConvertBodyProcessor(type));
+ Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
public <T> T requestBodyAndHeader(Endpoint endpoint, Object body, String header, Object headerValue, Class<T> type) {
- Object answer = requestBodyAndHeader(endpoint, body, header, headerValue);
+ Exchange exchange = producerCache.send(endpoint, ExchangePattern.InOut, createBodyAndHeaderProcessor(body, header, headerValue), createConvertBodyProcessor(type));
+ Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
public <T> T requestBodyAndHeader(String endpointUri, Object body, String header, Object headerValue, Class<T> type) {
- Object answer = requestBodyAndHeader(endpointUri, body, header, headerValue);
+ Exchange exchange = producerCache.send(resolveMandatoryEndpoint(endpointUri), ExchangePattern.InOut, createBodyAndHeaderProcessor(body, header, headerValue), createConvertBodyProcessor(type));
+ Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
public <T> T requestBodyAndHeaders(String endpointUri, Object body, Map<String, Object> headers, Class<T> type) {
- Object answer = requestBodyAndHeaders(endpointUri, body, headers);
+ Exchange exchange = producerCache.send(resolveMandatoryEndpoint(endpointUri), ExchangePattern.InOut, createBodyAndHeaders(body, headers), createConvertBodyProcessor(type));
+ Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
public <T> T requestBodyAndHeaders(Endpoint endpoint, Object body, Map<String, Object> headers, Class<T> type) {
- Object answer = requestBodyAndHeaders(endpoint, body, headers);
+ Exchange exchange = producerCache.send(endpoint, ExchangePattern.InOut, createBodyAndHeaders(body, headers), createConvertBodyProcessor(type));
+ Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
@@ -436,6 +444,20 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
};
}
+ protected Processor createBodyAndHeaders(final Object body, final Map<String, Object> headers) {
+ return new Processor() {
+ public void process(Exchange exchange) {
+ Message in = exchange.getIn();
+ if (headers != null) {
+ for (Map.Entry<String, Object> header : headers.entrySet()) {
+ in.setHeader(header.getKey(), header.getValue());
+ }
+ }
+ in.setBody(body);
+ }
+ };
+ }
+
protected Processor createBodyAndPropertyProcessor(final Object body, final String property, final Object propertyValue) {
return new Processor() {
public void process(Exchange exchange) {
@@ -455,6 +477,10 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
};
}
+ protected Processor createConvertBodyProcessor(final Class<?> type) {
+ return new ConvertBodyProcessor(type);
+ }
+
protected Endpoint resolveMandatoryEndpoint(String endpointUri) {
Endpoint endpoint = camelContext.getEndpoint(endpointUri);
if (endpoint == null) {
http://git-wip-us.apache.org/repos/asf/camel/blob/1cca6b7c/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java b/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java
index c016ea4..ebf641b 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.impl;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import org.apache.camel.AsyncCallback;
@@ -26,12 +28,12 @@ import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.FailedToCreateProducerException;
-import org.apache.camel.PollingConsumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.ProducerCallback;
import org.apache.camel.ServicePoolAware;
-import org.apache.camel.processor.UnitOfWorkProducer;
+import org.apache.camel.processor.CamelInternalProcessor;
+import org.apache.camel.processor.Pipeline;
import org.apache.camel.spi.EndpointUtilizationStatistics;
import org.apache.camel.spi.ServicePool;
import org.apache.camel.support.ServiceSupport;
@@ -203,7 +205,7 @@ public class ProducerCache extends ServiceSupport {
* @param exchange the exchange to send
*/
public void send(Endpoint endpoint, Exchange exchange) {
- sendExchange(endpoint, null, null, exchange);
+ sendExchange(endpoint, null, null, null, exchange);
}
/**
@@ -219,7 +221,7 @@ public class ProducerCache extends ServiceSupport {
* @return the exchange
*/
public Exchange send(Endpoint endpoint, Processor processor) {
- return sendExchange(endpoint, null, processor, null);
+ return sendExchange(endpoint, null, processor, null, null);
}
/**
@@ -236,7 +238,25 @@ public class ProducerCache extends ServiceSupport {
* @return the exchange
*/
public Exchange send(Endpoint endpoint, ExchangePattern pattern, Processor processor) {
- return sendExchange(endpoint, pattern, processor, null);
+ return sendExchange(endpoint, pattern, processor, null, null);
+ }
+
+ /**
+ * Sends an exchange to an endpoint using a supplied
+ * {@link Processor} to populate the exchange
+ * <p>
+ * This method will <b>not</b> throw an exception. If processing of the given
+ * Exchange failed then the exception is stored on the return Exchange
+ *
+ * @param endpoint the endpoint to send the exchange to
+ * @param pattern the message {@link ExchangePattern} such as
+ * {@link ExchangePattern#InOnly} or {@link ExchangePattern#InOut}
+ * @param processor the transformer used to populate the new exchange
+ * @param resultProcessor a processor to process the exchange when the send is complete.
+ * @return the exchange
+ */
+ public Exchange send(Endpoint endpoint, ExchangePattern pattern, Processor processor, Processor resultProcessor) {
+ return sendExchange(endpoint, pattern, processor, resultProcessor, null);
}
/**
@@ -377,7 +397,7 @@ public class ProducerCache extends ServiceSupport {
}
protected Exchange sendExchange(final Endpoint endpoint, ExchangePattern pattern,
- final Processor processor, Exchange exchange) {
+ final Processor processor, final Processor resultProcessor, Exchange exchange) {
return doInProducer(endpoint, exchange, pattern, new ProducerCallback<Exchange>() {
public Exchange doInProducer(Producer producer, Exchange exchange, ExchangePattern pattern) {
if (exchange == null) {
@@ -408,9 +428,23 @@ public class ProducerCache extends ServiceSupport {
watch = new StopWatch();
EventHelper.notifyExchangeSending(exchange.getContext(), exchange, endpoint);
}
- // ensure we run in an unit of work
- Producer target = new UnitOfWorkProducer(producer);
- target.process(exchange);
+
+ // if we have a result processor then wrap in pipeline to execute both of them in sequence
+ Processor target;
+ if (resultProcessor != null) {
+ List<Processor> processors = new ArrayList<Processor>(2);
+ processors.add(producer);
+ processors.add(resultProcessor);
+ target = Pipeline.newInstance(getCamelContext(), processors);
+ } else {
+ target = producer;
+ }
+
+ // wrap in unit of work
+ CamelInternalProcessor internal = new CamelInternalProcessor(target);
+ internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(null));
+
+ internal.process(exchange);
} catch (Throwable e) {
// ensure exceptions is caught and set on the exchange
exchange.setException(e);