[jboss-cvs] JBoss Messaging SVN: r1435 - in trunk: src/main/org/jboss/jms/server/endpoint tests/src/org/jboss/test/messaging/jms

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

Author: timfox
Date: 2006-10-05 13:39:25 -0400 (Thu, 05 Oct 2006)
New Revision: 1435

Modified:
   trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
   trunk/tests/src/org/jboss/test/messaging/jms/ManualClusteringTest.java
Log:
http://jira.jboss.com/jira/browse/JBMESSAGING-595


Modified: trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
===================================================================
--- trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java	2006-10-05 16:10:31 UTC (rev 1434)
+++ trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java	2006-10-05 17:39:25 UTC (rev 1435)
@@ -320,8 +320,17 @@
    
                      // Unbind the durable subscription
                      
-                     topicPostOffice.unbindQueue(name);
-   
+                     if (mDest.isClustered() && !topicPostOffice.isLocal())
+                     {
+                        ClusteredPostOffice cpo = (ClusteredPostOffice)topicPostOffice;
+                        
+                        cpo.unbindClusteredQueue(name);
+                     }
+                     else
+                     {         
+                        topicPostOffice.unbindQueue(name);
+                     }
+                        
                      // create a fresh new subscription
                      
                      QueuedExecutor executor = (QueuedExecutor)pool.get();
@@ -761,9 +770,21 @@
                                             subscriptionName + " since it has active subscribers");
          }
          
+         //Look up the topic
+         ManagedDestination mDest = dm.getDestination(binding.getCondition(), false);
+         
          //Unbind it
-  
-         topicPostOffice.unbindQueue(queueName);
+    
+         if (mDest.isClustered() && !topicPostOffice.isLocal())
+         {
+            ClusteredPostOffice cpo = (ClusteredPostOffice)topicPostOffice;
+            
+            cpo.unbindClusteredQueue(queueName);
+         }
+         else
+         {         
+            topicPostOffice.unbindQueue(queueName);
+         }
       }
       catch (Throwable t)
       {

Modified: trunk/tests/src/org/jboss/test/messaging/jms/ManualClusteringTest.java
===================================================================
--- trunk/tests/src/org/jboss/test/messaging/jms/ManualClusteringTest.java	2006-10-05 16:10:31 UTC (rev 1434)
+++ trunk/tests/src/org/jboss/test/messaging/jms/ManualClusteringTest.java	2006-10-05 17:39:25 UTC (rev 1435)
@@ -88,7 +88,7 @@
       Properties props1 = new Properties();
       
       props1.put(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory");
-      props1.put(Context.PROVIDER_URL, "jnp://localhost:1099");
+      props1.put(Context.PROVIDER_URL, "jnp://localhost:1199");
       props1.put(Context.URL_PKG_PREFIXES, "org.jnp.interfaces");
       
       ic1 = new InitialContext(props1);
@@ -96,7 +96,7 @@
       Properties props2 = new Properties();
       
       props2.put(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory");
-      props2.put(Context.PROVIDER_URL, "jnp://localhost:1199");
+      props2.put(Context.PROVIDER_URL, "jnp://localhost:1299");
       props2.put(Context.URL_PKG_PREFIXES, "org.jnp.interfaces");
       
       ic2 = new InitialContext(props2);
@@ -109,15 +109,15 @@
 //      
 //      ic3 = new InitialContext(props3);
       
-      queue1 = (Queue)ic1.lookup("queue/ClusteredQueue1");
+      queue1 = (Queue)ic1.lookup("queue/testDistributedQueue");
       
-      queue2 = (Queue)ic2.lookup("queue/ClusteredQueue1");
+      queue2 = (Queue)ic2.lookup("queue/testDistributedQueue");
       
       //queue3 = (Queue)ic3.lookup("queue/ClusteredQueue1");
             
-      topic1 = (Topic)ic1.lookup("topic/ClusteredTopic1");
+      topic1 = (Topic)ic1.lookup("topic/testDistributedTopic");
       
-      topic2 = (Topic)ic2.lookup("topic/ClusteredTopic1");
+      topic2 = (Topic)ic2.lookup("topic/testDistributedTopic");
       
       //topic3 = (Topic)ic3.lookup("topic/ClusteredTopic1");
       
@@ -602,17 +602,7 @@
          Session sess1 = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE);
          
          Session sess2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
-         
-         try
-         {
-            sess1.unsubscribe("sub1");
-                  
-            sess2.unsubscribe("sub1");
-         }
-         catch (Exception ignore)
-         {            
-         }
-                 
+               
          MessageConsumer durable1 = sess1.createDurableSubscriber(topic1, "sub1");
          
          MessageConsumer durable2 = sess2.createDurableSubscriber(topic2, "sub1");
@@ -655,7 +645,10 @@
          
          assertNull(m);
          
+         durable1.close();
          
+         durable2.close();
+         
          sess1.unsubscribe("sub1");
          
          sess2.unsubscribe("sub1");
@@ -787,14 +780,6 @@
          
          Session sess2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
          
-         try
-         {
-            sess2.unsubscribe("sub1");
-         }
-         catch (Exception ignore)
-         {            
-         }
-         
          MessageConsumer durable2 = sess2.createDurableSubscriber(topic2, "sub1");
          
          conn1.start();
@@ -830,6 +815,8 @@
             
             assertEquals("message" + i, tm.getText());
          }
+         
+         durable2.close();
           
          sess2.unsubscribe("sub1");
          
@@ -867,15 +854,7 @@
          Session sess1 = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE);
          
          Session sess2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
-         
-         try
-         { 
-            sess2.unsubscribe("sub1");
-         }
-         catch (Exception ignore)
-         {            
-         }
-         
+          
          MessageConsumer durable2 = sess2.createDurableSubscriber(topic2, "sub1");
          
          conn1.start();
@@ -912,6 +891,8 @@
             assertEquals("message" + i, tm.getText());
          }
          
+         durable2.close();
+         
          sess2.unsubscribe("sub1");
          
       }