svn commit: r547601 - /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java

[ra...@apache.org]

Author: rajdavies
Date: Fri Jun 15 03:24:23 2007
New Revision: 547601

URL: http://svn.apache.org/viewvc?view=rev&rev=547601
Log:
synchronized activate () and add() methods

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java?view=diff&rev=547601&r1=547600&r2=547601
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java Fri Jun 15 03:24:23 2007
@@ -74,8 +74,8 @@
         dispatchMatched();
     }
 
-    public void activate(UsageManager memoryManager,ConnectionContext context,ConsumerInfo info) throws Exception{
-        log.debug("Deactivating "+this);
+    public synchronized void activate(UsageManager memoryManager,ConnectionContext context,ConsumerInfo info) throws Exception{
+        log.debug("Activating "+this);
         if(!active){
             this.active=true;
             this.context=context;
@@ -85,11 +85,10 @@
                     Topic topic=(Topic)iter.next();
                     topic.activate(context,this);
                 }
-            }
-            synchronized(pending){
-                pending.setUsageManager(memoryManager);
-                pending.start();
-            }
+            }           
+            pending.setUsageManager(memoryManager);
+            pending.start();
+           
             // If nothing was in the persistent store, then try to use the recovery policy.
             if(pending.isEmpty()){
                 for(Iterator iter=destinations.values().iterator();iter.hasNext();){
@@ -160,7 +159,7 @@
         return md;
     }
 
-    public void add(MessageReference node) throws Exception{
+    public synchronized void add(MessageReference node) throws Exception{
         if(!active&&!keepDurableSubsActive){
             return;
         }
@@ -189,7 +188,7 @@
         return active;
     }
 
-    protected void acknowledge(ConnectionContext context,MessageAck ack,MessageReference node) throws IOException{
+    protected synchronized void acknowledge(ConnectionContext context,MessageAck ack,MessageReference node) throws IOException{
         node.getRegionDestination().acknowledge(context,this,ack,node);
         redeliveredMessages.remove(node.getMessageId());
         node.decrementReferenceCount();