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-10308 Provide a way to use async engine from ProducerTemplate style fixes, optiona rename & don't include out of camel callbacks into processing time
Date Sat, 17 Sep 2016 10:45:19 GMT
Repository: camel
Updated Branches:
  refs/heads/master 5104c1669 -> 1c8eda402


CAMEL-10308
Provide a way to use async engine from ProducerTemplate
style fixes, optiona rename & don't include out of camel callbacks into processing time


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

Branch: refs/heads/master
Commit: 1c8eda4020531e1e2ecf560fb1b91f1ddd435dd2
Parents: 51a8068
Author: Vitalii Tymchyshyn <vit@tym.im>
Authored: Fri Sep 16 20:39:27 2016 -0400
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Sat Sep 17 09:34:52 2016 +0200

----------------------------------------------------------------------
 .../java/org/apache/camel/ProducerTemplate.java |  23 ++--
 ...AsyncCallbackToCompletableFutureAdapter.java |   8 +-
 .../camel/impl/DefaultProducerTemplate.java     |  12 +-
 .../camel/impl/EventNotifierCallback.java       |  11 +-
 .../org/apache/camel/impl/ProducerCache.java    | 115 +++++++++----------
 ...ultProducerTemplateNonBlockingAsyncTest.java |  10 +-
 6 files changed, 85 insertions(+), 94 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/1c8eda40/camel-core/src/main/java/org/apache/camel/ProducerTemplate.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/ProducerTemplate.java b/camel-core/src/main/java/org/apache/camel/ProducerTemplate.java
index 6461e4e..8858e74 100644
--- a/camel-core/src/main/java/org/apache/camel/ProducerTemplate.java
+++ b/camel-core/src/main/java/org/apache/camel/ProducerTemplate.java
@@ -16,8 +16,6 @@
  */
 package org.apache.camel;
 
-import org.apache.camel.spi.Synchronization;
-
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
@@ -25,6 +23,8 @@ import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import org.apache.camel.spi.Synchronization;
+
 /**
  * Template for working with Camel and sending {@link Message} instances in an
  * {@link Exchange} to an {@link Endpoint}.
@@ -94,19 +94,20 @@ public interface ProducerTemplate extends Service {
     int getCurrentCacheSize();
 
     /**
-     * Reports if async* methods will dispath processing from the calling thread (true) or
through executor (false).
-     * They will still employ asynchronous engine, so this mode can be useful for high-speed
non-blocking processing.
-     * @return if async* methods will run in the calling thread
+     * Reports if async* methods will dispath processing from the calling thread (false)
or through executor (true).
+     * In both cases asynchronous engine will be used, so this non-threaded can be useful
for high-speed
+     * non-blocking processing.
+     * @return if async* methods will dipatch processing with the executor
      */
-    boolean isSynchronous();
+    boolean isThreadedAsyncMode();
 
     /**
-     * Reports if async* methods will dispath processing from the calling thread (true) or
through executor (false).
-     * In any case they would still employ asynchronous engine, so setting to true can be
useful
-     * for high-speed non-blocking processing.
-     * @param synchronous if async* methods will run in the calling thread
+     * Reports if async* methods will dispath processing from the calling thread (false)
or through executor (true).
+     * In both cases asynchronous engine will be used, so this non-threaded can be useful
for high-speed
+     * non-blocking processing.
+     * @param useExecutor if async* methods will dipatch processing with the executor
      */
-    void setSynchronous(boolean synchronous);
+    void setThreadedAsyncMode(boolean useExecutor);
     
     /**
      * Get the default endpoint to use if none is specified

http://git-wip-us.apache.org/repos/asf/camel/blob/1c8eda40/camel-core/src/main/java/org/apache/camel/impl/AsyncCallbackToCompletableFutureAdapter.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/AsyncCallbackToCompletableFutureAdapter.java
b/camel-core/src/main/java/org/apache/camel/impl/AsyncCallbackToCompletableFutureAdapter.java
index 0baa510..0576a10 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/AsyncCallbackToCompletableFutureAdapter.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/AsyncCallbackToCompletableFutureAdapter.java
@@ -16,14 +16,14 @@
  */
 package org.apache.camel.impl;
 
-import org.apache.camel.AsyncCallback;
-
 import java.util.concurrent.CompletableFuture;
 
+import org.apache.camel.AsyncCallback;
+
 /**
- * AsyncCallback that provides a CompletableFuture when async action is done
+ * AsyncCallback that provides a CompletableFuture completed when async action is done
  */
-public class AsyncCallbackToCompletableFutureAdapter<T> implements AsyncCallback{
+public class AsyncCallbackToCompletableFutureAdapter<T> implements AsyncCallback {
     private final CompletableFuture<T> future;
     private volatile T result;
 

http://git-wip-us.apache.org/repos/asf/camel/blob/1c8eda40/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 19958e4..a3dfd51 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
@@ -56,7 +56,7 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
     private Endpoint defaultEndpoint;
     private int maximumCacheSize;
     private boolean eventNotifierEnabled = true;
-    private volatile boolean synchronous;
+    private volatile boolean threadedAsyncMode = true;
 
     public DefaultProducerTemplate(CamelContext camelContext) {
         this.camelContext = camelContext;
@@ -86,13 +86,13 @@ public class DefaultProducerTemplate extends ServiceSupport implements
ProducerT
     }
 
     @Override
-    public boolean isSynchronous() {
-        return synchronous;
+    public boolean isThreadedAsyncMode() {
+        return threadedAsyncMode;
     }
 
     @Override
-    public void setSynchronous(boolean synchronous) {
-        this.synchronous = synchronous;
+    public void setThreadedAsyncMode(boolean useExecutor) {
+        this.threadedAsyncMode = useExecutor;
     }
 
     public int getCurrentCacheSize() {
@@ -705,7 +705,7 @@ public class DefaultProducerTemplate extends ServiceSupport implements
ProducerT
             if (executor != null) {
                 return executor;
             }
-            if (synchronous) {
+            if (!threadedAsyncMode) {
                 executor = new SynchronousExecutorService();
             } else {
                 executor = camelContext.getExecutorServiceManager().newDefaultThreadPool(this,
"ProducerTemplate");

http://git-wip-us.apache.org/repos/asf/camel/blob/1c8eda40/camel-core/src/main/java/org/apache/camel/impl/EventNotifierCallback.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/EventNotifierCallback.java b/camel-core/src/main/java/org/apache/camel/impl/EventNotifierCallback.java
index 734551f..e186d9a 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/EventNotifierCallback.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/EventNotifierCallback.java
@@ -25,7 +25,7 @@ import org.apache.camel.util.StopWatch;
 /**
  * Helper class to notify on exchange sending events in async engine
  */
-class EventNotifierCallback implements AsyncCallback {
+public class EventNotifierCallback implements AsyncCallback {
     private final AsyncCallback originalCallback;
     private final StopWatch watch;
     private final Exchange exchange;
@@ -42,11 +42,8 @@ class EventNotifierCallback implements AsyncCallback {
 
     @Override
     public void done(boolean doneSync) {
-        try {
-            originalCallback.done(doneSync);
-        } finally {
-            long timeTaken = watch.stop();
-            EventHelper.notifyExchangeSent(exchange.getContext(), exchange, endpoint, timeTaken);
-        }
+        long timeTaken = watch.stop();
+        EventHelper.notifyExchangeSent(exchange.getContext(), exchange, endpoint, timeTaken);
+        originalCallback.done(doneSync);
     }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/1c8eda40/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 b475d29..1062029 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
@@ -50,7 +50,7 @@ import org.slf4j.LoggerFactory;
 /**
  * Cache containing created {@link Producer}.
  *
- * @version 
+ * @version
  */
 public class ProducerCache extends ServiceSupport {
     private static final Logger LOG = LoggerFactory.getLogger(ProducerCache.class);
@@ -277,7 +277,7 @@ public class ProducerCache extends ServiceSupport {
      *              or new one if parameter was null
      */
     public CompletableFuture<Exchange> asyncSend(Endpoint endpoint, ExchangePattern
pattern, Processor processor, Processor resultProcessor,
-            CompletableFuture<Exchange> future) {
+                                                 CompletableFuture<Exchange> future)
{
         return asyncSendExchange(endpoint, pattern, processor, resultProcessor, null, future);
     }
 
@@ -299,41 +299,37 @@ public class ProducerCache extends ServiceSupport {
      *              or new one if parameter was null
      */
     public CompletableFuture<Exchange> asyncSendExchange(final Endpoint endpoint, ExchangePattern
pattern,
-            final Processor processor, final Processor resultProcessor, Exchange exchange,
-            CompletableFuture<Exchange> future) {
+                                                         final Processor processor, final
Processor resultProcessor, Exchange exchange,
+                                                         CompletableFuture<Exchange>
future) {
         AsyncCallbackToCompletableFutureAdapter<Exchange> futureAdapter = new AsyncCallbackToCompletableFutureAdapter<>(future,
exchange);
         doInAsyncProducer(endpoint, exchange, pattern, futureAdapter,
-                (producer, asyncProducer, innerExchange, exchangePattern, producerCallback)
-> {
-                    if (innerExchange == null) {
-                        innerExchange = pattern != null ?
-                                producer.getEndpoint().createExchange(pattern) :
-                                producer.getEndpoint().createExchange();
-                        futureAdapter.setResult(innerExchange);
-                    }
+            (producer, asyncProducer, innerExchange, exchangePattern, producerCallback) ->
{
+                if (innerExchange == null) {
+                    innerExchange = pattern != null
+                            ? producer.getEndpoint().createExchange(pattern)
+                            : producer.getEndpoint().createExchange();
+                    futureAdapter.setResult(innerExchange);
+                }
 
-                    if (processor != null) {
-                        // lets populate using the processor callback
-                        AsyncProcessor asyncProcessor = AsyncProcessorConverterHelper.convert(processor);
-                        try {
-                            final Exchange finalExchange = innerExchange;
-                            asyncProcessor.process(innerExchange, new AsyncCallback() {
-                                @Override
-                                public void done(boolean doneSync) {
-                                    asyncDispatchExchange(endpoint, producer, resultProcessor,
finalExchange,
-                                            producerCallback);
-                                }
-                            });
-                            return false;
-                        } catch (Exception e) {
-                            // populate failed so return
-                            innerExchange.setException(e);
-                            producerCallback.done(true);
-                            return true;
-                        }
+                if (processor != null) {
+                    // lets populate using the processor callback
+                    AsyncProcessor asyncProcessor = AsyncProcessorConverterHelper.convert(processor);
+                    try {
+                        final Exchange finalExchange = innerExchange;
+                        asyncProcessor.process(innerExchange,
+                            doneSync -> asyncDispatchExchange(endpoint, producer, resultProcessor,
+                                    finalExchange, producerCallback));
+                        return false;
+                    } catch (Exception e) {
+                        // populate failed so return
+                        innerExchange.setException(e);
+                        producerCallback.done(true);
+                        return true;
                     }
+                }
 
-                    return asyncDispatchExchange(endpoint, producer, resultProcessor, innerExchange,
producerCallback);
-                });
+                return asyncDispatchExchange(endpoint, producer, resultProcessor, innerExchange,
producerCallback);
+            });
         return futureAdapter.getFuture();
     }
 
@@ -437,31 +433,28 @@ public class ProducerCache extends ServiceSupport {
             }
             // invoke the callback
             AsyncProcessor asyncProcessor = AsyncProcessorConverterHelper.convert(producer);
-            return producerCallback.doInAsyncProducer(producer, asyncProcessor, exchange,
pattern, new AsyncCallback() {
-                @Override
-                public void done(boolean doneSync) {
-                    try {
-                        if (eventNotifierEnabled && watch != null) {
-                            long timeTaken = watch.stop();
-                            // emit event that the exchange was sent to the endpoint
-                            EventHelper.notifyExchangeSent(exchange.getContext(), exchange,
endpoint, timeTaken);
-                        }
+            return producerCallback.doInAsyncProducer(producer, asyncProcessor, exchange,
pattern, doneSync -> {
+                try {
+                    if (eventNotifierEnabled && watch != null) {
+                        long timeTaken = watch.stop();
+                        // emit event that the exchange was sent to the endpoint
+                        EventHelper.notifyExchangeSent(exchange.getContext(), exchange, endpoint,
timeTaken);
+                    }
 
-                        if (producer instanceof ServicePoolAware) {
-                            // release back to the pool
-                            pool.release(endpoint, producer);
-                        } else if (!producer.isSingleton()) {
-                            // stop and shutdown non-singleton producers as we should not
leak resources
-                            try {
-                                ServiceHelper.stopAndShutdownService(producer);
-                            } catch (Exception e) {
-                                // ignore and continue
-                                LOG.warn("Error stopping/shutting down producer: " + producer,
e);
-                            }
+                    if (producer instanceof ServicePoolAware) {
+                        // release back to the pool
+                        pool.release(endpoint, producer);
+                    } else if (!producer.isSingleton()) {
+                        // stop and shutdown non-singleton producers as we should not leak
resources
+                        try {
+                            ServiceHelper.stopAndShutdownService(producer);
+                        } catch (Exception e) {
+                            // ignore and continue
+                            LOG.warn("Error stopping/shutting down producer: " + producer,
e);
                         }
-                    } finally {
-                        callback.done(doneSync);
                     }
+                } finally {
+                    callback.done(doneSync);
                 }
             });
         } catch (Throwable e) {
@@ -475,7 +468,7 @@ public class ProducerCache extends ServiceSupport {
     }
 
     protected boolean asyncDispatchExchange(final Endpoint endpoint, Producer producer,
-            final Processor resultProcessor, Exchange exchange, AsyncCallback callback) {
+                                            final Processor resultProcessor, Exchange exchange,
AsyncCallback callback) {
         // now lets dispatch
         LOG.debug(">>>> {} {}", endpoint, exchange);
 
@@ -659,7 +652,7 @@ public class ProducerCache extends ServiceSupport {
     public int getCapacity() {
         int capacity = -1;
         if (producers instanceof LRUCache) {
-            LRUCache<String, Producer> cache = (LRUCache<String, Producer>)producers;
+            LRUCache<String, Producer> cache = (LRUCache<String, Producer>) producers;
             capacity = cache.getMaxCacheSize();
         }
         return capacity;
@@ -675,7 +668,7 @@ public class ProducerCache extends ServiceSupport {
     public long getHits() {
         long hits = -1;
         if (producers instanceof LRUCache) {
-            LRUCache<String, Producer> cache = (LRUCache<String, Producer>)producers;
+            LRUCache<String, Producer> cache = (LRUCache<String, Producer>) producers;
             hits = cache.getHits();
         }
         return hits;
@@ -691,7 +684,7 @@ public class ProducerCache extends ServiceSupport {
     public long getMisses() {
         long misses = -1;
         if (producers instanceof LRUCache) {
-            LRUCache<String, Producer> cache = (LRUCache<String, Producer>)producers;
+            LRUCache<String, Producer> cache = (LRUCache<String, Producer>) producers;
             misses = cache.getMisses();
         }
         return misses;
@@ -707,7 +700,7 @@ public class ProducerCache extends ServiceSupport {
     public long getEvicted() {
         long evicted = -1;
         if (producers instanceof LRUCache) {
-            LRUCache<String, Producer> cache = (LRUCache<String, Producer>)producers;
+            LRUCache<String, Producer> cache = (LRUCache<String, Producer>) producers;
             evicted = cache.getEvicted();
         }
         return evicted;
@@ -718,7 +711,7 @@ public class ProducerCache extends ServiceSupport {
      */
     public void resetCacheStatistics() {
         if (producers instanceof LRUCache) {
-            LRUCache<String, Producer> cache = (LRUCache<String, Producer>)producers;
+            LRUCache<String, Producer> cache = (LRUCache<String, Producer>) producers;
             cache.resetStatistics();
         }
         if (statistics != null) {
@@ -742,7 +735,7 @@ public class ProducerCache extends ServiceSupport {
      */
     public void cleanUp() {
         if (producers instanceof LRUCache) {
-            LRUCache<String, Producer> cache = (LRUCache<String, Producer>)producers;
+            LRUCache<String, Producer> cache = (LRUCache<String, Producer>) producers;
             cache.cleanUp();
         }
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/1c8eda40/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerTemplateNonBlockingAsyncTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerTemplateNonBlockingAsyncTest.java
b/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerTemplateNonBlockingAsyncTest.java
index f218a6e..17d32a1 100644
--- a/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerTemplateNonBlockingAsyncTest.java
+++ b/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerTemplateNonBlockingAsyncTest.java
@@ -16,21 +16,21 @@
  */
 package org.apache.camel.impl;
 
-import org.apache.camel.Exchange;
-import org.junit.Assert;
-
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
+import org.apache.camel.Exchange;
+import org.junit.Assert;
+
 /**
  * @version
  */
-public class DefaultProducerTemplateNonBlockingAsyncTest extends DefaultProducerTemplateAsyncTest{
+public class DefaultProducerTemplateNonBlockingAsyncTest extends DefaultProducerTemplateAsyncTest
{
     @Override
     protected void setUp() throws Exception {
         super.setUp();
         template.stop();
-        template.setSynchronous(true);
+        template.setThreadedAsyncMode(false);
         template.start();
     }
 


Mime
View raw message