QPID-854 : Changes to the client to make the dispatcher responsible for closing the queue browser when all the messages have been processed.

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

2 files changed, 178 insertions, 35 deletions

diff --git a/java/systests/src/main/java/org/apache/qpid/test/client/CancelTest.java b/java/systests/src/main/java/org/apache/qpid/test/client/CancelTest.java
new file mode 100644
index 0000000000..b07778cdaa
--- /dev/null
+++ b/java/systests/src/main/java/org/apache/qpid/test/client/CancelTest.java
@@ -0,0 +1,110 @@
+package org.apache.qpid.test.client;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.test.VMTestCase;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.QueueBrowser;
+import javax.jms.Session;
+import javax.jms.JMSException;
+import javax.naming.NamingException;
+import java.util.Enumeration;/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+public class CancelTest extends VMTestCase
+{
+    private static final Logger _logger = Logger.getLogger(QueueBrowserTest.class);
+
+    private Connection _clientConnection;
+    private Session _clientSession;
+    private Queue _queue;
+
+    public void setUp() throws Exception
+    {
+
+        super.setUp();
+
+        _queue = (Queue) _context.lookup("queue");
+
+        //Create Client
+        _clientConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
+
+        _clientConnection.start();
+
+        _clientSession = _clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        //Ensure _queue is created
+        _clientSession.createConsumer(_queue).close();
+    }
+
+    /**
+     * Simply
+     */
+    public void test() throws JMSException, NamingException
+    {
+        Connection producerConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
+
+        producerConnection.start();
+
+        Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageProducer producer = producerSession.createProducer(_queue);
+        producer.send(producerSession.createTextMessage());
+        producerConnection.close();
+
+
+        QueueBrowser browser = _clientSession.createBrowser(_queue);
+        Enumeration e = browser.getEnumeration();
+
+
+        while (e.hasMoreElements())
+        {
+            e.nextElement();
+        }
+
+        browser.close();
+
+        MessageConsumer consumer = _clientSession.createConsumer(_queue);
+        consumer.receive();
+        consumer.close();
+    }
+
+    public void loop()
+    {
+        try
+        {
+            int run = 0;
+            while (true)
+            {
+                System.err.println(run++);
+                test();
+            }
+        }
+        catch (Exception e)
+        {
+            _logger.error(e, e);
+        }
+    }
+
+}
diff --git a/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserTest.java b/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserTest.java
index 06f05462ba..72b8dfcb1c 100644
--- a/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserTest.java
@@ -14,14 +14,17 @@
  *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  *  KIND, either express or implied.  See the License for the
  *  specific language governing permissions and limitations
- *  under the License.    
+ *  under the License.
+ *
  *
- * 
  */
 package org.apache.qpid.test.client;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.test.VMTestCase;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.AMQException;
 
 import javax.jms.Queue;
 import javax.jms.ConnectionFactory;
@@ -88,7 +91,6 @@ public class QueueBrowserTest extends VMTestCase
 
     private void checkQueueDepth(int depth) throws JMSException, NamingException
     {
-        sendMessages(depth);
 
         // create QueueBrowser
         _logger.info("Creating Queue Browser");
@@ -101,6 +103,19 @@ public class QueueBrowserTest extends VMTestCase
             _logger.debug("Checking for " + depth + " messages with QueueBrowser");
         }
 
+        long queueDepth = 0;
+
+        try
+        {
+            queueDepth = ((AMQSession) _clientSession).getQueueDepth((AMQDestination) _queue);
+        }
+        catch (AMQException e)
+        {
+        }
+
+        assertEquals("Session reports Queue depth not as expected", depth, queueDepth);
+
+
         int msgCount = 0;
         Enumeration msgs = queueBrowser.getEnumeration();
 
@@ -116,11 +131,7 @@ public class QueueBrowserTest extends VMTestCase
         }
 
         // check to see if all messages found
-//        assertEquals("browser did not find all messages", MSG_COUNT, msgCount);
-        if (msgCount != depth)
-        {
-            _logger.warn(msgCount + " off" + depth + " messages received.");
-        }
+        assertEquals("Browser did not find all messages", depth, msgCount);
 
         //Close browser
         queueBrowser.close();
@@ -132,39 +143,61 @@ public class QueueBrowserTest extends VMTestCase
      *
      */
 
-     public void testQueueBrowserMsgsRemainOnQueue() throws Exception
-     {
-         int messages = 10;
+    public void testQueueBrowserMsgsRemainOnQueue() throws Exception
+    {
+        int messages = 10;
+
+        sendMessages(messages);
+
+        checkQueueDepth(messages);
 
-         checkQueueDepth(messages);
+        // VERIFY
 
-         // VERIFY
+        // continue and try to receive all messages
+        MessageConsumer consumer = _clientSession.createConsumer(_queue);
 
-         // continue and try to receive all messages
-         MessageConsumer consumer = _clientSession.createConsumer(_queue);
+        _logger.info("Verify messages are still on the queue");
 
-         _logger.info("Verify messages are still on the queue");
+        Message tempMsg;
 
-         Message tempMsg;
+        for (int msgCount = 0; msgCount < messages; msgCount++)
+        {
+            tempMsg = (TextMessage) consumer.receive(RECEIVE_TIMEOUT);
+            if (tempMsg == null)
+            {
+                fail("Message " + msgCount + " not retrieved from queue");
+            }
+        }
 
-         for (int msgCount = 0; msgCount < messages; msgCount++)
-         {
-             tempMsg = (TextMessage) consumer.receive(RECEIVE_TIMEOUT);
-             if (tempMsg == null)
-             {
-                 fail("Message " + msgCount + " not retrieved from queue");
-             }
-         }
+        consumer.close();
+
+        _logger.info("All messages recevied from queue");
+    }
 
-         _logger.info("All messages recevied from queue");
-     }
+    /**
+     * This tests you can browse an empty queue, see QPID-785
+     *
+     * @throws Exception
+     */
+    public void testBrowsingEmptyQueue() throws Exception
+    {
+        checkQueueDepth(0);
+    }
 
-     /**
-      * This tests you can browse an empty queue, see QPID-785
-      * @throws Exception
-      */
-     public void testBrowsingEmptyQueue() throws Exception
-     {
-         checkQueueDepth(0);
-     }
+    public void loop() throws JMSException
+    {
+        int run = 0;
+        try
+        {
+            while (true)
+            {
+                System.err.println(run++ + ":************************************************************************");
+                testQueueBrowserMsgsRemainOnQueue();
+            }
+        }
+        catch (Exception e)
+        {
+            _logger.error(e, e);
+        }
+    }
 }