[jboss-cvs] JBossAS SVN: r61445 - branches/Branch_4_0/messaging/src/main/org/jboss/mq/server.

[jboss-cvs-commits at lists.jboss.org]

Author: adrian at jboss.org
Date: 2007-03-19 12:47:26 -0400 (Mon, 19 Mar 2007)
New Revision: 61445

Modified:
   branches/Branch_4_0/messaging/src/main/org/jboss/mq/server/BasicQueue.java
   branches/Branch_4_0/messaging/src/main/org/jboss/mq/server/JMSTopic.java
Log:
[JBAS-4204] - Don't allow durable topic subscriptions to be unsubscribed if there are active subscribers or unacknowledged messages

Modified: branches/Branch_4_0/messaging/src/main/org/jboss/mq/server/BasicQueue.java
===================================================================
--- branches/Branch_4_0/messaging/src/main/org/jboss/mq/server/BasicQueue.java	2007-03-19 15:57:39 UTC (rev 61444)
+++ branches/Branch_4_0/messaging/src/main/org/jboss/mq/server/BasicQueue.java	2007-03-19 16:47:26 UTC (rev 61445)
@@ -194,7 +194,8 @@
    {
       synchronized (receivers)
       {
-         return subscribers.size() > 0;
+         // In use if we have subscribers or there are unacknowledged messages
+         return (subscribers.isEmpty() == false || getInProcessMessageCount() > 0);
       }
    }
 
@@ -342,7 +343,7 @@
          return unacknowledgedMessages.size();
       }
    }
-
+   
    /**
     * Add a message to the queue
     *

Modified: branches/Branch_4_0/messaging/src/main/org/jboss/mq/server/JMSTopic.java
===================================================================
--- branches/Branch_4_0/messaging/src/main/org/jboss/mq/server/JMSTopic.java	2007-03-19 15:57:39 UTC (rev 61444)
+++ branches/Branch_4_0/messaging/src/main/org/jboss/mq/server/JMSTopic.java	2007-03-19 16:47:26 UTC (rev 61445)
@@ -334,7 +334,20 @@
    //called by state manager when a durable sub is deleted
    public void destroyDurableSubscription(DurableSubscriptionID id) throws JMSException
    {
-      BasicQueue queue = (BasicQueue) durQueues.remove(id);
+      boolean inUse = false;
+      BasicQueue queue = null;
+      synchronized (durQueues)
+      {
+         queue = (BasicQueue) durQueues.get(id);
+         if (queue != null && queue.isInUse())
+            inUse = true;
+         else
+            durQueues.remove(id);
+      }
+      if (queue == null)
+         throw new IllegalStateException("Unable to find durable subscription: " + id);
+      if (inUse)
+         throw new IllegalStateException("The durable subscription is in use: " + id);
       queue.removeAllMessages();
    }