activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rajdav...@apache.org
Subject svn commit: r946600 [1/2] - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/ main/java/org/apache/activemq/broker/ main/java/org/apache/activemq/broker/jmx/ main/java/org/apache/activemq/broker/region/ main/java/org/apache/activemq/...
Date Thu, 20 May 2010 12:02:11 GMT
Author: rajdavies
Date: Thu May 20 12:02:10 2010
New Revision: 946600

URL: http://svn.apache.org/viewvc?rev=946600&view=rev
Log:
fixes for https://issues.apache.org/activemq/browse/AMQ-2620 and
https://issues.apache.org/activemq/browse/AMQ-2568

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/Broker.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/BrokerFilter.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/EmptyBroker.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/ErrorBroker.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/MutableBrokerFilter.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFactoryImpl.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/AbortSlowConsumerStrategy.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedCountSubscriptionRecoveryPolicy.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedSizedSubscriptionRecoveryPolicy.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/LastImageSubscriptionRecoveryPolicy.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/NoSubscriptionRecoveryPolicy.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/SlowConsumerStrategy.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/SubscriptionRecoveryPolicy.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/TimedSubscriptionRecoveryPolicy.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapterFactory.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/Scheduler.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/usage/SystemUsage.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/usage/Usage.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/BrokerRestartTestSupport.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/DurableConsumerTest.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/embedded/EmbeddedActiveMQ.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/DuplexNetworkTest.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usage/MemoryUsageTest.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=946600&r1=946599&r2=946600&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 May 20 12:02:10 2010
@@ -87,6 +87,7 @@ import org.apache.activemq.management.JM
 import org.apache.activemq.management.StatsCapable;
 import org.apache.activemq.management.StatsImpl;
 import org.apache.activemq.state.CommandVisitorAdapter;
+import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.thread.TaskRunnerFactory;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportListener;
@@ -114,7 +115,7 @@ public class ActiveMQConnection implemen
     protected boolean alwaysSessionAsync = true;
 
     private TaskRunnerFactory sessionTaskRunner;
-    private final ThreadPoolExecutor asyncConnectionThread;
+    private final ThreadPoolExecutor executor;
 
     // Connection state variables
     private final ConnectionInfo info;
@@ -188,6 +189,7 @@ public class ActiveMQConnection implemen
     private boolean useDedicatedTaskRunner;
     protected volatile CountDownLatch transportInterruptionProcessingComplete;
     private long consumerFailoverRedeliveryWaitPeriod;
+    private final Scheduler scheduler;
 
     /**
      * Construct an <code>ActiveMQConnection</code>
@@ -204,16 +206,16 @@ public class ActiveMQConnection implemen
 
         // Configure a single threaded executor who's core thread can timeout if
         // idle
-        asyncConnectionThread = new ThreadPoolExecutor(1, 1, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
+        executor = new ThreadPoolExecutor(1, 1, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
             public Thread newThread(Runnable r) {
-                Thread thread = new Thread(r, "ActiveMQ Connection Worker: " + transport);
+                Thread thread = new Thread(r, "ActiveMQ Connection Executor: " + transport);
                 thread.setDaemon(true);
                 return thread;
             }
         });
         // asyncConnectionThread.allowCoreThreadTimeOut(true);
-
-        this.info = new ConnectionInfo(new ConnectionId(CONNECTION_ID_GENERATOR.generateId()));
+        String uniqueId = CONNECTION_ID_GENERATOR.generateId();
+        this.info = new ConnectionInfo(new ConnectionId(uniqueId));
         this.info.setManageable(true);
         this.info.setFaultTolerant(transport.isFaultTolerant());
         this.connectionSessionId = new SessionId(info.getConnectionId(), -1);
@@ -224,6 +226,8 @@ public class ActiveMQConnection implemen
         this.factoryStats.addConnection(this);
         this.timeCreated = System.currentTimeMillis();
         this.connectionAudit.setCheckForDuplicates(transport.isFaultTolerant());
+        this.scheduler = new Scheduler("ActiveMQConnection["+uniqueId+"] Scheduler");
+        this.scheduler.start();
     }
 
     protected void setUserName(String userName) {
@@ -609,6 +613,14 @@ public class ActiveMQConnection implemen
                         advisoryConsumer.dispose();
                         advisoryConsumer = null;
                     }
+                    if (this.scheduler != null) {
+                        try {
+                            this.scheduler.stop();
+                        } catch (Exception e) {
+                            JMSException ex =  JMSExceptionSupport.create(e);
+                            throw ex;
+                        }
+                    }
 
                     long lastDeliveredSequenceId = 0;
                     for (Iterator<ActiveMQSession> i = this.sessions.iterator(); i.hasNext();) {
@@ -656,8 +668,8 @@ public class ActiveMQConnection implemen
             }
         } finally {
             try {
-                if (asyncConnectionThread != null){
-                    asyncConnectionThread.shutdown();
+                if (executor != null){
+                    executor.shutdown();
                 }
             }catch(Throwable e) {
                 LOG.error("Error shutting down thread pool " + e,e);
@@ -1719,7 +1731,7 @@ public class ActiveMQConnection implemen
 
                     @Override
                     public Response processConnectionError(final ConnectionError error) throws Exception {
-                        asyncConnectionThread.execute(new Runnable() {
+                        executor.execute(new Runnable() {
                             public void run() {
                                 onAsyncException(error.getException());
                             }
@@ -1779,7 +1791,7 @@ public class ActiveMQConnection implemen
     public void onClientInternalException(final Throwable error) {
         if ( !closed.get() && !closing.get() ) {
             if ( this.clientInternalExceptionListener != null ) {
-                asyncConnectionThread.execute(new Runnable() {
+                executor.execute(new Runnable() {
                     public void run() {
                         ActiveMQConnection.this.clientInternalExceptionListener.onException(error);
                     }
@@ -1804,7 +1816,7 @@ public class ActiveMQConnection implemen
                 }
                 final JMSException e = (JMSException)error;
 
-                asyncConnectionThread.execute(new Runnable() {
+                executor.execute(new Runnable() {
                     public void run() {
                         ActiveMQConnection.this.exceptionListener.onException(e);
                     }
@@ -1819,7 +1831,7 @@ public class ActiveMQConnection implemen
     public void onException(final IOException error) {
 		onAsyncException(error);
 		if (!closing.get() && !closed.get()) {
-			asyncConnectionThread.execute(new Runnable() {
+			executor.execute(new Runnable() {
 				public void run() {
 					transportFailed(error);
 					ServiceSupport.dispose(ActiveMQConnection.this.transport);
@@ -2297,4 +2309,12 @@ public class ActiveMQConnection implemen
     public long getConsumerFailoverRedeliveryWaitPeriod() {
         return consumerFailoverRedeliveryWaitPeriod;
     }
+    
+    protected Scheduler getScheduler() {
+        return this.scheduler;
+    }
+    
+    protected ThreadPoolExecutor getExecutor() {
+        return this.executor;
+    }
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java Thu May 20 12:02:10 2010
@@ -29,7 +29,6 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
-
 import javax.jms.IllegalStateException;
 import javax.jms.InvalidDestinationException;
 import javax.jms.JMSException;
@@ -37,7 +36,6 @@ import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageListener;
 import javax.jms.TransactionRolledBackException;
-
 import org.apache.activemq.blob.BlobDownloader;
 import org.apache.activemq.command.ActiveMQBlobMessage;
 import org.apache.activemq.command.ActiveMQDestination;
@@ -111,7 +109,7 @@ public class ActiveMQMessageConsumer imp
     }
 
     private static final Log LOG = LogFactory.getLog(ActiveMQMessageConsumer.class);
-    protected static final Scheduler scheduler = Scheduler.getInstance();
+    protected final Scheduler scheduler;
     protected final ActiveMQSession session;
     protected final ConsumerInfo info;
 
@@ -130,17 +128,17 @@ public class ActiveMQMessageConsumer imp
     private int ackCounter;
     private int dispatchedCount;
     private final AtomicReference<MessageListener> messageListener = new AtomicReference<MessageListener>();
-    private JMSConsumerStatsImpl stats;
+    private final JMSConsumerStatsImpl stats;
 
     private final String selector;
     private boolean synchronizationRegistered;
-    private AtomicBoolean started = new AtomicBoolean(false);
+    private final AtomicBoolean started = new AtomicBoolean(false);
 
     private MessageAvailableListener availableListener;
 
     private RedeliveryPolicy redeliveryPolicy;
     private boolean optimizeAcknowledge;
-    private AtomicBoolean deliveryingAcknowledgements = new AtomicBoolean();
+    private final AtomicBoolean deliveryingAcknowledgements = new AtomicBoolean();
     private ExecutorService executorService;
     private MessageTransformer transformer;
     private boolean clearDispatchList;
@@ -152,7 +150,7 @@ public class ActiveMQMessageConsumer imp
     private IOException failureError;
     
     private long optimizeAckTimestamp = System.currentTimeMillis();
-    private long optimizeAckTimeout = 300;
+    private final long optimizeAckTimeout = 300;
     private long failoverRedeliveryWaitPeriod = 0;
 
     /**
@@ -202,6 +200,7 @@ public class ActiveMQMessageConsumer imp
         }
 
         this.session = session;
+        this.scheduler = session.getScheduler();
         this.redeliveryPolicy = session.connection.getRedeliveryPolicy();
         setTransformer(session.getTransformer());
 
@@ -634,10 +633,12 @@ public class ActiveMQMessageConsumer imp
         if (!unconsumedMessages.isClosed()) {
             if (session.getTransactionContext().isInTransaction()) {
                 session.getTransactionContext().addSynchronization(new Synchronization() {
+                    @Override
                     public void afterCommit() throws Exception {
                         doClose();
                     }
 
+                    @Override
                     public void afterRollback() throws Exception {
                         doClose();
                     }
@@ -912,16 +913,19 @@ public class ActiveMQMessageConsumer imp
             if (!synchronizationRegistered) {
                 synchronizationRegistered = true;
                 session.getTransactionContext().addSynchronization(new Synchronization() {
+                    @Override
                     public void beforeEnd() throws Exception {
                         acknowledge();
                         synchronizationRegistered = false;
                     }
 
+                    @Override
                     public void afterCommit() throws Exception {
                         commit();
                         synchronizationRegistered = false;
                     }
 
+                    @Override
                     public void afterRollback() throws Exception {
                         rollback();
                         synchronizationRegistered = false;
@@ -1325,6 +1329,7 @@ public class ActiveMQMessageConsumer imp
         unconsumedMessages.stop();
     }
 
+    @Override
     public String toString() {
         return "ActiveMQMessageConsumer { value=" + info.getConsumerId() + ", started=" + started.get()
                + " }";

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java Thu May 20 12:02:10 2010
@@ -19,13 +19,11 @@ package org.apache.activemq;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
-
 import javax.jms.Destination;
 import javax.jms.IllegalStateException;
 import javax.jms.InvalidDestinationException;
 import javax.jms.JMSException;
 import javax.jms.Message;
-
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ProducerAck;
 import org.apache.activemq.command.ProducerId;
@@ -73,9 +71,9 @@ public class ActiveMQMessageProducer ext
     protected ProducerInfo info;
     protected boolean closed;
 
-    private JMSProducerStatsImpl stats;
+    private final JMSProducerStatsImpl stats;
     private AtomicLong messageSequence;
-    private long startTime;
+    private final long startTime;
     private MessageTransformer transformer;
     private MemoryUsage producerWindow;
 
@@ -93,6 +91,7 @@ public class ActiveMQMessageProducer ext
         // size > 0
         if (session.connection.getProtocolVersion() >= 3 && this.info.getWindowSize() > 0) {
             producerWindow = new MemoryUsage("Producer Window: " + producerId);
+            producerWindow.setExecutor(session.getConnectionExecutor());
             producerWindow.setLimit(this.info.getWindowSize());
             producerWindow.start();
         }
@@ -164,6 +163,7 @@ public class ActiveMQMessageProducer ext
      * 
      * @throws IllegalStateException
      */
+    @Override
     protected void checkClosed() throws IllegalStateException {
         if (closed) {
             throw new IllegalStateException("The producer is closed");
@@ -280,6 +280,7 @@ public class ActiveMQMessageProducer ext
         this.info = info;
     }
 
+    @Override
     public String toString() {
         return "ActiveMQMessageProducer { value=" + info.getProducerId() + " }";
     }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java Thu May 20 12:02:10 2010
@@ -24,8 +24,8 @@ import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicBoolean;
-
 import javax.jms.BytesMessage;
 import javax.jms.Destination;
 import javax.jms.IllegalStateException;
@@ -53,7 +53,6 @@ import javax.jms.TopicPublisher;
 import javax.jms.TopicSession;
 import javax.jms.TopicSubscriber;
 import javax.jms.TransactionRolledBackException;
-
 import org.apache.activemq.blob.BlobDownloader;
 import org.apache.activemq.blob.BlobTransferPolicy;
 import org.apache.activemq.blob.BlobUploader;
@@ -198,7 +197,8 @@ public class ActiveMQSession implements 
     }
 
     private static final Log LOG = LogFactory.getLog(ActiveMQSession.class);
-    protected static final Scheduler scheduler = Scheduler.getInstance();
+    private final Scheduler scheduler;
+    private final ThreadPoolExecutor connectionExecutor;
 
     protected int acknowledgementMode;
     protected final ActiveMQConnection connection;
@@ -220,7 +220,7 @@ public class ActiveMQSession implements 
     protected Object sendMutex = new Object();
 
     private MessageListener messageListener;
-    private JMSSessionStatsImpl stats;
+    private final JMSSessionStatsImpl stats;
     private TransactionContext transactionContext;
     private DeliveryListener deliveryListener;
     private MessageTransformer transformer;
@@ -251,7 +251,8 @@ public class ActiveMQSession implements 
         this.connection.asyncSendPacket(info);
         setTransformer(connection.getTransformer());
         setBlobTransferPolicy(connection.getBlobTransferPolicy());
-
+        this.scheduler=connection.getScheduler();
+        this.connectionExecutor=connection.getExecutor();
         if (connection.isStarted()) {
             start();
         }
@@ -613,11 +614,13 @@ public class ActiveMQSession implements 
                     synchronizationRegistered = true;
                     getTransactionContext().addSynchronization(new Synchronization() {
 
+                                        @Override
                                         public void afterCommit() throws Exception {
                                             doClose();
                                             synchronizationRegistered = false;
                                         }
 
+                                        @Override
                                         public void afterRollback() throws Exception {
                                             doClose();
                                             synchronizationRegistered = false;
@@ -846,6 +849,7 @@ public class ActiveMQSession implements 
                 if (ack.getTransactionId() != null) {
                     getTransactionContext().addSynchronization(new Synchronization() {
 
+                        @Override
                         public void afterRollback() throws Exception {
                             md.getMessage().onMessageRolledBack();
                             // ensure we don't filter this as a duplicate
@@ -1947,6 +1951,7 @@ public class ActiveMQSession implements 
         return executor.getUnconsumedMessages();
     }
 
+    @Override
     public String toString() {
         return "ActiveMQSession {id=" + info.getSessionId() + ",started=" + started.get() + "}";
     }
@@ -2025,4 +2030,12 @@ public class ActiveMQSession implements 
             syncSendPacket(ack);
         }
     }
+    
+    protected Scheduler getScheduler() {
+        return this.scheduler;
+    }
+    
+    protected ThreadPoolExecutor getConnectionExecutor() {
+        return this.connectionExecutor;
+    }
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/Broker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/Broker.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/Broker.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/Broker.java Thu May 20 12:02:10 2010
@@ -18,6 +18,7 @@ package org.apache.activemq.broker;
 
 import java.net.URI;
 import java.util.Set;
+import java.util.concurrent.ThreadPoolExecutor;
 import org.apache.activemq.Service;
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.MessageReference;
@@ -33,6 +34,7 @@ import org.apache.activemq.command.Produ
 import org.apache.activemq.command.SessionInfo;
 import org.apache.activemq.command.TransactionId;
 import org.apache.activemq.store.kahadb.plist.PListStore;
+import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.usage.Usage;
 
 /**
@@ -372,6 +374,10 @@ public interface Broker extends Region, 
      *  configuration
      */
     void nowMasterBroker();
+    
+    Scheduler getScheduler();
+    
+    ThreadPoolExecutor getExecutor();
 
 
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/BrokerFilter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/BrokerFilter.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/BrokerFilter.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/BrokerFilter.java Thu May 20 12:02:10 2010
@@ -19,6 +19,7 @@ package org.apache.activemq.broker;
 import java.net.URI;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ThreadPoolExecutor;
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.MessageReference;
 import org.apache.activemq.broker.region.Subscription;
@@ -40,6 +41,7 @@ import org.apache.activemq.command.Respo
 import org.apache.activemq.command.SessionInfo;
 import org.apache.activemq.command.TransactionId;
 import org.apache.activemq.store.kahadb.plist.PListStore;
+import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.usage.Usage;
 
 /**
@@ -300,4 +302,12 @@ public class BrokerFilter implements Bro
             ConsumerControl control) {
         next.processConsumerControl(consumerExchange, control);
     }
+
+    public Scheduler getScheduler() {
+       return next.getScheduler();
+    }
+
+    public ThreadPoolExecutor getExecutor() {
+       return next.getExecutor();
+    }
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java Thu May 20 12:02:10 2010
@@ -29,6 +29,9 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import javax.annotation.PostConstruct;
@@ -78,9 +81,10 @@ import org.apache.activemq.security.Secu
 import org.apache.activemq.selector.SelectorParser;
 import org.apache.activemq.store.PersistenceAdapter;
 import org.apache.activemq.store.PersistenceAdapterFactory;
-import org.apache.activemq.store.amq.AMQPersistenceAdapterFactory;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
 import org.apache.activemq.store.kahadb.plist.PListStore;
 import org.apache.activemq.store.memory.MemoryPersistenceAdapter;
+import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.thread.TaskRunnerFactory;
 import org.apache.activemq.transport.TransportFactory;
 import org.apache.activemq.transport.TransportServer;
@@ -188,9 +192,10 @@ public class BrokerService implements Se
     private IOExceptionHandler ioExceptionHandler;
     private boolean schedulerSupport = true;
     private File schedulerDirectoryFile;
-    
+    private Scheduler scheduler;
+    private ThreadPoolExecutor executor;
     private boolean slave = true;
-
+    
 	static {
         String localHostName = "localhost";
         try {
@@ -589,6 +594,15 @@ public class BrokerService implements Se
                 }
             }
         }
+        if (this.taskRunnerFactory != null) {
+            this.taskRunnerFactory.shutdown();
+        }
+        if (this.scheduler != null) {
+            this.scheduler.stop();
+        }
+        if (this.executor != null) {
+            this.executor.shutdownNow();
+        }
         LOG.info("ActiveMQ JMS Message Broker (" + getBrokerName() + ", " + brokerId + ") stopped");
         synchronized (shutdownHooks) {
             for (Runnable hook : shutdownHooks) {
@@ -756,9 +770,6 @@ public class BrokerService implements Se
     }
 
     public PersistenceAdapterFactory getPersistenceFactory() {
-        if (persistenceFactory == null) {
-            persistenceFactory = createPersistenceFactory();
-        }
         return persistenceFactory;
     }
 
@@ -848,6 +859,7 @@ public class BrokerService implements Se
         try {
             if (systemUsage == null) {
                 systemUsage = new SystemUsage("Main", getPersistenceAdapter(), getTempDataStore());
+                systemUsage.setExecutor(getExecutor());
                 systemUsage.getMemoryUsage().setLimit(1024 * 1024 * 64); // Default
                                                                          // 64
                                                                          // Meg
@@ -869,6 +881,9 @@ public class BrokerService implements Se
             removeService(this.systemUsage);
         }
         this.systemUsage = memoryManager;
+        if (this.systemUsage.getExecutor()==null) {
+            this.systemUsage.setExecutor(getExecutor());
+        }
         addService(this.systemUsage);
     }
 
@@ -953,11 +968,11 @@ public class BrokerService implements Se
     }
 
     public TaskRunnerFactory getTaskRunnerFactory() {
-        if (taskRunnerFactory == null) {
-            taskRunnerFactory = new TaskRunnerFactory("BrokerService", getTaskRunnerPriority(), true, 1000,
+        if (this.taskRunnerFactory == null) {
+            this.taskRunnerFactory = new TaskRunnerFactory("BrokerService", getTaskRunnerPriority(), true, 1000,
                     isDedicatedTaskRunner());
         }
-        return taskRunnerFactory;
+        return this.taskRunnerFactory;
     }
 
     public void setTaskRunnerFactory(TaskRunnerFactory taskRunnerFactory) {
@@ -1769,10 +1784,10 @@ public class BrokerService implements Se
         RegionBroker regionBroker;
         if (isUseJmx()) {
             regionBroker = new ManagedRegionBroker(this, getManagementContext(), getBrokerObjectName(),
-                    getTaskRunnerFactory(), getConsumerSystemUsage(), destinationFactory, destinationInterceptor);
+                    getTaskRunnerFactory(), getConsumerSystemUsage(), destinationFactory, destinationInterceptor,getScheduler(),getExecutor());
         } else {
             regionBroker = new RegionBroker(this, getTaskRunnerFactory(), getConsumerSystemUsage(), destinationFactory,
-                    destinationInterceptor);
+                    destinationInterceptor,getScheduler(),getExecutor());
         }
         destinationFactory.setRegionBroker(regionBroker);
         regionBroker.setKeepDurableSubsActive(keepDurableSubsActive);
@@ -1850,20 +1865,20 @@ public class BrokerService implements Se
 
     protected PersistenceAdapter createPersistenceAdapter() throws IOException {
         if (isPersistent()) {
-            return getPersistenceFactory().createPersistenceAdapter();
+            PersistenceAdapterFactory fac = getPersistenceFactory();
+            if (fac != null) {
+                return fac.createPersistenceAdapter();
+            }else {
+                KahaDBPersistenceAdapter adaptor = new KahaDBPersistenceAdapter();
+                File dir = new File(getBrokerDataDirectory(),"KahaDB");
+                adaptor.setDirectory(dir);
+                return adaptor;
+            }
         } else {
             return new MemoryPersistenceAdapter();
         }
     }
 
-    protected AMQPersistenceAdapterFactory createPersistenceFactory() {
-        AMQPersistenceAdapterFactory factory = new AMQPersistenceAdapterFactory();
-        factory.setDataDirectory(getBrokerDataDirectory());
-        factory.setTaskRunnerFactory(getPersistenceTaskRunnerFactory());
-        factory.setBrokerName(getBrokerName());
-        return factory;
-    }
-
     protected ObjectName createBrokerObjectName() throws IOException {
         try {
             return new ObjectName(getManagementContext().getJmxDomainName() + ":" + "BrokerName="
@@ -2124,6 +2139,31 @@ public class BrokerService implements Se
             }
         }
     }
+    
+    protected synchronized ThreadPoolExecutor getExecutor() {
+        if (this.executor == null) {
+        this.executor = new ThreadPoolExecutor(1, 10, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
+            public Thread newThread(Runnable runnable) {
+                Thread thread = new Thread(runnable, "Usage Async Task");
+                thread.setDaemon(true);
+                return thread;
+            }
+        });
+        }
+        return this.executor;
+    }
+    
+    protected synchronized Scheduler getScheduler() {
+        if (this.scheduler==null) {
+            this.scheduler = new Scheduler("ActiveMQ Broker["+getBrokerName()+"] Scheduler");
+            try {
+                this.scheduler.start();
+            } catch (Exception e) {
+               LOG.error("Failed to start Scheduler ",e);
+            }
+        }
+        return this.scheduler;
+    }
 
     public Broker getRegionBroker() {
         return regionBroker;
@@ -2251,7 +2291,5 @@ public class BrokerService implements Se
     
     public void setSchedulerDirectory(String schedulerDirectory) {
         setSchedulerDirectoryFile(new File(schedulerDirectory));
-    }
-    
-   
+    }   
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/EmptyBroker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/EmptyBroker.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/EmptyBroker.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/EmptyBroker.java Thu May 20 12:02:10 2010
@@ -20,6 +20,7 @@ import java.net.URI;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ThreadPoolExecutor;
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.MessageReference;
 import org.apache.activemq.broker.region.Subscription;
@@ -41,6 +42,7 @@ import org.apache.activemq.command.Respo
 import org.apache.activemq.command.SessionInfo;
 import org.apache.activemq.command.TransactionId;
 import org.apache.activemq.store.kahadb.plist.PListStore;
+import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.usage.Usage;
 
 /**
@@ -283,4 +285,12 @@ public class EmptyBroker implements Brok
     public void processConsumerControl(ConsumerBrokerExchange consumerExchange,
             ConsumerControl control) {     
     }
+
+    public Scheduler getScheduler() {
+        return null;
+    }
+
+    public ThreadPoolExecutor getExecutor() {
+        return null;
+    }
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/ErrorBroker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/ErrorBroker.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/ErrorBroker.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/ErrorBroker.java Thu May 20 12:02:10 2010
@@ -20,6 +20,7 @@ import java.net.URI;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ThreadPoolExecutor;
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.MessageReference;
 import org.apache.activemq.broker.region.Subscription;
@@ -41,6 +42,7 @@ import org.apache.activemq.command.Respo
 import org.apache.activemq.command.SessionInfo;
 import org.apache.activemq.command.TransactionId;
 import org.apache.activemq.store.kahadb.plist.PListStore;
+import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.usage.Usage;
 
 /**
@@ -302,4 +304,12 @@ public class ErrorBroker implements Brok
             ConsumerControl control) {
         throw new BrokerStoppedException(this.message);
     }
+
+    public Scheduler getScheduler() {
+        throw new BrokerStoppedException(this.message);
+    }
+
+    public ThreadPoolExecutor getExecutor() {
+        throw new BrokerStoppedException(this.message);
+    }
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/MutableBrokerFilter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/MutableBrokerFilter.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/MutableBrokerFilter.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/MutableBrokerFilter.java Thu May 20 12:02:10 2010
@@ -19,6 +19,7 @@ package org.apache.activemq.broker;
 import java.net.URI;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.MessageReference;
@@ -41,6 +42,7 @@ import org.apache.activemq.command.Respo
 import org.apache.activemq.command.SessionInfo;
 import org.apache.activemq.command.TransactionId;
 import org.apache.activemq.store.kahadb.plist.PListStore;
+import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.usage.Usage;
 
 /**
@@ -312,4 +314,12 @@ public class MutableBrokerFilter impleme
         getNext().processConsumerControl(consumerExchange, control);
     }
 
+    public Scheduler getScheduler() {
+       return getNext().getScheduler();
+    }
+
+    public ThreadPoolExecutor getExecutor() {
+       return getNext().getExecutor();
+    }
+
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java Thu May 20 12:02:10 2010
@@ -16,6 +16,28 @@
  */
 package org.apache.activemq.broker.jmx;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Hashtable;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.ThreadPoolExecutor;
+import javax.management.InstanceNotFoundException;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.management.openmbean.CompositeType;
+import javax.management.openmbean.OpenDataException;
+import javax.management.openmbean.TabularData;
+import javax.management.openmbean.TabularDataSupport;
+import javax.management.openmbean.TabularType;
 import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.ConnectionContext;
@@ -41,6 +63,7 @@ import org.apache.activemq.command.Subsc
 import org.apache.activemq.store.MessageRecoveryListener;
 import org.apache.activemq.store.PersistenceAdapter;
 import org.apache.activemq.store.TopicMessageStore;
+import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.thread.TaskRunnerFactory;
 import org.apache.activemq.usage.SystemUsage;
 import org.apache.activemq.util.JMXSupport;
@@ -48,27 +71,6 @@ import org.apache.activemq.util.ServiceS
 import org.apache.activemq.util.SubscriptionKey;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Hashtable;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArraySet;
-import javax.management.InstanceNotFoundException;
-import javax.management.MalformedObjectNameException;
-import javax.management.ObjectName;
-import javax.management.openmbean.CompositeData;
-import javax.management.openmbean.CompositeDataSupport;
-import javax.management.openmbean.CompositeType;
-import javax.management.openmbean.OpenDataException;
-import javax.management.openmbean.TabularData;
-import javax.management.openmbean.TabularDataSupport;
-import javax.management.openmbean.TabularType;
 
 public class ManagedRegionBroker extends RegionBroker {
     private static final Log LOG = LogFactory.getLog(ManagedRegionBroker.class);
@@ -91,18 +93,20 @@ public class ManagedRegionBroker extends
     private Broker contextBroker;
 
     public ManagedRegionBroker(BrokerService brokerService, ManagementContext context, ObjectName brokerObjectName, TaskRunnerFactory taskRunnerFactory, SystemUsage memoryManager,
-                               DestinationFactory destinationFactory, DestinationInterceptor destinationInterceptor) throws IOException {
-        super(brokerService, taskRunnerFactory, memoryManager, destinationFactory, destinationInterceptor);
+                               DestinationFactory destinationFactory, DestinationInterceptor destinationInterceptor,Scheduler scheduler,ThreadPoolExecutor executor) throws IOException {
+        super(brokerService, taskRunnerFactory, memoryManager, destinationFactory, destinationInterceptor,scheduler,executor);
         this.managementContext = context;
         this.brokerObjectName = brokerObjectName;
     }
 
+    @Override
     public void start() throws Exception {
         super.start();
         // build all existing durable subscriptions
         buildExistingSubscriptions();
     }
 
+    @Override
     protected void doStop(ServiceStopper stopper) {
         super.doStop(stopper);
         // lets remove any mbeans not yet removed
@@ -119,18 +123,22 @@ public class ManagedRegionBroker extends
         registeredMBeans.clear();
     }
 
+    @Override
     protected Region createQueueRegion(SystemUsage memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
         return new ManagedQueueRegion(this, destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
     }
 
+    @Override
     protected Region createTempQueueRegion(SystemUsage memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
         return new ManagedTempQueueRegion(this, brokerService, destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
     }
 
+    @Override
     protected Region createTempTopicRegion(SystemUsage memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
         return new ManagedTempTopicRegion(this, destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
     }
 
+    @Override
     protected Region createTopicRegion(SystemUsage memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
         return new ManagedTopicRegion(this, destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
     }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java Thu May 20 12:02:10 2010
@@ -17,9 +17,7 @@
 package org.apache.activemq.broker.region;
 
 import java.io.IOException;
-
 import javax.jms.ResourceAllocationException;
-
 import org.apache.activemq.advisory.AdvisorySupport;
 import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.BrokerService;

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFactoryImpl.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFactoryImpl.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFactoryImpl.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFactoryImpl.java Thu May 20 12:02:10 2010
@@ -55,6 +55,7 @@ public class DestinationFactoryImpl exte
         this.persistenceAdapter = persistenceAdapter;
     }
 
+    @Override
     public void setRegionBroker(RegionBroker broker) {
         if (broker == null) {
             throw new IllegalArgumentException("null broker");
@@ -62,6 +63,7 @@ public class DestinationFactoryImpl exte
         this.broker = broker;
     }
 
+    @Override
     public Set<ActiveMQDestination> getDestinations() {
         return persistenceAdapter.getDestinations();
     }
@@ -69,6 +71,7 @@ public class DestinationFactoryImpl exte
     /**
      * @return instance of {@link Queue} or {@link Topic}
      */
+    @Override
     public Destination createDestination(ConnectionContext context, ActiveMQDestination destination, DestinationStatistics destinationStatistics) throws Exception {
         if (destination.isQueue()) {
             if (destination.isTemporary()) {
@@ -100,6 +103,7 @@ public class DestinationFactoryImpl exte
         }
     }
 
+    @Override
     public void removeDestination(Destination dest) {
         ActiveMQDestination destination = dest.getActiveMQDestination();
         if (!destination.isTemporary()) {
@@ -131,11 +135,12 @@ public class DestinationFactoryImpl exte
         if (broker.getDestinationPolicy() != null) {
             PolicyEntry entry = broker.getDestinationPolicy().getEntryFor(destination);
             if (entry != null) {
-                entry.configure(topic);
+                entry.configure(broker,topic);
             }
         }
     }
 
+    @Override
     public long getLastMessageBrokerSequenceId() throws IOException {
         return persistenceAdapter.getLastMessageBrokerSequenceId();
     }
@@ -144,6 +149,7 @@ public class DestinationFactoryImpl exte
         return persistenceAdapter;
     }
 
+    @Override
     public SubscriptionInfo[] getAllDurableSubscriptions(ActiveMQTopic topic) throws IOException {
         return persistenceAdapter.createTopicMessageStore(topic).getAllSubscriptions();
     }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java Thu May 20 12:02:10 2010
@@ -23,10 +23,8 @@ import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
-
 import javax.jms.InvalidSelectorException;
 import javax.jms.JMSException;
-
 import org.apache.activemq.ActiveMQMessageAudit;
 import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
@@ -55,7 +53,7 @@ import org.apache.commons.logging.LogFac
 public abstract class PrefetchSubscription extends AbstractSubscription {
 
     private static final Log LOG = LogFactory.getLog(PrefetchSubscription.class);
-    protected static final Scheduler scheduler = Scheduler.getInstance();
+    protected final Scheduler scheduler;
     
     protected PendingMessageCursor pending;
     protected final List<MessageReference> dispatched = new CopyOnWriteArrayList<MessageReference>();
@@ -70,12 +68,13 @@ public abstract class PrefetchSubscripti
     private final Object pendingLock = new Object();
     private final Object dispatchLock = new Object();
     protected ActiveMQMessageAudit audit = new ActiveMQMessageAudit();
-    private CountDownLatch okForAckAsDispatchDone = new CountDownLatch(1);
+    private final CountDownLatch okForAckAsDispatchDone = new CountDownLatch(1);
     
     public PrefetchSubscription(Broker broker, SystemUsage usageManager, ConnectionContext context, ConsumerInfo info, PendingMessageCursor cursor) throws InvalidSelectorException {
         super(broker,context, info);
         this.usageManager=usageManager;
         pending = cursor;
+        this.scheduler = broker.getScheduler();
     }
 
     public PrefetchSubscription(Broker broker,SystemUsage usageManager, ConnectionContext context, ConsumerInfo info) throws InvalidSelectorException {
@@ -230,6 +229,7 @@ public abstract class PrefetchSubscripti
                             context.getTransaction().addSynchronization(
                                     new Synchronization() {
 
+                                        @Override
                                         public void afterCommit()
                                                 throws Exception {
                                             synchronized(dispatchLock) {
@@ -239,6 +239,7 @@ public abstract class PrefetchSubscripti
                                             }
                                         }
 
+                                        @Override
                                         public void afterRollback() throws Exception {
                                             synchronized(dispatchLock) {
                                                 if (isSlave()) {
@@ -486,6 +487,7 @@ public abstract class PrefetchSubscripti
         return (dispatched.size() - prefetchExtension) >= (info.getPrefetchSize() * .9);
     }
 
+    @Override
     public int countBeforeFull() {
         return info.getPrefetchSize() + prefetchExtension - dispatched.size();
     }
@@ -510,6 +512,7 @@ public abstract class PrefetchSubscripti
         return enqueueCounter;
     }
 
+    @Override
     public boolean isRecoveryRequired() {
         return pending.isRecoveryRequired();
     }
@@ -526,13 +529,15 @@ public abstract class PrefetchSubscripti
         }
     }
 
-   public void add(ConnectionContext context, Destination destination) throws Exception {
+   @Override
+public void add(ConnectionContext context, Destination destination) throws Exception {
         synchronized(pendingLock) {
             super.add(context, destination);
             pending.add(context, destination);
         }
     }
 
+    @Override
     public List<MessageReference> remove(ConnectionContext context, Destination destination) throws Exception {
         List<MessageReference> rc = new ArrayList<MessageReference>();
         synchronized(pendingLock) {
@@ -546,7 +551,7 @@ public abstract class PrefetchSubscripti
             synchronized(dispatchLock) {
 	            for (MessageReference r : dispatched) {
 	                if( r.getRegionDestination() == destination) {
-	                	rc.add((QueueMessageReference)r);
+	                	rc.add(r);
 	                }
 	            }
                 destination.getDestinationStatistics().getDispatched().subtract(dispatched.size());

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java Thu May 20 12:02:10 2010
@@ -125,7 +125,7 @@ public class Queue extends BaseDestinati
     };
     
     private final Object iteratingMutex = new Object() {};
-    private static final Scheduler scheduler = Scheduler.getInstance();
+    private final Scheduler scheduler;
     
     class TimeoutMessage implements Delayed {
 
@@ -203,6 +203,7 @@ public class Queue extends BaseDestinati
         super(brokerService, store, destination, parentStats);
         this.taskFactory = taskFactory;
         this.dispatchSelector = new QueueDispatchSelector(destination);
+        this.scheduler = brokerService.getBroker().getScheduler();
     }
 
     public List<Subscription> getConsumers() {

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java Thu May 20 12:02:10 2010
@@ -26,6 +26,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ThreadPoolExecutor;
 import javax.jms.InvalidClientIDException;
 import javax.jms.JMSException;
 import org.apache.activemq.broker.Broker;
@@ -57,6 +58,7 @@ import org.apache.activemq.command.Respo
 import org.apache.activemq.command.TransactionId;
 import org.apache.activemq.state.ConnectionState;
 import org.apache.activemq.store.kahadb.plist.PListStore;
+import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.thread.TaskRunnerFactory;
 import org.apache.activemq.usage.SystemUsage;
 import org.apache.activemq.util.BrokerSupport;
@@ -98,10 +100,14 @@ public class RegionBroker extends EmptyB
     private final Map<String, ConnectionContext> clientIdSet = new HashMap<String, ConnectionContext>();
     private final DestinationInterceptor destinationInterceptor;
     private ConnectionContext adminConnectionContext;
+    private final Scheduler scheduler;
+    private final ThreadPoolExecutor executor;
 
     public RegionBroker(BrokerService brokerService, TaskRunnerFactory taskRunnerFactory, SystemUsage memoryManager, DestinationFactory destinationFactory,
-                        DestinationInterceptor destinationInterceptor) throws IOException {
+                        DestinationInterceptor destinationInterceptor,Scheduler scheduler,ThreadPoolExecutor executor) throws IOException {
         this.brokerService = brokerService;
+        this.executor=executor;
+        this.scheduler = scheduler;
         if (destinationFactory == null) {
             throw new IllegalArgumentException("null destinationFactory");
         }
@@ -810,6 +816,16 @@ public class RegionBroker extends EmptyB
         }
     }
     
+    
+    @Override
+    public Scheduler getScheduler() {
+        return this.scheduler;
+    }
+    
+    public ThreadPoolExecutor getExecutor() {
+        return this.executor;
+    }
+    
     @Override
     public void processConsumerControl(ConsumerBrokerExchange consumerExchange, ConsumerControl control) {
         ActiveMQDestination destination = control.getDestination();

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java Thu May 20 12:02:10 2010
@@ -231,7 +231,7 @@ public class TopicRegion extends Abstrac
         if (broker.getDestinationPolicy() != null) {
             PolicyEntry entry = broker.getDestinationPolicy().getEntryFor(destination);
             if (entry != null) {
-                entry.configure(topic);
+                entry.configure(broker,topic);
             }
         }
     }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/AbortSlowConsumerStrategy.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/AbortSlowConsumerStrategy.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/AbortSlowConsumerStrategy.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/AbortSlowConsumerStrategy.java Thu May 20 12:02:10 2010
@@ -5,7 +5,6 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
-
 import org.apache.activemq.broker.Connection;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.Subscription;
@@ -24,15 +23,19 @@ public class AbortSlowConsumerStrategy i
     
     private static final Log LOG = LogFactory.getLog(AbortSlowConsumerStrategy.class);
 
-    private static final Scheduler scheduler = Scheduler.getInstance();
-    private AtomicBoolean taskStarted = new AtomicBoolean(false);
-    private Map<Subscription, SlowConsumerEntry> slowConsumers = new ConcurrentHashMap<Subscription, SlowConsumerEntry>();
+    private Scheduler scheduler;
+    private final AtomicBoolean taskStarted = new AtomicBoolean(false);
+    private final Map<Subscription, SlowConsumerEntry> slowConsumers = new ConcurrentHashMap<Subscription, SlowConsumerEntry>();
 
     private long maxSlowCount = -1;
     private long maxSlowDuration = 30*1000;
     private long checkPeriod = 30*1000;
     private boolean abortConnection = false;
 
+   public void setScheduler(Scheduler s) {
+       this.scheduler=s;
+   }
+   
     public void slowConsumer(ConnectionContext context, Subscription subs) {
         if (maxSlowCount < 0 && maxSlowDuration < 0) {
             // nothing to do

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedCountSubscriptionRecoveryPolicy.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedCountSubscriptionRecoveryPolicy.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedCountSubscriptionRecoveryPolicy.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedCountSubscriptionRecoveryPolicy.java Thu May 20 12:02:10 2010
@@ -18,7 +18,7 @@ package org.apache.activemq.broker.regio
 
 import java.util.ArrayList;
 import java.util.List;
-
+import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.MessageReference;
 import org.apache.activemq.broker.region.SubscriptionRecovery;
@@ -118,4 +118,7 @@ public class FixedCountSubscriptionRecov
         return result.toArray(new Message[result.size()]);
     }
 
+    public void setBroker(Broker broker) {        
+    }
+
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedSizedSubscriptionRecoveryPolicy.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedSizedSubscriptionRecoveryPolicy.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedSizedSubscriptionRecoveryPolicy.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedSizedSubscriptionRecoveryPolicy.java Thu May 20 12:02:10 2010
@@ -18,6 +18,7 @@ package org.apache.activemq.broker.regio
 
 import java.util.Iterator;
 import java.util.List;
+import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.MessageReference;
 import org.apache.activemq.broker.region.SubscriptionRecovery;
@@ -109,6 +110,9 @@ public class FixedSizedSubscriptionRecov
     public Message[] browse(ActiveMQDestination destination) throws Exception {
         return buffer.browse(destination);
     }
+    
+    public void setBroker(Broker broker) {        
+    }
 
     // Implementation methods
 

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/LastImageSubscriptionRecoveryPolicy.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/LastImageSubscriptionRecoveryPolicy.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/LastImageSubscriptionRecoveryPolicy.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/LastImageSubscriptionRecoveryPolicy.java Thu May 20 12:02:10 2010
@@ -18,6 +18,7 @@ package org.apache.activemq.broker.regio
 
 import java.util.ArrayList;
 import java.util.List;
+import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.MessageReference;
 import org.apache.activemq.broker.region.SubscriptionRecovery;
@@ -68,5 +69,8 @@ public class LastImageSubscriptionRecove
     public SubscriptionRecoveryPolicy copy() {
         return new LastImageSubscriptionRecoveryPolicy();
     }
+    
+    public void setBroker(Broker broker) {        
+    }
 
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/NoSubscriptionRecoveryPolicy.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/NoSubscriptionRecoveryPolicy.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/NoSubscriptionRecoveryPolicy.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/NoSubscriptionRecoveryPolicy.java Thu May 20 12:02:10 2010
@@ -16,6 +16,7 @@
  */
 package org.apache.activemq.broker.region.policy;
 
+import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.MessageReference;
 import org.apache.activemq.broker.region.SubscriptionRecovery;
@@ -52,5 +53,8 @@ public class NoSubscriptionRecoveryPolic
     public Message[] browse(ActiveMQDestination dest) throws Exception {
         return new Message[0];
     }
+    
+    public void setBroker(Broker broker) {        
+    }
 
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java Thu May 20 12:02:10 2010
@@ -90,7 +90,7 @@ public class PolicyEntry extends Destina
     
    
     public void configure(Broker broker,Queue queue) {
-        baseConfiguration(queue);
+        baseConfiguration(broker,queue);
         if (dispatchPolicy != null) {
             queue.setDispatchPolicy(dispatchPolicy);
         }
@@ -112,14 +112,16 @@ public class PolicyEntry extends Destina
         queue.setConsumersBeforeDispatchStarts(getConsumersBeforeDispatchStarts());
     }
 
-    public void configure(Topic topic) {
-        baseConfiguration(topic);
+    public void configure(Broker broker,Topic topic) {
+        baseConfiguration(broker,topic);
         if (dispatchPolicy != null) {
             topic.setDispatchPolicy(dispatchPolicy);
         }
         topic.setDeadLetterStrategy(getDeadLetterStrategy());
         if (subscriptionRecoveryPolicy != null) {
-            topic.setSubscriptionRecoveryPolicy(subscriptionRecoveryPolicy.copy());
+            SubscriptionRecoveryPolicy srp = subscriptionRecoveryPolicy.copy();
+            srp.setBroker(broker);
+            topic.setSubscriptionRecoveryPolicy(srp);
         }
         if (memoryLimit > 0) {
             topic.getMemoryUsage().setLimit(memoryLimit);
@@ -127,7 +129,7 @@ public class PolicyEntry extends Destina
         topic.setLazyDispatch(isLazyDispatch());
     }
     
-    public void baseConfiguration(BaseDestination destination) {
+    public void baseConfiguration(Broker broker,BaseDestination destination) {
         destination.setProducerFlowControl(isProducerFlowControl());
         destination.setBlockedProducerWarningInterval(getBlockedProducerWarningInterval());
         destination.setEnableAudit(isEnableAudit());
@@ -148,7 +150,11 @@ public class PolicyEntry extends Destina
         destination.setMaxExpirePageSize(getMaxExpirePageSize());
         destination.setCursorMemoryHighWaterMark(getCursorMemoryHighWaterMark());
         destination.setStoreUsageHighWaterMark(getStoreUsageHighWaterMark());
-        destination.setSlowConsumerStrategy(getSlowConsumerStrategy());
+        SlowConsumerStrategy scs = getSlowConsumerStrategy();
+        if (scs != null) {
+            scs.setScheduler(broker.getScheduler());
+        }
+        destination.setSlowConsumerStrategy(scs);
     }
 
     public void configure(Broker broker, SystemUsage memoryManager, TopicSubscription subscription) {

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java Thu May 20 12:02:10 2010
@@ -17,12 +17,11 @@
 package org.apache.activemq.broker.region.policy;
 
 import java.util.concurrent.atomic.AtomicLong;
-
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageListener;
-
 import org.apache.activemq.ActiveMQMessageTransformation;
+import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.MessageReference;
@@ -50,9 +49,9 @@ public class QueryBasedSubscriptionRecov
     private static final Log LOG = LogFactory.getLog(QueryBasedSubscriptionRecoveryPolicy.class);
 
     private MessageQuery query;
-    private AtomicLong messageSequence = new AtomicLong(0);
-    private IdGenerator idGenerator = new IdGenerator();
-    private ProducerId producerId = createProducerId();
+    private final AtomicLong messageSequence = new AtomicLong(0);
+    private final IdGenerator idGenerator = new IdGenerator();
+    private final ProducerId producerId = createProducerId();
 
     public SubscriptionRecoveryPolicy copy() {
         QueryBasedSubscriptionRecoveryPolicy rc = new QueryBasedSubscriptionRecoveryPolicy();
@@ -99,6 +98,9 @@ public class QueryBasedSubscriptionRecov
     public org.apache.activemq.command.Message[] browse(ActiveMQDestination dest) throws Exception {
         return new org.apache.activemq.command.Message[0];
     }
+    
+    public void setBroker(Broker broker) {        
+    }
 
     protected void dispatchInitialMessage(Message message, Destination regionDestination, ConnectionContext context, SubscriptionRecovery sub) {
         try {

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/SlowConsumerStrategy.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/SlowConsumerStrategy.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/SlowConsumerStrategy.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/SlowConsumerStrategy.java Thu May 20 12:02:10 2010
@@ -2,6 +2,7 @@ package org.apache.activemq.broker.regio
 
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.Subscription;
+import org.apache.activemq.thread.Scheduler;
 
 /*
  * a strategy for dealing with slow consumers
@@ -9,5 +10,6 @@ import org.apache.activemq.broker.region
 public interface SlowConsumerStrategy {
 
     void slowConsumer(ConnectionContext context, Subscription subs);
+    void setScheduler(Scheduler scheduler);
 
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/SubscriptionRecoveryPolicy.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/SubscriptionRecoveryPolicy.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/SubscriptionRecoveryPolicy.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/SubscriptionRecoveryPolicy.java Thu May 20 12:02:10 2010
@@ -18,6 +18,7 @@ package org.apache.activemq.broker.regio
 
 
 import org.apache.activemq.Service;
+import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.MessageReference;
 import org.apache.activemq.broker.region.SubscriptionRecovery;
@@ -69,4 +70,6 @@ public interface SubscriptionRecoveryPol
      * @return the copy
      */
     SubscriptionRecoveryPolicy copy();
+    
+    void setBroker(Broker broker);
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/TimedSubscriptionRecoveryPolicy.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/TimedSubscriptionRecoveryPolicy.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/TimedSubscriptionRecoveryPolicy.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/TimedSubscriptionRecoveryPolicy.java Thu May 20 12:02:10 2010
@@ -21,6 +21,7 @@ import java.util.Collections;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
+import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.MessageReference;
 import org.apache.activemq.broker.region.SubscriptionRecovery;
@@ -28,7 +29,6 @@ import org.apache.activemq.broker.region
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.Message;
 import org.apache.activemq.filter.DestinationFilter;
-import org.apache.activemq.filter.MessageEvaluationContext;
 import org.apache.activemq.thread.Scheduler;
 
 /**
@@ -42,7 +42,7 @@ import org.apache.activemq.thread.Schedu
 public class TimedSubscriptionRecoveryPolicy implements SubscriptionRecoveryPolicy {
 
     private static final int GC_INTERVAL = 1000;
-    protected static final Scheduler scheduler = Scheduler.getInstance();
+    private Scheduler scheduler;
     
     // TODO: need to get a better synchronized linked list that has little
     // contention between enqueuing and dequeuing
@@ -89,6 +89,10 @@ public class TimedSubscriptionRecoveryPo
             }
         }
     }
+    
+    public void setBroker(Broker broker) {  
+        this.scheduler = broker.getScheduler();
+    }
 
     public void start() throws Exception {
         scheduler.executePeriodically(gcTask, GC_INTERVAL);
@@ -97,6 +101,7 @@ public class TimedSubscriptionRecoveryPo
     public void stop() throws Exception {
         scheduler.cancel(gcTask);
     }
+    
 
     public void gc() {
         lastGCRun = System.currentTimeMillis();

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java Thu May 20 12:02:10 2010
@@ -35,7 +35,6 @@ import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
-
 import org.apache.activemq.kaha.impl.async.DataFileAppender.WriteCommand;
 import org.apache.activemq.kaha.impl.async.DataFileAppender.WriteKey;
 import org.apache.activemq.thread.Scheduler;
@@ -75,7 +74,7 @@ public class AsyncDataManager {
     public static final int PREFERED_DIFF = 1024 * 512;
 
     private static final Log LOG = LogFactory.getLog(AsyncDataManager.class);
-    protected static Scheduler scheduler  = Scheduler.getInstance();
+    protected Scheduler scheduler;
 
     protected final Map<WriteKey, WriteCommand> inflightWrites = new ConcurrentHashMap<WriteKey, WriteCommand>();
 
@@ -193,7 +192,13 @@ public class AsyncDataManager {
                 cleanup();
             }
         };
-        scheduler.executePeriodically(cleanupTask, DEFAULT_CLEANUP_INTERVAL);
+        this.scheduler = new Scheduler("AsyncDataManager Scheduler");
+        try {
+            this.scheduler.start();
+        } catch (Exception e) {
+            throw new IOException(e);
+        }
+        this.scheduler.executePeriodically(cleanupTask, DEFAULT_CLEANUP_INTERVAL);
     }
 
     public void lock() throws IOException {
@@ -328,7 +333,12 @@ public class AsyncDataManager {
         if (!started) {
             return;
         }
-        scheduler.cancel(cleanupTask);
+        this.scheduler.cancel(cleanupTask);
+        try {
+            this.scheduler.stop();
+        } catch (Exception e) {
+            throw new IOException(e);
+        }
         accessorPool.close();
         storeState(false);
         appender.close();
@@ -376,7 +386,7 @@ public class AsyncDataManager {
     public synchronized void addInterestInFile(int file) throws IOException {
         if (file >= 0) {
             Integer key = Integer.valueOf(file);
-            DataFile dataFile = (DataFile)fileMap.get(key);
+            DataFile dataFile = fileMap.get(key);
             if (dataFile == null) {
                 throw new IOException("That data file does not exist");
             }
@@ -393,7 +403,7 @@ public class AsyncDataManager {
     public synchronized void removeInterestInFile(int file) throws IOException {
         if (file >= 0) {
             Integer key = Integer.valueOf(file);
-            DataFile dataFile = (DataFile)fileMap.get(key);
+            DataFile dataFile = fileMap.get(key);
             removeInterestInFile(dataFile);
         }
        
@@ -414,7 +424,7 @@ public class AsyncDataManager {
                 
         List<DataFile> purgeList = new ArrayList<DataFile>();
         for (Integer key : unUsed) {
-            DataFile dataFile = (DataFile)fileMap.get(key);
+            DataFile dataFile = fileMap.get(key);
             purgeList.add(dataFile);
         }
         for (DataFile dataFile : purgeList) {
@@ -432,7 +442,7 @@ public class AsyncDataManager {
         for (Integer key : unUsed) {
         	// Only add files less than the lastFile..
         	if( key.intValue() < lastFile.intValue() ) {
-                DataFile dataFile = (DataFile)fileMap.get(key);
+                DataFile dataFile = fileMap.get(key);
                 purgeList.add(dataFile);
         	}
         }
@@ -499,6 +509,7 @@ public class AsyncDataManager {
         this.maxFileLength = maxFileLength;
     }
 
+    @Override
     public String toString() {
         return "DataManager:(" + filePrefix + ")";
     }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java?rev=946600&r1=946599&r2=946600&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java Thu May 20 12:02:10 2010
@@ -30,7 +30,6 @@ import java.util.concurrent.CountDownLat
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
-
 import org.apache.activeio.journal.Journal;
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.BrokerServiceAware;
@@ -58,7 +57,6 @@ import org.apache.activemq.store.TopicMe
 import org.apache.activemq.store.TopicReferenceStore;
 import org.apache.activemq.store.TransactionStore;
 import org.apache.activemq.store.kahadaptor.KahaReferenceStoreAdapter;
-import org.apache.activemq.thread.DefaultThreadPools;
 import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.thread.Task;
 import org.apache.activemq.thread.TaskRunner;
@@ -85,7 +83,7 @@ import org.apache.commons.logging.LogFac
 public class AMQPersistenceAdapter implements PersistenceAdapter, UsageListener, BrokerServiceAware {
 
     private static final Log LOG = LogFactory.getLog(AMQPersistenceAdapter.class);
-    private static final Scheduler scheduler = Scheduler.getInstance();
+    private Scheduler scheduler;
     private final ConcurrentHashMap<ActiveMQQueue, AMQMessageStore> queues = new ConcurrentHashMap<ActiveMQQueue, AMQMessageStore>();
     private final ConcurrentHashMap<ActiveMQTopic, AMQTopicMessageStore> topics = new ConcurrentHashMap<ActiveMQTopic, AMQTopicMessageStore>();
     private static final String PROPERTY_PREFIX = "org.apache.activemq.store.amq";
@@ -99,7 +97,7 @@ public class AMQPersistenceAdapter imple
     private SystemUsage usageManager;
     private long checkpointInterval = 1000 * 20;
     private int maxCheckpointMessageAddSize = 1024 * 4;
-    private AMQTransactionStore transactionStore = new AMQTransactionStore(this);
+    private final AMQTransactionStore transactionStore = new AMQTransactionStore(this);
     private TaskRunner checkpointTask;
     private CountDownLatch nextCheckpointCountDownLatch = new CountDownLatch(1);
     private final AtomicBoolean started = new AtomicBoolean(false);
@@ -112,7 +110,7 @@ public class AMQPersistenceAdapter imple
     private File directory;
     private File directoryArchive;
     private BrokerService brokerService;
-    private AtomicLong storeSize = new AtomicLong();
+    private final AtomicLong storeSize = new AtomicLong();
     private boolean persistentIndex=true;
     private boolean useNio = true;
     private boolean archiveDataLogs=false;
@@ -124,8 +122,7 @@ public class AMQPersistenceAdapter imple
     private int indexMaxBinSize = HashIndex.MAXIMUM_CAPACITY;
     private int indexLoadFactor = HashIndex.DEFAULT_LOAD_FACTOR;
     private int maxReferenceFileLength=AMQPersistenceAdapterFactory.DEFAULT_MAX_REFERNCE_FILE_LENGTH;
-    private Map<AMQMessageStore,Map<Integer, AtomicInteger>> dataFilesInProgress = new ConcurrentHashMap<AMQMessageStore,Map<Integer, AtomicInteger>> ();
-    private String directoryPath = "";
+    private final Map<AMQMessageStore,Map<Integer, AtomicInteger>> dataFilesInProgress = new ConcurrentHashMap<AMQMessageStore,Map<Integer, AtomicInteger>> ();
     private RandomAccessFile lockFile;
     private FileLock lock;
     private boolean disableLocking = DISABLE_LOCKING;
@@ -134,6 +131,8 @@ public class AMQPersistenceAdapter imple
     private boolean lockAquired;
     private boolean recoverReferenceStore=true;
     private boolean forceRecoverReferenceStore=false;
+    private boolean useDedicatedTaskRunner=false;
+    private int journalThreadPriority = Thread.MAX_PRIORITY;
 
     public String getBrokerName() {
         return this.brokerName;
@@ -165,12 +164,19 @@ public class AMQPersistenceAdapter imple
             } else {
                 this.directory = new File(IOHelper.getDefaultDataDirectory(), IOHelper.toFileSystemSafeName(brokerName));
                 this.directory = new File(directory, "amqstore");
-                this.directoryPath=directory.getAbsolutePath();
+                directory.getAbsolutePath();
             }
         }
         if (this.directoryArchive == null) {
             this.directoryArchive = new File(this.directory,"archive");
         }
+        if (this.brokerService != null) {
+            this.taskRunnerFactory = this.brokerService.getTaskRunnerFactory();
+        }else {
+            this.scheduler = new Scheduler("AMQPersistenceAdapter Scheduler");
+        }
+        this.taskRunnerFactory= new TaskRunnerFactory("AMQPersistenceAdaptor Task", getJournalThreadPriority(),
+                true, 1000, isUseDedicatedTaskRunner());
         IOHelper.mkdirs(this.directory);
         lockFile = new RandomAccessFile(new File(directory, "lock"), "rw");
         lock();
@@ -192,10 +198,11 @@ public class AMQPersistenceAdapter imple
         referenceStoreAdapter.setBrokerName(getBrokerName());
         referenceStoreAdapter.setUsageManager(usageManager);
         referenceStoreAdapter.setMaxDataFileLength(getMaxReferenceFileLength());
-        if (taskRunnerFactory == null) {
-            taskRunnerFactory = createTaskRunnerFactory();
-        }
         
+        if (brokerService != null) {
+            this.scheduler = this.brokerService.getBroker().getScheduler();
+        }
+                
         if (failIfJournalIsLocked) {
             asyncDataManager.lock();
         } else {
@@ -389,7 +396,7 @@ public class AMQPersistenceAdapter imple
             Iterator<AMQMessageStore> queueIterator = queues.values().iterator();
             while (queueIterator.hasNext()) {
                 final AMQMessageStore ms = queueIterator.next();
-                Location mark = (Location)ms.getMark();
+                Location mark = ms.getMark();
                 if (mark != null && (newMark == null || mark.compareTo(newMark) > 0)) {
                     newMark = mark;
                 }
@@ -397,7 +404,7 @@ public class AMQPersistenceAdapter imple
             Iterator<AMQTopicMessageStore> topicIterator = topics.values().iterator();
             while (topicIterator.hasNext()) {
                 final AMQTopicMessageStore ms = topicIterator.next();
-                Location mark = (Location)ms.getMark();
+                Location mark = ms.getMark();
                 if (mark != null && (newMark == null || mark.compareTo(newMark) > 0)) {
                     newMark = mark;
                 }
@@ -726,6 +733,7 @@ public class AMQPersistenceAdapter imple
         deleteAllMessages = true;
     }
 
+    @Override
     public String toString() {
         return "AMQPersistenceAdapter(" + directory + ")";
     }
@@ -754,10 +762,6 @@ public class AMQPersistenceAdapter imple
         return adaptor;
     }
 
-    protected TaskRunnerFactory createTaskRunnerFactory() {
-        return DefaultThreadPools.getDefaultTaskRunnerFactory();
-    }
-
     // /////////////////////////////////////////////////////////////////
     // Property Accessors
     // /////////////////////////////////////////////////////////////////
@@ -991,6 +995,28 @@ public class AMQPersistenceAdapter imple
     public void setForceRecoverReferenceStore(boolean forceRecoverReferenceStore) {
         this.forceRecoverReferenceStore = forceRecoverReferenceStore;
     }
+    
+    public boolean isUseDedicatedTaskRunner() {
+        return useDedicatedTaskRunner;
+    }
+    
+    public void setUseDedicatedTaskRunner(boolean useDedicatedTaskRunner) {
+        this.useDedicatedTaskRunner = useDedicatedTaskRunner;
+    }
+    
+    /**
+     * @return the journalThreadPriority
+     */
+    public int getJournalThreadPriority() {
+        return this.journalThreadPriority;
+    }
+
+    /**
+     * @param journalThreadPriority the journalThreadPriority to set
+     */
+    public void setJournalThreadPriority(int journalThreadPriority) {
+        this.journalThreadPriority = journalThreadPriority;
+    }
 
 	
 	protected void addInProgressDataFile(AMQMessageStore store,int dataFileId) {



Mime
View raw message