activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dej...@apache.org
Subject svn commit: r1365943 - in /activemq/trunk/activemq-core/src/main/java/org/apache/activemq: ActiveMQConnection.java ActiveMQConnectionFactory.java thread/TaskRunnerFactory.java
Date Thu, 26 Jul 2012 10:45:42 GMT
Author: dejanb
Date: Thu Jul 26 10:45:42 2012
New Revision: 1365943

URL: http://svn.apache.org/viewvc?rev=1365943&view=rev
Log:
https://issues.apache.org/jira/browse/AMQ-3885 - allow setting resjected task handler on thread
pool executor

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/TaskRunnerFactory.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java?rev=1365943&r1=1365942&r2=1365943&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
Thu Jul 26 10:45:42 2012
@@ -28,6 +28,7 @@ import java.util.concurrent.ConcurrentHa
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -203,6 +204,7 @@ public class ActiveMQConnection implemen
     private boolean nonBlockingRedelivery = false;
 
     private int maxThreadPoolSize = DEFAULT_THREAD_POOL_SIZE;
+    private RejectedExecutionHandler rejectedTaskHandler = null;
 
     /**
      * Construct an <code>ActiveMQConnection</code>
@@ -985,9 +987,11 @@ public class ActiveMQConnection implemen
     }
 
     public TaskRunnerFactory getSessionTaskRunner() {
+        System.out.println(maxThreadPoolSize);
         synchronized (this) {
             if (sessionTaskRunner == null) {
                 sessionTaskRunner = new TaskRunnerFactory("ActiveMQ Session Task", ThreadPriorities.INBOUND_CLIENT_SESSION,
false, 1000, isUseDedicatedTaskRunner(), maxThreadPoolSize);
+                sessionTaskRunner.setRejectedTaskHandler(rejectedTaskHandler);
             }
         }
         return sessionTaskRunner;
@@ -2595,4 +2599,12 @@ public class ActiveMQConnection implemen
         this.queueOnlyConnection = true;
         return this;
     }
+
+    public RejectedExecutionHandler getRejectedTaskHandler() {
+        return rejectedTaskHandler;
+    }
+
+    public void setRejectedTaskHandler(RejectedExecutionHandler rejectedTaskHandler) {
+        this.rejectedTaskHandler = rejectedTaskHandler;
+    }
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java?rev=1365943&r1=1365942&r2=1365943&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
Thu Jul 26 10:45:42 2012
@@ -22,6 +22,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
 
@@ -131,6 +132,7 @@ public class ActiveMQConnectionFactory e
     private boolean nonBlockingRedelivery = false;
     private int maxThreadPoolSize = ActiveMQConnection.DEFAULT_THREAD_POOL_SIZE;
     private TaskRunnerFactory sessionTaskRunner;
+    private RejectedExecutionHandler rejectedTaskHandler = null;
 
     // /////////////////////////////////////////////
     //
@@ -344,6 +346,7 @@ public class ActiveMQConnectionFactory e
         connection.setNonBlockingRedelivery(isNonBlockingRedelivery());
         connection.setMaxThreadPoolSize(getMaxThreadPoolSize());
         connection.setSessionTaskRunner(getSessionTaskRunner());
+        connection.setRejectedTaskHandler(getRejectedTaskHandler());
         if (transportListener != null) {
             connection.addTransportListener(transportListener);
         }
@@ -1116,4 +1119,12 @@ public class ActiveMQConnectionFactory e
     public void setSessionTaskRunner(TaskRunnerFactory sessionTaskRunner) {
         this.sessionTaskRunner = sessionTaskRunner;
     }
+
+    public RejectedExecutionHandler getRejectedTaskHandler() {
+        return rejectedTaskHandler;
+    }
+
+    public void setRejectedTaskHandler(RejectedExecutionHandler rejectedTaskHandler) {
+        this.rejectedTaskHandler = rejectedTaskHandler;
+    }
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/TaskRunnerFactory.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/TaskRunnerFactory.java?rev=1365943&r1=1365942&r2=1365943&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/TaskRunnerFactory.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/TaskRunnerFactory.java
Thu Jul 26 10:45:42 2012
@@ -18,7 +18,7 @@ package org.apache.activemq.thread;
 
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -46,6 +46,7 @@ public class TaskRunnerFactory implement
     private boolean dedicatedTaskRunner;
     private AtomicBoolean initDone = new AtomicBoolean(false);
     private int maxThreadPoolSize = Integer.MAX_VALUE;
+    private RejectedExecutionHandler rejectedTaskHandler = null;
 
     public TaskRunnerFactory() {
         this("ActiveMQ Task", Thread.NORM_PRIORITY, true, 1000);
@@ -118,6 +119,9 @@ public class TaskRunnerFactory implement
                 return thread;
             }
         });
+        if (rejectedTaskHandler != null) {
+            rc.setRejectedExecutionHandler(rejectedTaskHandler);
+        }
         return rc;
     }
 
@@ -176,4 +180,12 @@ public class TaskRunnerFactory implement
     public void setMaxThreadPoolSize(int maxThreadPoolSize) {
         this.maxThreadPoolSize = maxThreadPoolSize;
     }
+
+    public RejectedExecutionHandler getRejectedTaskHandler() {
+        return rejectedTaskHandler;
+    }
+
+    public void setRejectedTaskHandler(RejectedExecutionHandler rejectedTaskHandler) {
+        this.rejectedTaskHandler = rejectedTaskHandler;
+    }
 }



Mime
View raw message