https://issues.apache.org/jira/browse/AMQ-4181 - jms queue browsing

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1411632 13f79535-47bb-0310-9956-ffa450edef68

activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java

@@ -69,14 +69,19 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
     protected final Object dispatchLock = new Object();
     private final CountDownLatch okForAckAsDispatchDone = new CountDownLatch(1);
 
-    public PrefetchSubscription(Broker broker, SystemUsage usageManager, ConnectionContext context, ConsumerInfo info, PendingMessageCursor cursor) throws InvalidSelectorException {
+    public PrefetchSubscription(Broker broker, SystemUsage usageManager, ConnectionContext context, ConsumerInfo info, PendingMessageCursor cursor) throws JMSException {
         super(broker,context, info);
         this.usageManager=usageManager;
         pending = cursor;
+        try {
+            pending.start();
+        } catch (Exception e) {
+            throw new JMSException(e.getMessage());
+        }
         this.scheduler = broker.getScheduler();
     }
 
-    public PrefetchSubscription(Broker broker,SystemUsage usageManager, ConnectionContext context, ConsumerInfo info) throws InvalidSelectorException {
+    public PrefetchSubscription(Broker broker,SystemUsage usageManager, ConnectionContext context, ConsumerInfo info) throws JMSException {
         this(broker,usageManager,context, info, new VMPendingMessageCursor(false));
     }
 

activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java

@@ -568,6 +568,10 @@ public class Queue extends BaseDestination implements Task, UsageListener {
                     }
                     redeliveredWaitingDispatch.addMessageLast(qmr);
                 }
+                if (sub instanceof QueueBrowserSubscription) {
+                    ((QueueBrowserSubscription)sub).decrementQueueRef();
+                    browserDispatches.remove(sub);
+                }
                 if (!redeliveredWaitingDispatch.isEmpty()) {
                     doDispatch(new OrderedPendingList());
                 }
@@ -1460,7 +1464,6 @@ public class Queue extends BaseDestination implements Task, UsageListener {
                 }
             }
 
-            BrowserDispatch pendingBrowserDispatch = browserDispatches.poll();
 
             messagesLock.readLock().lock();
             try{
@@ -1481,16 +1484,18 @@ public class Queue extends BaseDestination implements Task, UsageListener {
             // !messages.isEmpty(), and then if
             // !pagedInPendingDispatch.isEmpty()
             // then we do a dispatch.
-            if (pageInMoreMessages || pendingBrowserDispatch != null || !redeliveredWaitingDispatch.isEmpty()) {
+            boolean hasBrowsers = browserDispatches.size() > 0;
+
+            if (pageInMoreMessages || hasBrowsers || !redeliveredWaitingDispatch.isEmpty()) {
                 try {
-                    pageInMessages(pendingBrowserDispatch != null);
+                    pageInMessages(hasBrowsers);
 
                 } catch (Throwable e) {
                     LOG.error("Failed to page in more queue messages ", e);
                 }
             }
 
-            if (pendingBrowserDispatch != null) {
+            if (hasBrowsers) {
                 ArrayList<QueueMessageReference> alreadyDispatchedMessages = null;
                 pagedInMessagesLock.readLock().lock();
                 try{
@@ -1498,30 +1503,38 @@ public class Queue extends BaseDestination implements Task, UsageListener {
                 }finally {
                     pagedInMessagesLock.readLock().unlock();
                 }
-                if (LOG.isDebugEnabled()) {
+
-                    LOG.debug("dispatch to browser: " + pendingBrowserDispatch.getBrowser()
+                Iterator<BrowserDispatch> browsers = browserDispatches.iterator();
-                            + ", already dispatched/paged count: " + alreadyDispatchedMessages.size());
+                while (browsers.hasNext()) {
-                }
+                    BrowserDispatch browserDispatch = browsers.next();
-                do {
                     try {
                         MessageEvaluationContext msgContext = new NonCachedMessageEvaluationContext();
                         msgContext.setDestination(destination);
 
-                        QueueBrowserSubscription browser = pendingBrowserDispatch.getBrowser();
+                        QueueBrowserSubscription browser = browserDispatch.getBrowser();
+
+                        if (LOG.isDebugEnabled()) {
+                            LOG.debug("dispatch to browser: " + browser
+                                    + ", already dispatched/paged count: " + alreadyDispatchedMessages.size());
+                        }
+                        boolean added = false;
                         for (QueueMessageReference node : alreadyDispatchedMessages) {
-                            if (!node.isAcked()) {
+                            if (!node.isAcked() && !browser.getPending().getMessageAudit().isDuplicate(node.getMessageId())) {
                                 msgContext.setMessageReference(node);
                                 if (browser.matches(node, msgContext)) {
                                     browser.add(node);
+                                    added = true;
                                 }
                             }
                         }
-                        pendingBrowserDispatch.done();
+                        // are we done browsing? no new messages paged
+                        if (!added) {
+                            browser.decrementQueueRef();
+                        }
                     } catch (Exception e) {
-                        LOG.warn("exception on dispatch to browser: " + pendingBrowserDispatch.getBrowser(), e);
+                        LOG.warn("exception on dispatch to browser: " + browserDispatch.getBrowser(), e);
                     }
-
+                }
-                } while ((pendingBrowserDispatch = browserDispatches.poll()) != null);
             }
 
             if (pendingWakeups.get() > 0) {

activemq-broker/src/main/java/org/apache/activemq/broker/region/QueueBrowserSubscription.java

@@ -20,8 +20,11 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import javax.jms.InvalidSelectorException;
+import javax.jms.JMSException;
+
 import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
+import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor;
 import org.apache.activemq.command.ConsumerInfo;
 import org.apache.activemq.command.MessageAck;
 import org.apache.activemq.filter.MessageEvaluationContext;
@@ -34,7 +37,7 @@ public class QueueBrowserSubscription extends QueueSubscription {
     boolean destinationsAdded;
 
     public QueueBrowserSubscription(Broker broker,SystemUsage usageManager, ConnectionContext context, ConsumerInfo info)
-        throws InvalidSelectorException {
+        throws JMSException {
         super(broker,usageManager, context, info);
     }
 

activemq-broker/src/main/java/org/apache/activemq/broker/region/QueueSubscription.java

@@ -34,7 +34,7 @@ public class QueueSubscription extends PrefetchSubscription implements LockOwner
 
     private static final Logger LOG = LoggerFactory.getLogger(QueueSubscription.class);
 
-    public QueueSubscription(Broker broker, SystemUsage usageManager, ConnectionContext context, ConsumerInfo info) throws InvalidSelectorException {
+    public QueueSubscription(Broker broker, SystemUsage usageManager, ConnectionContext context, ConsumerInfo info) throws JMSException {
         super(broker,usageManager, context, info);
     }
 

activemq-client/src/main/java/org/apache/activemq/ActiveMQQueueBrowser.java

@@ -202,6 +202,7 @@ public class ActiveMQQueueBrowser implements QueueBrowser, Enumeration {
     }
 
     public synchronized void close() throws JMSException {
+        browseDone.set(true);
         destroyConsumer();
         closed = true;
     }

activemq-core/src/test/java/org/apache/activemq/broker/BrokerTest.java

@@ -264,7 +264,7 @@ public class BrokerTest extends BrokerTestSupport {
             messages.add(m1);
         }
 
-        for (int i = 0; i < 1; i++) {
+        for (int i = 0; i < 4; i++) {
             Message m1 = messages.get(i);
             Message m2 = receiveMessage(connection2);
             assertNotNull("m2 is null for index: " + i, m2);

activemq-core/src/test/java/org/apache/activemq/usecases/QueueBrowsingTest.java

@@ -0,0 +1,198 @@
+/**
+ * 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.
+ */
+package org.apache.activemq.usecases;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.*;
+
+import java.net.URI;
+import java.util.Enumeration;
+
+import static org.junit.Assert.*;
+
+public class QueueBrowsingTest {
+
+    private static final Logger LOG = LoggerFactory.getLogger(QueueBrowsingTest.class);
+
+    private BrokerService broker;
+    private URI connectUri;
+    private ActiveMQConnectionFactory factory;
+
+
+    @Before
+    public void startBroker() throws Exception {
+        broker = new BrokerService();
+        TransportConnector connector = broker.addConnector("tcp://0.0.0.0:0");
+        broker.deleteAllMessages();
+        broker.start();
+        broker.waitUntilStarted();
+        connectUri = connector.getConnectUri();
+        factory = new ActiveMQConnectionFactory(connectUri);
+    }
+
+    @After
+    public void stopBroker() throws Exception {
+        broker.stop();
+        broker.waitUntilStopped();
+    }
+
+    @Test
+    public void testBrowsing() throws JMSException {
+
+        int messageToSend = 370;
+
+        ActiveMQQueue queue = new ActiveMQQueue("TEST");
+        Connection connection = factory.createConnection();
+        connection.start();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageProducer producer = session.createProducer(queue);
+
+        String data = "";
+        for( int i=0; i < 1024*2; i++ ) {
+            data += "x";
+        }
+
+        for( int i=0; i < messageToSend; i++ ) {
+            producer.send(session.createTextMessage(data));
+        }
+
+        QueueBrowser browser = session.createBrowser(queue);
+        Enumeration enumeration = browser.getEnumeration();
+        int received = 0;
+        while (enumeration.hasMoreElements()) {
+            Message m = (Message) enumeration.nextElement();
+            received++;
+        }
+
+        browser.close();
+
+        assertEquals(messageToSend, received);
+    }
+
+    @Test
+    public void testBrowseConcurrent() throws Exception {
+        final int messageToSend = 370;
+
+        final ActiveMQQueue queue = new ActiveMQQueue("TEST");
+        Connection connection = factory.createConnection();
+        connection.start();
+        final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        MessageProducer producer = session.createProducer(queue);
+
+        String data = "";
+        for( int i=0; i < 1024*2; i++ ) {
+            data += "x";
+        }
+
+        for( int i=0; i < messageToSend; i++ ) {
+            producer.send(session.createTextMessage(data));
+        }
+
+        Thread browserThread = new Thread() {
+            @Override
+            public void run() {
+                try {
+                    QueueBrowser browser = session.createBrowser(queue);
+                    Enumeration enumeration = browser.getEnumeration();
+                    int received = 0;
+                    while (enumeration.hasMoreElements()) {
+                        Message m = (Message) enumeration.nextElement();
+                        received++;
+                    }
+                    assertEquals("Browsed all messages", messageToSend, received);
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
+            }
+        };
+
+        browserThread.start();
+
+        Thread consumerThread = new Thread() {
+            @Override
+            public void run() {
+                try {
+                    MessageConsumer consumer = session.createConsumer(queue);
+                    int received = 0;
+                    while (true) {
+                        Message m = (Message) consumer.receive(1000);
+                        if (m == null)
+                            break;
+                        received++;
+                    }
+                    assertEquals("Consumed all messages", messageToSend, received);
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
+            }
+        };
+
+        consumerThread.start();
+
+        browserThread.join();
+        consumerThread.join();
+
+    }
+
+    @Test
+    public void testMemoryLimit() throws Exception {
+        broker.getSystemUsage().getMemoryUsage().setLimit(10 * 1024);
+
+        int messageToSend = 370;
+
+        ActiveMQQueue queue = new ActiveMQQueue("TEST");
+        Connection connection = factory.createConnection();
+        connection.start();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageProducer producer = session.createProducer(queue);
+
+        String data = "";
+        for( int i=0; i < 1024*2; i++ ) {
+            data += "x";
+        }
+
+        for( int i=0; i < messageToSend; i++ ) {
+            producer.send(session.createTextMessage(data));
+        }
+
+        QueueBrowser browser = session.createBrowser(queue);
+        Enumeration enumeration = browser.getEnumeration();
+        int received = 0;
+        while (enumeration.hasMoreElements()) {
+            Message m = (Message) enumeration.nextElement();
+            received++;
+            LOG.info("Browsed message " + received + ": " + m.getJMSMessageID());
+        }
+
+        browser.close();
+
+        assertEquals(3, received);
+    }
+
+
+
+}