Merged revisions 658166 via svnmerge from

https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.x

................
  r658166 | ritchiem | 2008-05-20 09:37:33 +0100 (Tue, 20 May 2008) | 9 lines
  
  Merged revisions 657859 via svnmerge from 
  https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid
  
  ........
    r657859 | ritchiem | 2008-05-19 17:54:06 +0100 (Mon, 19 May 2008) | 1 line
    
    QPID-1066 : Removed isInfo wrapping. Added test that is missing from trunk from M2.x QueueDepthSelectorTest. 
  ........
................


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1.x@658175 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 42 insertions, 22 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java b/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
index cf607548f8..0e8cff0f2a 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
@@ -751,13 +751,7 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager
             }
             else if (messageQueue == sub.getPreDeliveryQueue() && !sub.isBrowser())
             {
-                if (_log.isInfoEnabled())
-                {
-                    //fixme - we should do the clean up as the message remains on the _message queue
-                    // this is resulting in the next consumer receiving the message and then attempting to purge it
-                    //
-                    cleanMainQueue(sub);
-                }
+                cleanMainQueue(sub);
             }
 
         }
diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/QueueDepthWithSelectorTest.java b/java/systests/src/main/java/org/apache/qpid/server/queue/QueueDepthWithSelectorTest.java
index 7bbfdb5543..4f9129e95f 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/queue/QueueDepthWithSelectorTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/queue/QueueDepthWithSelectorTest.java
@@ -22,6 +22,9 @@
 package org.apache.qpid.server.queue;
 
 import java.util.Hashtable;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQSession;
 
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
@@ -34,18 +37,13 @@ import javax.jms.Session;
 import javax.naming.Context;
 import javax.naming.NamingException;
 import javax.naming.spi.InitialContextFactory;
+import java.util.Hashtable;
 
 import junit.framework.TestCase;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
-import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.registry.IApplicationRegistry;
-import org.apache.qpid.server.virtualhost.VirtualHost;
-
-
 /**
  * Test Case to ensure that messages are correctly returned.
  * This includes checking:
@@ -107,7 +105,7 @@ public class QueueDepthWithSelectorTest extends TestCase
         {
             TransportConnection.killAllVMBrokers();
         }
-    } 
+    }
 
     public void test() throws Exception
     {
@@ -153,14 +151,42 @@ public class QueueDepthWithSelectorTest extends TestCase
 
     private void verifyBrokerState()
     {
-        IApplicationRegistry registry = ApplicationRegistry.getInstance();
-
-        VirtualHost testVhost = registry.getVirtualHostRegistry().getVirtualHost(VHOST);
-        assertNotNull("Unable to get test Vhost", testVhost);
-        assertNotNull("Unable to get test queue registry", testVhost.getQueueRegistry());
-        AMQQueue q = testVhost.getQueueRegistry().getQueue(new AMQShortString(QUEUE));
-        assertNotNull("Unable to get test queue", q);
-        assertEquals("Queue count too big", 0, q.getMessageCount());
+        try
+        {
+            _clientConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
+
+            _clientSession = _clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        }
+        catch (Exception e)
+        {
+            fail(e.getMessage());
+        }
+
+        try
+        {
+            long queueDepth = ((AMQSession) _clientSession).getQueueDepth((AMQDestination) _context.lookup("queue"));
+            assertEquals("Session reports Queue depth not as expected", 0, queueDepth);
+        }
+        catch (NamingException e)
+        {
+            fail(e.getMessage());
+        }
+        catch (AMQException e)
+        {
+            fail(e.getMessage());
+        }
+        finally
+        {
+            try
+            {
+                _clientConnection.close();
+            }
+            catch (JMSException e)
+            {
+                fail(e.getMessage());
+            }
+        }
+
     }
 
     private void verifyAllMessagesRecevied() throws JMSException