camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [1/2] camel git commit: CAMEL-9873: Component should provide detail if a consumer/producer is native async supported
Date Mon, 18 Apr 2016 08:02:39 GMT
Repository: camel
Updated Branches:
  refs/heads/master 05b97d4bd -> fec035a51


CAMEL-9873: Component should provide detail if a consumer/producer is native async supported


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/fec035a5
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/fec035a5
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/fec035a5

Branch: refs/heads/master
Commit: fec035a515b6e71d1d059bf319ce6f9b181c6a86
Parents: e3f391b
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Mon Apr 18 09:59:30 2016 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Mon Apr 18 10:02:26 2016 +0200

----------------------------------------------------------------------
 .../component/scheduler/SchedulerEndpoint.java  |  3 +-
 .../camel/component/timer/TimerEndpoint.java    |  3 +-
 .../apache/camel/component/ahc/AhcEndpoint.java |  3 +-
 .../camel/component/amqp/AMQPEndpoint.java      |  3 +-
 .../camel/component/avro/AvroEndpoint.java      |  3 +-
 .../component/beanstalk/BeanstalkEndpoint.java  |  3 +-
 .../crypto/DigitalSignatureProducer.java        |  6 +--
 .../apache/camel/component/cxf/CxfEndpoint.java |  3 +-
 .../component/disruptor/DisruptorEndpoint.java  |  3 +-
 .../component/jetty8/JettyHttpEndpoint8.java    |  3 +-
 .../component/jetty9/JettyHttpEndpoint9.java    |  3 +-
 .../apache/camel/component/jms/JmsEndpoint.java |  3 +-
 .../camel/component/mqtt/MQTTEndpoint.java      |  3 +-
 .../component/netty/http/NettyHttpEndpoint.java |  3 +-
 .../camel/component/netty/NettyEndpoint.java    |  3 +-
 .../netty4/http/NettyHttpEndpoint.java          |  3 +-
 .../camel/component/netty4/NettyEndpoint.java   |  3 +-
 .../camel/component/paho/PahoProducer.java      |  2 -
 .../component/pgevent/PgEventProducer.java      | 40 ++++++++------------
 .../component/rabbitmq/RabbitMQEndpoint.java    |  3 +-
 .../component/restlet/RestletEndpoint.java      |  3 +-
 .../routebox/seda/RouteboxSedaEndpoint.java     |  3 +-
 .../camel/component/sjms/SjmsEndpoint.java      |  3 +-
 .../camel/component/stomp/StompEndpoint.java    |  3 +-
 .../component/undertow/UndertowEndpoint.java    |  3 +-
 .../camel/component/vertx/VertxEndpoint.java    |  3 +-
 26 files changed, 60 insertions(+), 57 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/camel-core/src/main/java/org/apache/camel/component/scheduler/SchedulerEndpoint.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/scheduler/SchedulerEndpoint.java
b/camel-core/src/main/java/org/apache/camel/component/scheduler/SchedulerEndpoint.java
index 1430030..8f03af8 100644
--- a/camel-core/src/main/java/org/apache/camel/component/scheduler/SchedulerEndpoint.java
+++ b/camel-core/src/main/java/org/apache/camel/component/scheduler/SchedulerEndpoint.java
@@ -18,7 +18,6 @@ package org.apache.camel.component.scheduler;
 
 import java.util.concurrent.ScheduledExecutorService;
 
-import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Processor;
 import org.apache.camel.Producer;
@@ -35,7 +34,7 @@ import org.apache.camel.spi.UriPath;
  * Also this component uses JDK ScheduledExecutorService. Where as the timer uses a JDK Timer.
  */
 @UriEndpoint(scheme = "scheduler", title = "Scheduler", syntax = "scheduler:name", consumerOnly
= true, consumerClass = SchedulerConsumer.class, label = "core,scheduling")
-public class SchedulerEndpoint extends ScheduledPollEndpoint implements AsyncEndpoint {
+public class SchedulerEndpoint extends ScheduledPollEndpoint {
 
     @UriPath @Metadata(required = "true")
     private String name;

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/camel-core/src/main/java/org/apache/camel/component/timer/TimerEndpoint.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/timer/TimerEndpoint.java
b/camel-core/src/main/java/org/apache/camel/component/timer/TimerEndpoint.java
index e3c705d..9473bc7 100644
--- a/camel-core/src/main/java/org/apache/camel/component/timer/TimerEndpoint.java
+++ b/camel-core/src/main/java/org/apache/camel/component/timer/TimerEndpoint.java
@@ -19,7 +19,6 @@ package org.apache.camel.component.timer;
 import java.util.Date;
 import java.util.Timer;
 
-import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Component;
 import org.apache.camel.Consumer;
 import org.apache.camel.MultipleConsumersSupport;
@@ -41,7 +40,7 @@ import org.apache.camel.spi.UriPath;
  */
 @ManagedResource(description = "Managed TimerEndpoint")
 @UriEndpoint(scheme = "timer", title = "Timer", syntax = "timer:timerName", consumerOnly
= true, consumerClass = TimerConsumer.class, label = "core,scheduling")
-public class TimerEndpoint extends DefaultEndpoint implements AsyncEndpoint, MultipleConsumersSupport
{
+public class TimerEndpoint extends DefaultEndpoint implements MultipleConsumersSupport {
     @UriPath @Metadata(required = "true")
     private String timerName;
     @UriParam(defaultValue = "1000")

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-ahc/src/main/java/org/apache/camel/component/ahc/AhcEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ahc/src/main/java/org/apache/camel/component/ahc/AhcEndpoint.java
b/components/camel-ahc/src/main/java/org/apache/camel/component/ahc/AhcEndpoint.java
index 9a2c098..89b2c8e 100644
--- a/components/camel-ahc/src/main/java/org/apache/camel/component/ahc/AhcEndpoint.java
+++ b/components/camel-ahc/src/main/java/org/apache/camel/component/ahc/AhcEndpoint.java
@@ -22,6 +22,7 @@ import javax.net.ssl.SSLContext;
 
 import com.ning.http.client.AsyncHttpClient;
 import com.ning.http.client.AsyncHttpClientConfig;
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Processor;
 import org.apache.camel.Producer;
@@ -39,7 +40,7 @@ import org.apache.camel.util.jsse.SSLContextParameters;
  * To call external HTTP services using <a href="http://github.com/sonatype/async-http-client">Async
Http Client</a>.
  */
 @UriEndpoint(scheme = "ahc", title = "AHC", syntax = "ahc:httpUri", producerOnly = true,
label = "http", lenientProperties = true)
-public class AhcEndpoint extends DefaultEndpoint implements HeaderFilterStrategyAware {
+public class AhcEndpoint extends DefaultEndpoint implements AsyncEndpoint, HeaderFilterStrategyAware
{
 
     private AsyncHttpClient client;
     @UriPath @Metadata(required = "true")

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPEndpoint.java
b/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPEndpoint.java
index ec6e95a..70d7a8a 100644
--- a/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPEndpoint.java
+++ b/components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPEndpoint.java
@@ -16,6 +16,7 @@
  */
 package org.apache.camel.component.amqp;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.component.jms.JmsConsumer;
 import org.apache.camel.component.jms.JmsEndpoint;
 import org.apache.camel.spi.UriEndpoint;
@@ -28,6 +29,6 @@ import org.apache.camel.spi.UriEndpoint;
  */
 @UriEndpoint(scheme = "amqp", extendsScheme = "jms", title = "AMQP",
         syntax = "amqp:destinationType:destinationName", consumerClass = JmsConsumer.class,
label = "messaging")
-public class AMQPEndpoint extends JmsEndpoint {
+public class AMQPEndpoint extends JmsEndpoint implements AsyncEndpoint {
 
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-avro/src/main/java/org/apache/camel/component/avro/AvroEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-avro/src/main/java/org/apache/camel/component/avro/AvroEndpoint.java
b/components/camel-avro/src/main/java/org/apache/camel/component/avro/AvroEndpoint.java
index 8ea5720..4117f8a 100644
--- a/components/camel-avro/src/main/java/org/apache/camel/component/avro/AvroEndpoint.java
+++ b/components/camel-avro/src/main/java/org/apache/camel/component/avro/AvroEndpoint.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.avro;
 import org.apache.avro.Protocol;
 import org.apache.avro.Schema;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Component;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
@@ -32,7 +33,7 @@ import org.apache.camel.spi.UriParam;
  * Working with Apache Avro for data serialization.
  */
 @UriEndpoint(scheme = "avro", title = "Avro", syntax = "avro:transport:host:port/messageName",
consumerClass = AvroConsumer.class, label = "messaging,transformation")
-public abstract class AvroEndpoint extends DefaultEndpoint {
+public abstract class AvroEndpoint extends DefaultEndpoint implements AsyncEndpoint {
 
     @UriParam
     private AvroConfiguration configuration;

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkEndpoint.java
b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkEndpoint.java
index 3fe4f77..4864157 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkEndpoint.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkEndpoint.java
@@ -17,6 +17,7 @@
 package org.apache.camel.component.beanstalk;
 
 import com.surftools.BeanstalkClient.Client;
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Component;
 import org.apache.camel.Consumer;
 import org.apache.camel.Processor;
@@ -37,7 +38,7 @@ import org.apache.camel.spi.UriPath;
  * The beanstalk component is used for job retrieval and post-processing of Beanstalk jobs.
  */
 @UriEndpoint(scheme = "beanstalk", title = "Beanstalk", syntax = "beanstalk:connectionSettings",
consumerClass = BeanstalkConsumer.class, label = "messaging")
-public class BeanstalkEndpoint extends ScheduledPollEndpoint {
+public class BeanstalkEndpoint extends ScheduledPollEndpoint implements AsyncEndpoint {
     final ConnectionSettings conn;
 
     @UriPath(description = "Connection settings host:port/tube")

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-crypto/src/main/java/org/apache/camel/component/crypto/DigitalSignatureProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-crypto/src/main/java/org/apache/camel/component/crypto/DigitalSignatureProducer.java
b/components/camel-crypto/src/main/java/org/apache/camel/component/crypto/DigitalSignatureProducer.java
index b39cf13..342fa35 100644
--- a/components/camel-crypto/src/main/java/org/apache/camel/component/crypto/DigitalSignatureProducer.java
+++ b/components/camel-crypto/src/main/java/org/apache/camel/component/crypto/DigitalSignatureProducer.java
@@ -32,10 +32,6 @@ public class DigitalSignatureProducer extends DefaultProducer {
     }
 
     public void process(Exchange exchange) throws Exception {
-        try {
-            processor.process(exchange);
-        } catch (Exception e) {
-            exchange.setException(e);
-        }
+        processor.process(exchange);
     }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfEndpoint.java
b/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfEndpoint.java
index 2794100..b9d6749 100644
--- a/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfEndpoint.java
+++ b/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfEndpoint.java
@@ -40,6 +40,7 @@ import javax.xml.ws.Provider;
 import javax.xml.ws.WebServiceProvider;
 import javax.xml.ws.handler.Handler;
 
+import org.apache.camel.AsyncEndpoint;
 import org.w3c.dom.Document;
 import org.w3c.dom.Element;
 import org.w3c.dom.Node;
@@ -113,7 +114,7 @@ import org.slf4j.LoggerFactory;
  * The cxf component is used for SOAP WebServices using Apache CXF.
  */
 @UriEndpoint(scheme = "cxf", title = "CXF", syntax = "cxf:beanId:address", consumerClass
= CxfConsumer.class, label = "soap,webservice")
-public class CxfEndpoint extends DefaultEndpoint implements HeaderFilterStrategyAware, Service,
Cloneable {
+public class CxfEndpoint extends DefaultEndpoint implements AsyncEndpoint, HeaderFilterStrategyAware,
Service, Cloneable {
 
     private static final Logger LOG = LoggerFactory.getLogger(CxfEndpoint.class);
 

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorEndpoint.java
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorEndpoint.java
index ec758a8..ce23073 100644
--- a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorEndpoint.java
+++ b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorEndpoint.java
@@ -24,6 +24,7 @@ import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 
 import com.lmax.disruptor.InsufficientCapacityException;
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Component;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
@@ -49,7 +50,7 @@ import org.slf4j.LoggerFactory;
  */
 @ManagedResource(description = "Managed Disruptor Endpoint")
 @UriEndpoint(scheme = "disruptor,disruptor-vm", title = "Disruptor,Disruptor VM", syntax
= "disruptor:name", consumerClass = DisruptorConsumer.class, label = "endpoint")
-public class DisruptorEndpoint extends DefaultEndpoint implements MultipleConsumersSupport
{
+public class DisruptorEndpoint extends DefaultEndpoint implements AsyncEndpoint, MultipleConsumersSupport
{
     public static final String DISRUPTOR_IGNORE_EXCHANGE = "disruptor.ignoreExchange";
     private static final Logger LOGGER = LoggerFactory.getLogger(DisruptorEndpoint.class);
 

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-jetty8/src/main/java/org/apache/camel/component/jetty8/JettyHttpEndpoint8.java
----------------------------------------------------------------------
diff --git a/components/camel-jetty8/src/main/java/org/apache/camel/component/jetty8/JettyHttpEndpoint8.java
b/components/camel-jetty8/src/main/java/org/apache/camel/component/jetty8/JettyHttpEndpoint8.java
index c6d766f..215e499 100644
--- a/components/camel-jetty8/src/main/java/org/apache/camel/component/jetty8/JettyHttpEndpoint8.java
+++ b/components/camel-jetty8/src/main/java/org/apache/camel/component/jetty8/JettyHttpEndpoint8.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.jetty8;
 import java.net.URI;
 import java.net.URISyntaxException;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.component.jetty.JettyContentExchange;
 import org.apache.camel.component.jetty.JettyHttpComponent;
 import org.apache.camel.component.jetty.JettyHttpEndpoint;
@@ -30,7 +31,7 @@ import org.apache.camel.spi.UriEndpoint;
  */
 @UriEndpoint(scheme = "jetty", extendsScheme = "http", title = "Jetty",
         syntax = "jetty:httpUri", consumerClass = HttpConsumer.class, label = "http", lenientProperties
= true)
-public class JettyHttpEndpoint8 extends JettyHttpEndpoint {
+public class JettyHttpEndpoint8 extends JettyHttpEndpoint implements AsyncEndpoint {
 
     public JettyHttpEndpoint8(JettyHttpComponent component, String uri, URI httpURL) throws
URISyntaxException {
         super(component, uri, httpURL);

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-jetty9/src/main/java/org/apache/camel/component/jetty9/JettyHttpEndpoint9.java
----------------------------------------------------------------------
diff --git a/components/camel-jetty9/src/main/java/org/apache/camel/component/jetty9/JettyHttpEndpoint9.java
b/components/camel-jetty9/src/main/java/org/apache/camel/component/jetty9/JettyHttpEndpoint9.java
index add21e4..7f00a58 100644
--- a/components/camel-jetty9/src/main/java/org/apache/camel/component/jetty9/JettyHttpEndpoint9.java
+++ b/components/camel-jetty9/src/main/java/org/apache/camel/component/jetty9/JettyHttpEndpoint9.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.jetty9;
 import java.net.URI;
 import java.net.URISyntaxException;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.component.jetty.JettyContentExchange;
 import org.apache.camel.component.jetty.JettyHttpComponent;
 import org.apache.camel.component.jetty.JettyHttpEndpoint;
@@ -31,7 +32,7 @@ import org.apache.camel.spi.UriEndpoint;
  */
 @UriEndpoint(scheme = "jetty", extendsScheme = "http", title = "Jetty 9",
         syntax = "jetty:httpUri", consumerClass = HttpConsumer.class, label = "http", lenientProperties
= true)
-public class JettyHttpEndpoint9 extends JettyHttpEndpoint {
+public class JettyHttpEndpoint9 extends JettyHttpEndpoint implements AsyncEndpoint {
     private HttpBinding binding;
 
     public JettyHttpEndpoint9(JettyHttpComponent component, String uri, URI httpURL) throws
URISyntaxException {

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
index b1c0038..c671e75 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
@@ -30,6 +30,7 @@ import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.Topic;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePattern;
 import org.apache.camel.LoggingLevel;
@@ -69,7 +70,7 @@ import org.springframework.util.ErrorHandler;
  */
 @ManagedResource(description = "Managed JMS Endpoint")
 @UriEndpoint(scheme = "jms", title = "JMS", syntax = "jms:destinationType:destinationName",
consumerClass = JmsConsumer.class, label = "messaging")
-public class JmsEndpoint extends DefaultEndpoint implements HeaderFilterStrategyAware, MultipleConsumersSupport,
Service {
+public class JmsEndpoint extends DefaultEndpoint implements AsyncEndpoint, HeaderFilterStrategyAware,
MultipleConsumersSupport, Service {
     protected final Logger log = LoggerFactory.getLogger(getClass());
     private final AtomicInteger runningMessageListeners = new AtomicInteger();
     private boolean pubSubDomain;

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
b/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
index d079778..71e333e 100644
--- a/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
+++ b/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
@@ -21,6 +21,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
@@ -61,7 +62,7 @@ import org.slf4j.LoggerFactory;
  * Component for communicating with MQTT M2M message brokers using FuseSource MQTT Client.
  */
 @UriEndpoint(scheme = "mqtt", title = "MQTT", syntax = "mqtt:name", consumerClass = MQTTConsumer.class,
label = "messaging,iot")
-public class MQTTEndpoint extends DefaultEndpoint {
+public class MQTTEndpoint extends DefaultEndpoint implements AsyncEndpoint {
     private static final Logger LOG = LoggerFactory.getLogger(MQTTEndpoint.class);
 
     private static final int PUBLISH_MAX_RECONNECT_ATTEMPTS = 3;

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java
b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java
index 93fba20..545591b 100644
--- a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java
+++ b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.netty.http;
 
 import java.util.Map;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.Message;
@@ -45,7 +46,7 @@ import org.slf4j.LoggerFactory;
         syntax = "netty-http:protocol:host:port/path", consumerClass = NettyHttpConsumer.class,
label = "http", lenientProperties = true,
         excludeProperties = "textline,delimiter,autoAppendDelimiter,decoderMaxLineLength,encoding,allowDefaultCodec,udpConnectionlessSending,networkInterface"
                 + ",clientMode,reconnect,reconnectInterval,broadcast")
-public class NettyHttpEndpoint extends NettyEndpoint implements HeaderFilterStrategyAware
{
+public class NettyHttpEndpoint extends NettyEndpoint implements AsyncEndpoint, HeaderFilterStrategyAware
{
 
     private static final Logger LOG = LoggerFactory.getLogger(NettyHttpEndpoint.class);
     @UriParam

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyEndpoint.java
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyEndpoint.java
index b86bfbc..6d5fa6a 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyEndpoint.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyEndpoint.java
@@ -22,6 +22,7 @@ import javax.net.ssl.SSLPeerUnverifiedException;
 import javax.net.ssl.SSLSession;
 import javax.security.cert.X509Certificate;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.Message;
@@ -41,7 +42,7 @@ import org.jboss.netty.util.Timer;
  * Socket level networking using TCP or UDP with the Netty 3.x library.
  */
 @UriEndpoint(scheme = "netty", title = "Netty", syntax = "netty:protocol:host:port", consumerClass
= NettyConsumer.class, label = "networking,tcp,udp")
-public class NettyEndpoint extends DefaultEndpoint {
+public class NettyEndpoint extends DefaultEndpoint implements AsyncEndpoint {
     @UriParam
     private NettyConfiguration configuration;
     @UriParam(label = "advanced", javaType = "org.apache.camel.component.netty.NettyServerBootstrapConfiguration",

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/NettyHttpEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/NettyHttpEndpoint.java
b/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/NettyHttpEndpoint.java
index 894a606..43adfcf 100644
--- a/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/NettyHttpEndpoint.java
+++ b/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/NettyHttpEndpoint.java
@@ -20,6 +20,7 @@ import java.util.Map;
 
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.FullHttpRequest;
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.Message;
@@ -44,7 +45,7 @@ import org.slf4j.LoggerFactory;
         syntax = "netty4-http:protocol:host:port/path", consumerClass = NettyHttpConsumer.class,
label = "http", lenientProperties = true,
         excludeProperties = "textline,delimiter,autoAppendDelimiter,decoderMaxLineLength,encoding,allowDefaultCodec,udpConnectionlessSending,networkInterface"
                 + ",clientMode,reconnect,reconnectInterval,useByteBuf,udpByteArrayCodec,broadcast")
-public class NettyHttpEndpoint extends NettyEndpoint implements HeaderFilterStrategyAware
{
+public class NettyHttpEndpoint extends NettyEndpoint implements AsyncEndpoint, HeaderFilterStrategyAware
{
 
     private static final Logger LOG = LoggerFactory.getLogger(NettyHttpEndpoint.class);
     @UriParam

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyEndpoint.java
b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyEndpoint.java
index 04dbc11..b16e45c 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyEndpoint.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyEndpoint.java
@@ -26,6 +26,7 @@ import javax.security.cert.X509Certificate;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.ssl.SslHandler;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.Message;
@@ -41,7 +42,7 @@ import org.apache.camel.util.ObjectHelper;
  * Socket level networking using TCP or UDP with the Netty 4.x library.
  */
 @UriEndpoint(scheme = "netty4", title = "Netty4", syntax = "netty4:protocol:host:port", consumerClass
= NettyConsumer.class, label = "networking,tcp,udp")
-public class NettyEndpoint extends DefaultEndpoint {
+public class NettyEndpoint extends DefaultEndpoint implements AsyncEndpoint {
     @UriParam
     private NettyConfiguration configuration;
     @UriParam(label = "advanced", javaType = "org.apache.camel.component.netty4.NettyServerBootstrapConfiguration",

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
index cdd360a..777f52d 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
@@ -43,8 +43,6 @@ public class PahoProducer extends DefaultProducer {
         client.publish(topic, message);
     }
 
-   
-
     @Override
     public PahoEndpoint getEndpoint() {
         return (PahoEndpoint)super.getEndpoint();

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
index a586a22..6b21304 100644
--- a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
+++ b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
@@ -18,18 +18,16 @@ package org.apache.camel.component.pgevent;
 
 import java.sql.CallableStatement;
 import java.sql.PreparedStatement;
-import java.sql.SQLException;
 
 import com.impossibl.postgres.api.jdbc.PGConnection;
 
-import org.apache.camel.AsyncCallback;
 import org.apache.camel.Exchange;
-import org.apache.camel.impl.DefaultAsyncProducer;
+import org.apache.camel.impl.DefaultProducer;
 
 /**
  * The PgEvent producer.
  */
-public class PgEventProducer extends DefaultAsyncProducer {
+public class PgEventProducer extends DefaultProducer {
     private final PgEventEndpoint endpoint;
     private PGConnection dbConnection;
 
@@ -39,36 +37,28 @@ public class PgEventProducer extends DefaultAsyncProducer {
     }
 
     @Override
-    public boolean process(final Exchange exchange, final AsyncCallback callback) {
+    public void process(Exchange exchange) throws Exception {
         try {
             if (dbConnection.isClosed()) {
                 dbConnection = endpoint.initJdbc();
             }
         } catch (Exception e) {
-            exchange.setException(new InvalidStateException("Database connection closed and
could not be re-opened.", e));
-            callback.done(true);
-            return true;
+            throw new InvalidStateException("Database connection closed and could not be
re-opened.", e);
         }
 
-        try {
-            String payload = exchange.getIn().getBody(String.class);
-            if (dbConnection.isServerMinimumVersion(9, 0)) {
-                try (CallableStatement statement = dbConnection.prepareCall("{call pg_notify(?,
?)}")) {
-                    statement.setString(1, endpoint.getChannel());
-                    statement.setString(2, payload);
-                    statement.execute();
-                }
-            } else {
-                String sql = String.format("NOTIFY %s, '%s'", endpoint.getChannel(), payload);
-                try (PreparedStatement statement = dbConnection.prepareStatement(sql)) {
-                    statement.execute();
-                }
+        String payload = exchange.getIn().getBody(String.class);
+        if (dbConnection.isServerMinimumVersion(9, 0)) {
+            try (CallableStatement statement = dbConnection.prepareCall("{call pg_notify(?,
?)}")) {
+                statement.setString(1, endpoint.getChannel());
+                statement.setString(2, payload);
+                statement.execute();
+            }
+        } else {
+            String sql = String.format("NOTIFY %s, '%s'", endpoint.getChannel(), payload);
+            try (PreparedStatement statement = dbConnection.prepareStatement(sql)) {
+                statement.execute();
             }
-        } catch (SQLException e) {
-            exchange.setException(e);
         }
-        callback.done(true);
-        return true;
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
index 71721cf..145bd49 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
@@ -33,6 +33,7 @@ import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.Envelope;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.Message;
@@ -45,7 +46,7 @@ import org.apache.camel.spi.UriParam;
 import org.apache.camel.spi.UriPath;
 
 @UriEndpoint(scheme = "rabbitmq", title = "RabbitMQ", syntax = "rabbitmq:hostname:portNumber/exchangeName",
consumerClass = RabbitMQConsumer.class, label = "messaging")
-public class RabbitMQEndpoint extends DefaultEndpoint {
+public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
     // header to indicate that the message body needs to be de-serialized
     public static final String SERIALIZE_HEADER = "CamelSerialize";
 

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-restlet/src/main/java/org/apache/camel/component/restlet/RestletEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-restlet/src/main/java/org/apache/camel/component/restlet/RestletEndpoint.java
b/components/camel-restlet/src/main/java/org/apache/camel/component/restlet/RestletEndpoint.java
index 5fb8e43..63ae6fa 100644
--- a/components/camel-restlet/src/main/java/org/apache/camel/component/restlet/RestletEndpoint.java
+++ b/components/camel-restlet/src/main/java/org/apache/camel/component/restlet/RestletEndpoint.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.restlet;
 import java.util.List;
 import java.util.Map;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePattern;
@@ -40,7 +41,7 @@ import org.restlet.data.Method;
  */
 @UriEndpoint(scheme = "restlet", title = "Restlet", syntax = "restlet:protocol:host:port/uriPattern",
         consumerClass = RestletConsumer.class, label = "rest", lenientProperties = true)
-public class RestletEndpoint extends DefaultEndpoint implements HeaderFilterStrategyAware
{
+public class RestletEndpoint extends DefaultEndpoint implements AsyncEndpoint, HeaderFilterStrategyAware
{
     private static final int DEFAULT_PORT = 80;
     private static final String DEFAULT_PROTOCOL = "http";
     private static final String DEFAULT_HOST = "localhost";

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-routebox/src/main/java/org/apache/camel/component/routebox/seda/RouteboxSedaEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-routebox/src/main/java/org/apache/camel/component/routebox/seda/RouteboxSedaEndpoint.java
b/components/camel-routebox/src/main/java/org/apache/camel/component/routebox/seda/RouteboxSedaEndpoint.java
index bd99705..0bebcd2 100644
--- a/components/camel-routebox/src/main/java/org/apache/camel/component/routebox/seda/RouteboxSedaEndpoint.java
+++ b/components/camel-routebox/src/main/java/org/apache/camel/component/routebox/seda/RouteboxSedaEndpoint.java
@@ -24,6 +24,7 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.LinkedBlockingQueue;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.MultipleConsumersSupport;
@@ -37,7 +38,7 @@ import org.apache.camel.component.routebox.RouteboxEndpoint;
 import org.apache.camel.component.routebox.RouteboxProducer;
 import org.apache.camel.spi.BrowsableEndpoint;
 
-public class RouteboxSedaEndpoint extends RouteboxEndpoint implements BrowsableEndpoint,
MultipleConsumersSupport {
+public class RouteboxSedaEndpoint extends RouteboxEndpoint implements AsyncEndpoint, BrowsableEndpoint,
MultipleConsumersSupport {
     private WaitForTaskToComplete waitForTaskToComplete = WaitForTaskToComplete.IfReplyExpected;
     private volatile BlockingQueue<Exchange> queue;
     private volatile Set<RouteboxProducer> producers = new CopyOnWriteArraySet<RouteboxProducer>();

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
index 6a68a34..272e816 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.sjms;
 import javax.jms.Message;
 import javax.jms.Session;
 
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Component;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
@@ -53,7 +54,7 @@ import org.slf4j.LoggerFactory;
  * This component uses plain JMS API where as the jms component uses Spring JMS.
  */
 @UriEndpoint(scheme = "sjms", title = "Simple JMS", syntax = "sjms:destinationType:destinationName",
consumerClass = SjmsConsumer.class, label = "messaging")
-public class SjmsEndpoint extends DefaultEndpoint implements MultipleConsumersSupport, HeaderFilterStrategyAware
{
+public class SjmsEndpoint extends DefaultEndpoint implements AsyncEndpoint, MultipleConsumersSupport,
HeaderFilterStrategyAware {
     protected final Logger logger = LoggerFactory.getLogger(getClass());
 
     private boolean topic;

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-stomp/src/main/java/org/apache/camel/component/stomp/StompEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-stomp/src/main/java/org/apache/camel/component/stomp/StompEndpoint.java
b/components/camel-stomp/src/main/java/org/apache/camel/component/stomp/StompEndpoint.java
index f8b73b2..5e600f9 100644
--- a/components/camel-stomp/src/main/java/org/apache/camel/component/stomp/StompEndpoint.java
+++ b/components/camel-stomp/src/main/java/org/apache/camel/component/stomp/StompEndpoint.java
@@ -20,6 +20,7 @@ import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 
 import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
@@ -49,7 +50,7 @@ import static org.fusesource.stomp.client.Constants.UNSUBSCRIBE;
  * The stomp component is used for communicating with Stomp compliant message brokers.
  */
 @UriEndpoint(scheme = "stomp", title = "Stomp", syntax = "stomp:destination", consumerClass
= StompConsumer.class, label = "messaging")
-public class StompEndpoint extends DefaultEndpoint {
+public class StompEndpoint extends DefaultEndpoint implements AsyncEndpoint {
 
     @UriPath(description = "Name of the queue") @Metadata(required = "true")
     private String destination;

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/UndertowEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/UndertowEndpoint.java
b/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/UndertowEndpoint.java
index 8a19679..1123c29 100644
--- a/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/UndertowEndpoint.java
+++ b/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/UndertowEndpoint.java
@@ -23,6 +23,7 @@ import java.util.Map;
 import javax.net.ssl.SSLContext;
 
 import io.undertow.server.HttpServerExchange;
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.Message;
@@ -48,7 +49,7 @@ import org.xnio.Options;
  */
 @UriEndpoint(scheme = "undertow", title = "Undertow", syntax = "undertow:httpURI",
         consumerClass = UndertowConsumer.class, label = "http", lenientProperties = true)
-public class UndertowEndpoint extends DefaultEndpoint implements HeaderFilterStrategyAware
{
+public class UndertowEndpoint extends DefaultEndpoint implements AsyncEndpoint, HeaderFilterStrategyAware
{
 
     private static final Logger LOG = LoggerFactory.getLogger(UndertowEndpoint.class);
     private UndertowComponent component;

http://git-wip-us.apache.org/repos/asf/camel/blob/fec035a5/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxEndpoint.java
b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxEndpoint.java
index 7fbdccd..aeca5e6 100644
--- a/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxEndpoint.java
+++ b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxEndpoint.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.vertx;
 
 import io.vertx.core.Vertx;
 import io.vertx.core.eventbus.EventBus;
+import org.apache.camel.AsyncEndpoint;
 import org.apache.camel.Consumer;
 import org.apache.camel.MultipleConsumersSupport;
 import org.apache.camel.Processor;
@@ -32,7 +33,7 @@ import org.apache.camel.spi.UriPath;
  * The vertx component is used for sending and receive messages from a vertx event bus.
  */
 @UriEndpoint(scheme = "vertx", title = "Vert.x", syntax = "vertx:address", consumerClass
= VertxConsumer.class, label = "eventbus")
-public class VertxEndpoint extends DefaultEndpoint implements MultipleConsumersSupport {
+public class VertxEndpoint extends DefaultEndpoint implements AsyncEndpoint, MultipleConsumersSupport
{
 
     @UriPath @Metadata(required = "true")
     private String address;


Mime
View raw message