[jboss-cvs] JBoss Messaging SVN: r3109 - in trunk: tests/src/org/jboss/test/messaging/jms/clustering and 1 other directory.

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

Author: timfox
Date: 2007-09-17 13:27:13 -0400 (Mon, 17 Sep 2007)
New Revision: 3109

Modified:
   trunk/src/main/org/jboss/jms/server/endpoint/ServerConsumerEndpoint.java
   trunk/tests/src/org/jboss/test/messaging/jms/clustering/DistributedTopicTest.java
Log:
http://jira.jboss.com/jira/browse/JBMESSAGING-1069


Modified: trunk/src/main/org/jboss/jms/server/endpoint/ServerConsumerEndpoint.java
===================================================================
--- trunk/src/main/org/jboss/jms/server/endpoint/ServerConsumerEndpoint.java	2007-09-17 17:22:09 UTC (rev 3108)
+++ trunk/src/main/org/jboss/jms/server/endpoint/ServerConsumerEndpoint.java	2007-09-17 17:27:13 UTC (rev 3109)
@@ -301,7 +301,7 @@
 
             if (trace) { log.trace("message connection id: " + conId + " current connection connection id: " + sessionEndpoint.getConnectionEndpoint().getConnectionID()); }
 
-            accept = !conId.equals(sessionEndpoint.getConnectionEndpoint().getConnectionID());
+            accept = !sessionEndpoint.getConnectionEndpoint().getConnectionID().equals(conId);
 
             if (trace) { log.trace("accepting? " + accept); }
          }

Modified: trunk/tests/src/org/jboss/test/messaging/jms/clustering/DistributedTopicTest.java
===================================================================
--- trunk/tests/src/org/jboss/test/messaging/jms/clustering/DistributedTopicTest.java	2007-09-17 17:22:09 UTC (rev 3108)
+++ trunk/tests/src/org/jboss/test/messaging/jms/clustering/DistributedTopicTest.java	2007-09-17 17:27:13 UTC (rev 3109)
@@ -72,7 +72,17 @@
    {
       clusteredTopicNonDurable(true);
    }
+   
+   public void testClusteredTopicNonDurableNoLocalNonPersistent() throws Exception
+   {
+      clusteredTopicNoLocal(false);
+   }
 
+   public void testClusteredTopicNonDurableNoLocalPersistent() throws Exception
+   {
+   	clusteredTopicNoLocal(true);
+   }
+
    public void testClusteredTopicNonDurableWithSelectorsNonPersistent() throws Exception
    {
       clusteredTopicNonDurableWithSelectors(false);
@@ -417,7 +427,101 @@
          }
       }
    }
+   
+   
+   /* Test with noLocal set to true */
+   private void clusteredTopicNoLocal(boolean persistent) throws Exception
+   {
+      Connection conn0 = null;
+      Connection conn1 = null;
+      Connection conn2 = null;
+      try
+      {
+         //This will create 3 different connection on 3 different nodes, since
+         //the cf is clustered
+         conn0 = this.createConnectionOnServer(cf, 0);
+         conn1 = this.createConnectionOnServer(cf, 1);
+         conn2 = this.createConnectionOnServer(cf, 2);
+         
+         checkConnectionsDifferentServers(new Connection[] {conn0, conn1, conn2});
 
+         Session sess0 = conn0.createSession(false, Session.AUTO_ACKNOWLEDGE);
+         Session sess1 = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE);
+         Session sess2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+         MessageConsumer cons0 = sess0.createConsumer(topic[0], null, true);
+         MessageConsumer cons1 = sess1.createConsumer(topic[1], null, true);
+         MessageConsumer cons2 = sess2.createConsumer(topic[2], null, true);        
+
+         conn0.start();
+         conn1.start();
+         conn2.start();
+
+         // Send at node 0
+
+         MessageProducer prod = sess0.createProducer(topic[0]);
+
+         prod.setDeliveryMode(persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
+
+         final int NUM_MESSAGES = 100;
+
+         for (int i = 0; i < NUM_MESSAGES; i++)
+         {
+            TextMessage tm = sess0.createTextMessage("message" + i);
+
+            prod.send(tm);
+         }
+
+         Message msg = cons0.receive(3000);
+         
+         assertNull(msg);
+
+         for (int i = 0; i < NUM_MESSAGES; i++)
+         {
+            TextMessage tm = (TextMessage)cons1.receive(3000);
+
+            assertNotNull(tm);
+            
+ 
+            assertEquals("message" + i, tm.getText());
+         }
+         
+         msg = cons1.receive(3000);
+         
+         assertNull(msg);
+
+         for (int i = 0; i < NUM_MESSAGES; i++)
+         {
+            TextMessage tm = (TextMessage)cons2.receive(3000);
+
+            assertNotNull(tm);
+            
+            assertEquals("message" + i, tm.getText());
+         }
+         
+         msg = cons2.receive(3000);
+         
+         assertNull(msg);         
+      }
+      finally
+      {
+         if (conn0 != null)
+         {
+            conn0.close();
+         }
+
+         if (conn1 != null)
+         {
+            conn1.close();
+         }
+
+         if (conn2 != null)
+         {
+            conn2.close();
+         }
+      }
+   }
+
    /*
     * Create non durable subscriptions on all nodes of the cluster.
     * Include some with selectors