camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ra...@apache.org
Subject svn commit: r1442004 - in /camel/branches/camel-2.10.x/components/camel-jms: ./ src/main/java/org/apache/camel/component/jms/ src/test/java/org/apache/camel/component/jms/
Date Mon, 04 Feb 2013 01:44:25 GMT
Author: raulk
Date: Mon Feb  4 01:44:24 2013
New Revision: 1442004

URL: http://svn.apache.org/viewvc?rev=1442004&view=rev
Log:
CAMEL-5974 Allow specifying the default type of TaskExecutor used by the DMLC (camel-jms)

Added:
    camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultTaskExecutorType.java
      - copied unchanged from r1442002, camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultTaskExecutorType.java
    camel/branches/camel-2.10.x/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsDefaultTaskExecutorTypeTest.java
      - copied unchanged from r1442002, camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsDefaultTaskExecutorTypeTest.java
Modified:
    camel/branches/camel-2.10.x/components/camel-jms/   (props changed)
    camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultJmsMessageListenerContainer.java
    camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java
    camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
    camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java

Propchange: camel/branches/camel-2.10.x/components/camel-jms/
------------------------------------------------------------------------------
--- svn:mergeinfo (added)
+++ svn:mergeinfo Mon Feb  4 01:44:24 2013
@@ -0,0 +1,2 @@
+/camel/trunk:1441998,1442002
+/camel/trunk/components/camel-jms:1359013,1359197,1359226,1359265,1359341,1360031,1360525-1360527,1360581,1360583,1360719,1361116-1361117,1361126,1361477,1361919,1361973,1365013,1365061,1365616,1365917,1365939,1366272,1366321,1369457,1369490,1369671,1370254,1370783,1371570,1371967,1372244-1372245,1372259,1372329,1372354,1372485,1372833,1372845,1372863,1373374,1373402,1373442,1373453,1373691-1373692,1373722,1373751,1373763,1373785,1374110,1374133,1374290,1374332,1374388,1374530,1374539,1374582-1374584,1374639,1374709,1374926,1374949,1375019,1375027,1375439,1375458,1375460,1375900,1375904,1375936,1376088,1376110,1376140,1376375,1376377,1376870,1377243,1377256,1377272,1377395,1377400,1377402,1377407,1377413,1377608,1378476,1378731,1378825,1378827,1379320,1379324,1379354,1379709,1380157,1380232,1381089,1381137,1381182,1381196,1381552,1381898,1382039,1382416,1382423,1382433,1383220,1383824,1383826,1383858,1384122,1384226,1384316,1384369,1384447,1384609,1384799,1384815,1386495,138
 6607,1386671,1387019-1387020,1387142,1387545,1387808,1388345,1388750,1388762,1389251,1389326,1389813,1389941,1390820,1390946,1391008,1391021,1391044,1391339,1391346,1391373,1391378,1391439,1391466,1391954,1391958,1392188,1392869,1393291,1393294,1393962,1394388,1394408,1394479,1394577,1394932,1394940,1394943,1394991,1395046,1395085,1395226,1395580,1395626,1395645,1395883,1395891,1398047,1398694,1398756,1398760,1398789,1398797,1398810,1399134,1399165,1399317,1399333,1399519,1399664,1399808,1400074,1400097,1400403,1400729,1400734,1400831,1400863,1400876,1401151,1401159-1401166,1401209,1401264,1401578,1401585,1402383,1402485,1402738,1403186,1403202,1403725,1403863,1403971,1403988,1403990,1404015,1404020,1404031,1404056,1404172,1404233,1404291,1404303,1404474,1404490-1404491,1404496,1404508-1404510,1404519,1404532,1404921,1404934,1405313,1405662,1405678,1405727,1405932,1406066,1406948-1406952,1407104,1407395,1407487-1407488,1407736,1407744,1407748,1407750,1407752,1407764,1407768,
 1407776,1407818,1407826,1407970,1407975,1407978,1407985,1407987,1407992,1407995,1408185-1408186,1408188,1408193,1408262,1408269,1408297,1408306,1408365,1408373,1408660-1408661,1408667,1408670,1408704,1409155,1409215,1409246,1409398,1409424,1409651,1409678,1409689,1409734,1409787,1410217,1410276,1410336,1410346,1410709,1410906,1410952,1410959,1411129,1411217,1411414,1411617,1411664,1412012,1412100,1412106,1412144,1412434,1412473,1412492,1412495,1412499,1412524,1412672,1412774,1412907,1413312,1413317,1413979,1414022,1414564,1414571,1414740,1414836,1414857,1415129,1415534,1415564,1415838,1415884,1415888,1415979,1416006,1416246,1417318-1417330,1418239,1418595,1418616,1418708,1419305,1419415,1419629,1419655,1419729,1419757,1420038,1420051,1420112,1420523-1420524,1420534,1420610,1421306,1421712,1421723,1421764,1421864,1422241,1422568,1422792,1422807,1422957,1422963,1423299,1423301,1423304,1423797,1423808,1423929,1424386,1424398,1424437,1424915,1425102,1425106,1425262,1425295,14254
 45,1425653,1425724,1426385,1426459,1426748,1427806,1427815,1427825,1427853,1427989,1428463,1429218,1429271,1429280,1429282,1429477,1429527,1429529,1429707,1430724,1430731,1431152,1431273,1431280,1431291,1431297,1431336,1431626,1431642,1431652,1431705,1431781,1431784,1431813,1431819-1431820,1431891,1431902,1431933,1432413,1432435,1432629,1432632,1433518-1433519,1433855,1433891,1435219,1435497,1435580,1435700,1435812,1435861,1436244,1436308,1436384,1436389,1436420,1436438,1436441,1436487,1436505,1436603,1436619,1436759,1436904,1436944,1436980,1437139,1437158,1437160,1437208,1437283,1438017,1438063,1438363,1438380,1438499-1438500,1438662,1438880,1439433,1439451,1439468,1439473,1439639,1439771,1440280,1440328,1440360,1441387,1441392,1441508,1441518,1441876,1441879,1442002

Modified: camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultJmsMessageListenerContainer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultJmsMessageListenerContainer.java?rev=1442004&r1=1442003&r2=1442004&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultJmsMessageListenerContainer.java
(original)
+++ camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultJmsMessageListenerContainer.java
Mon Feb  4 01:44:24 2013
@@ -20,6 +20,7 @@ import org.apache.camel.util.concurrent.
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 /**
  * The default {@link DefaultMessageListenerContainer container} which listen for messages
@@ -46,19 +47,38 @@ public class DefaultJmsMessageListenerCo
 
     /**
      * Create a default TaskExecutor. Called if no explicit TaskExecutor has been specified.
-     * <p>The default implementation builds a {@link org.springframework.core.task.SimpleAsyncTaskExecutor}
-     * with the specified bean name and using Camel's {@link org.apache.camel.spi.ExecutorServiceManager}
+     * <p />
+     * The type of {@link TaskExecutor} will depend on the value of
+     * {@link JmsConfiguration#getDefaultTaskExecutorType()}. For more details, refer to
the Javadoc of
+     * {@link DefaultTaskExecutorType}.
+     * <p />
+     * In all cases, it uses the specified bean name and Camel's {@link org.apache.camel.spi.ExecutorServiceManager}
      * to resolve the thread name.
-     * @see org.springframework.core.task.SimpleAsyncTaskExecutor#SimpleAsyncTaskExecutor(String)
+     * @see JmsConfiguration#setDefaultTaskExecutorType(DefaultTaskExecutorType)
+     * @see ThreadPoolTaskExecutor#setBeanName(String)
      */
     @Override
     protected TaskExecutor createDefaultTaskExecutor() {
         String pattern = endpoint.getCamelContext().getExecutorServiceManager().getThreadNamePattern();
-        String beanName = getBeanName();
+        String beanName = getBeanName() == null ? endpoint.getThreadName() : getBeanName();
 
-        SimpleAsyncTaskExecutor answer = new SimpleAsyncTaskExecutor(beanName);
-        answer.setThreadFactory(new CamelThreadFactory(pattern, beanName, true));
-        return answer;
+        if (endpoint.getDefaultTaskExecutorType() == DefaultTaskExecutorType.ThreadPool)
{
+            ThreadPoolTaskExecutor answer = new ThreadPoolTaskExecutor();
+            answer.setBeanName(beanName);
+            answer.setThreadFactory(new CamelThreadFactory(pattern, beanName, true));
+            answer.setCorePoolSize(endpoint.getConcurrentConsumers());
+            // Direct hand-off mode. Do not queue up tasks: assign it to a thread immediately.
+            // We set no upper-bound on the thread pool (no maxPoolSize) as it's already
implicitly constrained by
+            // maxConcurrentConsumers on the DMLC itself (i.e. DMLC will only grow up to
a level of concurrency as
+            // defined by maxConcurrentConsumers).
+            answer.setQueueCapacity(0);
+            answer.initialize();
+            return answer;
+        } else {
+            SimpleAsyncTaskExecutor answer = new SimpleAsyncTaskExecutor(beanName);
+            answer.setThreadFactory(new CamelThreadFactory(pattern, beanName, true));
+            return answer;
+        }
     }
-
+    
 }

Modified: camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java?rev=1442004&r1=1442003&r2=1442004&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java
(original)
+++ camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java
Mon Feb  4 01:44:24 2013
@@ -18,6 +18,7 @@ package org.apache.camel.component.jms;
 
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
+
 import javax.jms.ConnectionFactory;
 import javax.jms.ExceptionListener;
 import javax.jms.Session;
@@ -184,7 +185,7 @@ public class JmsComponent extends Defaul
     public void setCacheLevelName(String cacheName) {
         getConfiguration().setCacheLevelName(cacheName);
     }
-    
+
     public void setReplyToCacheLevelName(String cacheName) {
         getConfiguration().setReplyToCacheLevelName(cacheName);
     }
@@ -236,7 +237,7 @@ public class JmsComponent extends Defaul
     public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) {
         getConfiguration().setIdleTaskExecutionLimit(idleTaskExecutionLimit);
     }
-    
+
     public void setIdleConsumerLimit(int idleConsumerLimit) {
         getConfiguration().setIdleConsumerLimit(idleConsumerLimit);
     }

Modified: camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java?rev=1442004&r1=1442003&r2=1442004&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
(original)
+++ camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
Mon Feb  4 01:44:24 2013
@@ -49,7 +49,7 @@ import org.springframework.util.ErrorHan
 import static org.apache.camel.component.jms.JmsMessageHelper.normalizeDestinationName;
 
 /**
- * @version 
+ * @version
  */
 public class JmsConfiguration implements Cloneable {
 
@@ -137,6 +137,7 @@ public class JmsConfiguration implements
     private boolean allowNullBody = true;
     private MessageListenerContainerFactory messageListenerContainerFactory;
     private boolean includeSentJMSMessageID;
+    private DefaultTaskExecutorType defaultTaskExecutorType;
 
     public JmsConfiguration() {
     }
@@ -386,7 +387,7 @@ public class JmsConfiguration implements
         case Default:
             return new DefaultJmsMessageListenerContainer(endpoint);
         case Custom:
-            return getCustomMessageListenerContainer(endpoint);            
+            return getCustomMessageListenerContainer(endpoint);
         default:
             throw new IllegalArgumentException("Unknown consumer type: " + consumerType);
         }
@@ -1313,4 +1314,16 @@ public class JmsConfiguration implements
     public void setIncludeSentJMSMessageID(boolean includeSentJMSMessageID) {
         this.includeSentJMSMessageID = includeSentJMSMessageID;
     }
+
+    public DefaultTaskExecutorType getDefaultTaskExecutorType() {
+        return defaultTaskExecutorType;
+    }
+
+    /**
+     * Indicates what type of {@link TaskExecutor} to use by default for JMS consumers.
+     * Refer to the documentation of {@link DefaultTaskExecutorType} for available options.
+     */
+    public void setDefaultTaskExecutorType(DefaultTaskExecutorType defaultTaskExecutorType)
{
+        this.defaultTaskExecutorType = defaultTaskExecutorType;
+    }
 }

Modified: camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java?rev=1442004&r1=1442003&r2=1442004&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
(original)
+++ camel/branches/camel-2.10.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
Mon Feb  4 01:44:24 2013
@@ -62,7 +62,7 @@ import org.springframework.util.ErrorHan
 /**
  * A <a href="http://activemq.apache.org/jms.html">JMS Endpoint</a>
  *
- * @version 
+ * @version
  */
 @ManagedResource(description = "Managed JMS Endpoint")
 public class JmsEndpoint extends DefaultEndpoint implements HeaderFilterStrategyAware, MultipleConsumersSupport,
Service {
@@ -176,24 +176,32 @@ public class JmsEndpoint extends Default
         listenerContainer.setPubSubDomain(pubSubDomain);
 
         // include destination name as part of thread and transaction name
-        String consumerName = "JmsConsumer[" + getEndpointConfiguredDestinationName() + "]";
+        String consumerName = getThreadName();
 
         if (configuration.getTaskExecutor() != null) {
             if (log.isDebugEnabled()) {
                 log.debug("Using custom TaskExecutor: {} on listener container: {}", configuration.getTaskExecutor(),
listenerContainer);
             }
             setContainerTaskExecutor(listenerContainer, configuration.getTaskExecutor());
-        } else {
+        } else if ((listenerContainer instanceof DefaultJmsMessageListenerContainer &&
configuration.getDefaultTaskExecutorType() == null)
+                || !(listenerContainer instanceof DefaultJmsMessageListenerContainer)) {
+            // preserve backwards compatibility if an explicit Default TaskExecutor Type
was not set;
+            // otherwise, defer the creation of the TaskExecutor
             // use a cached pool as DefaultMessageListenerContainer will throttle pool sizing
             ExecutorService executor = getCamelContext().getExecutorServiceManager().newCachedThreadPool(consumer,
consumerName);
             setContainerTaskExecutor(listenerContainer, executor);
+        } else {
+            // do nothing, as we're working with a DefaultJmsMessageListenerContainer with
an explicit DefaultTaskExecutorType,
+            // so DefaultJmsMessageListenerContainer#createDefaultTaskExecutor will handle
the creation
+            log.debug("Deferring creation of TaskExecutor for listener container: {} as per
policy: {}", 
+                    listenerContainer, configuration.getDefaultTaskExecutorType());
         }
-        
+
         // set a default transaction name if none provided
         if (configuration.getTransactionName() == null) {
             if (listenerContainer instanceof DefaultMessageListenerContainer) {
                 ((DefaultMessageListenerContainer) listenerContainer).setTransactionName(consumerName);
-            }            
+            }
         }
     }
 
@@ -271,6 +279,10 @@ public class JmsEndpoint extends Default
         return true;
     }
 
+    public String getThreadName() {
+        return "JmsConsumer[" + getEndpointConfiguredDestinationName() + "]";
+    }
+    
     // Properties
     // -------------------------------------------------------------------------
 
@@ -448,7 +460,7 @@ public class JmsEndpoint extends Default
     public String getCacheLevelName() {
         return getConfiguration().getCacheLevelName();
     }
-    
+
     @ManagedAttribute
     public String getReplyToCacheLevelName() {
         return getConfiguration().getReplyToCacheLevelName();
@@ -489,7 +501,7 @@ public class JmsEndpoint extends Default
     public LoggingLevel getErrorHandlerLoggingLevel() {
         return getConfiguration().getErrorHandlerLoggingLevel();
     }
-    
+
     @ManagedAttribute
     public boolean isErrorHandlerLogStackTrace() {
         return getConfiguration().isErrorHandlerLogStackTrace();
@@ -509,7 +521,7 @@ public class JmsEndpoint extends Default
     public int getIdleConsumerLimit() {
         return getConfiguration().getIdleConsumerLimit();
     }
-    
+
     public JmsOperations getJmsOperations() {
         return getConfiguration().getJmsOperations();
     }
@@ -724,7 +736,7 @@ public class JmsEndpoint extends Default
     public void setCacheLevelName(String cacheName) {
         getConfiguration().setCacheLevelName(cacheName);
     }
-    
+
     @ManagedAttribute
     public void setReplyToCacheLevelName(String cacheName) {
         getConfiguration().setReplyToCacheLevelName(cacheName);
@@ -795,7 +807,7 @@ public class JmsEndpoint extends Default
     public void setIdleConsumerLimit(int idleConsumerLimit) {
         getConfiguration().setIdleConsumerLimit(idleConsumerLimit);
     }
-    
+
     public void setJmsOperations(JmsOperations jmsOperations) {
         getConfiguration().setJmsOperations(jmsOperations);
     }
@@ -1040,7 +1052,7 @@ public class JmsEndpoint extends Default
     public void setAllowNullBody(boolean allowNullBody) {
         configuration.setAllowNullBody(allowNullBody);
     }
-    
+
     @ManagedAttribute
     public boolean isIncludeSentJMSMessageID() {
         return configuration.isIncludeSentJMSMessageID();
@@ -1051,6 +1063,14 @@ public class JmsEndpoint extends Default
         configuration.setIncludeSentJMSMessageID(includeSentJMSMessageID);
     }
 
+    public DefaultTaskExecutorType getDefaultTaskExecutorType() {
+        return configuration.getDefaultTaskExecutorType();
+    }
+
+    public void setDefaultTaskExecutorType(DefaultTaskExecutorType type) {
+        configuration.setDefaultTaskExecutorType(type);
+    }
+
     public MessageListenerContainerFactory getMessageListenerContainerFactory() {
         return configuration.getMessageListenerContainerFactory();
     }



Mime
View raw message