https://issues.apache.org/jira/browse/AMQ-4656 - first stab at improving keepDurableSubsActive feature, by not stoping/starting cursor on subscription (de)activating

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1511333 13f79535-47bb-0310-9956-ffa450edef68

activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java

@@ -162,6 +162,7 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
             }
 
             synchronized (pendingLock) {
+                if (!((StoreDurableSubscriberCursor) pending).isStarted() || !keepDurableSubsActive) {
                     pending.setSystemUsage(memoryManager);
                     pending.setMemoryUsageHighWaterMark(getCursorMemoryHighWaterMark());
                     pending.setMaxAuditDepth(getMaxAuditDepth());
@@ -176,6 +177,7 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
                         }
                     }
                 }
+            }
             this.active.set(true);
             this.offlineTimestamp.set(-1);
             dispatchPending();
@@ -195,7 +197,9 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
         List<MessageReference> savedDispateched = null;
 
         synchronized (pendingLock) {
+            if (!keepDurableSubsActive) {
                 pending.stop();
+            }
 
             synchronized (dispatchLock) {
                 for (Destination destination : durableDestinations.values()) {

activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursor.java

@@ -309,7 +309,7 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
         }
     }
     
-    protected synchronized boolean isStarted() {
+    public synchronized boolean isStarted() {
         return started;
     }
     

activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java

@@ -211,7 +211,7 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i
     }
 
     
-    public final synchronized void addMessageFirst(MessageReference node) throws Exception {
+    public synchronized void addMessageFirst(MessageReference node) throws Exception {
         setCacheEnabled(false);
         size++;
     }

activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/TopicStorePrefetch.java

@@ -16,6 +16,7 @@
  */
 package org.apache.activemq.broker.region.cursors;
 
+import org.apache.activemq.broker.region.MessageReference;
 import org.apache.activemq.broker.region.Subscription;
 import org.apache.activemq.broker.region.Topic;
 import org.apache.activemq.command.Message;
@@ -60,6 +61,11 @@ class TopicStorePrefetch extends AbstractStoreCursor {
         throw new RuntimeException("Not supported");
     }
 
+    public synchronized void addMessageFirst(MessageReference node) throws Exception {
+        batchList.addMessageFirst(node);
+        size++;
+    }
+    
         
     @Override
     public synchronized boolean recoverMessage(Message message, boolean cached) throws Exception {

activemq-unit-tests/src/test/java/org/apache/activemq/bugs/DurableConsumerTest.java

@@ -213,6 +213,7 @@ public class DurableConsumerTest extends CombinationTestSupport{
     public void testConcurrentDurableConsumer() throws Exception{
         
         broker.start();
+        broker.waitUntilStarted();
         
         factory = createConnectionFactory();
         final String topicName = getName();
@@ -408,6 +409,7 @@ public class DurableConsumerTest extends CombinationTestSupport{
         super.tearDown();
         if (broker != null) {
             broker.stop();
+            broker.waitUntilStopped();
             broker = null;
         }
     }

activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubSelectorDelayWithRestartTest.java

@@ -225,8 +225,10 @@ public class DurableSubSelectorDelayWithRestartTest {
                 } while (true);
 
             } finally {
+                try {
                     sess.close();
                     con.close();
+                } catch (Exception e) {}
 
                 LOG.info(toString() + " OFFLINE.");
             }

activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java

@@ -130,6 +130,7 @@ public class DurableSubscriptionOfflineTest extends org.apache.activemq.TestSupp
             ((KahaDBPersistenceAdapter)broker.getPersistenceAdapter()).setJournalMaxFileLength(journalMaxFileLength);
         }
         broker.start();
+        broker.waitUntilStarted();
     }
 
     private void destroyBroker() throws Exception {
@@ -1120,6 +1121,7 @@ public class DurableSubscriptionOfflineTest extends org.apache.activemq.TestSupp
             LOG.info("Iteration: " + i);
             doTestOrderOnActivateDeactivate();
             broker.stop();
+            broker.waitUntilStopped();
             createBroker(true /*deleteAllMessages*/);
         }
     }