resolve https://issues.apache.org/activemq/browse/AMQ-2580 - patch was good for AMQ store, fix needed for kahaDB and JDBC now done, test case was great, thanks.

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

activemq-core/src/main/java/org/apache/activemq/store/amq/RecoveryListenerAdapter.java

@@ -46,10 +46,11 @@ final class RecoveryListenerAdapter implements MessageRecoveryListener {
     
     public boolean recoverMessage(Message message) throws Exception {
         if (listener.hasSpace()) {
-            listener.recoverMessage(message);
-            lastRecovered = message.getMessageId();
-            count++;
-            return true;
+            if (listener.recoverMessage(message)) {
+                lastRecovered = message.getMessageId();
+                count++;
+                return true;
+            }
         }
         return false;
     }

activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCTopicMessageStore.java

@@ -108,10 +108,11 @@ public class JDBCTopicMessageStore extends JDBCMessageStore implements TopicMess
                     if (listener.hasSpace()) {
                         Message msg = (Message)wireFormat.unmarshal(new ByteSequence(data));
                         msg.getMessageId().setBrokerSequenceId(sequenceId);
-                        listener.recoverMessage(msg);
-                        finalLast.set(sequenceId);
-                        finalPriority.set(msg.getPriority());
-                        return true;
+                        if (listener.recoverMessage(msg)) {
+                            finalLast.set(sequenceId);
+                            finalPriority.set(msg.getPriority());
+                            return true;
+                        }
                     }
                     return false;
                 }

activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/DefaultJDBCAdapter.java

@@ -451,7 +451,7 @@ public class DefaultJDBCAdapter implements JDBCAdapter {
             } else {
                 s = c.getConnection().prepareStatement(this.statements.getFindDurableSubMessagesStatement());
             }
-            s.setMaxRows(maxReturned);
+            // no set max rows as selectors may need to scan more than maxReturned messages to get what they need
             s.setString(1, destination.getQualifiedName());
             s.setString(2, clientId);
             s.setString(3, subscriptionName);
@@ -466,16 +466,12 @@ public class DefaultJDBCAdapter implements JDBCAdapter {
                 while (rs.next() && count < maxReturned) {
                     if (listener.recoverMessageReference(rs.getString(1))) {
                         count++;
-                    } else {
-                        break;
                     }
                 }
             } else {
                 while (rs.next() && count < maxReturned) {
                     if (listener.recoverMessage(rs.getLong(1), getBinaryData(rs, 2))) {
                         count++;
-                    } else {
-                        break;
                     }
                 }
             }

activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTopicReferenceStore.java

@@ -305,8 +305,6 @@ public class KahaTopicReferenceStore extends KahaReferenceStore implements Topic
                             if (recoverReference(listener, msg)) {
                                 count++;
                                 container.setBatchEntry(msg.getMessageId(), entry);
-                            } else {
-                                break;
                             }
                         } else {
                             container.reset();

activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java

@@ -798,8 +798,9 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
                         for (Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx, cursorPos); iterator
                                 .hasNext();) {
                             entry = iterator.next();
-                            listener.recoverMessage(loadMessage(entry.getValue().location));
-                            counter++;
+                            if (listener.recoverMessage(loadMessage(entry.getValue().location))) {
+                                counter++;
+                            }
                             if (counter >= maxReturned || listener.hasSpace() == false) {
                                 break;
                             }

activemq-core/src/test/java/org/apache/activemq/JmsTopicSelectorTest.java

@@ -47,14 +47,6 @@ public class JmsTopicSelectorTest extends TestSupport {
     protected boolean durable;
     protected int deliveryMode = DeliveryMode.PERSISTENT;
 
-    public JmsTopicSelectorTest() {
-        super();
-    }
-
-    public JmsTopicSelectorTest(String name) {
-        super(name);
-    }
-
     public void setUp() throws Exception {
         super.setUp();
 

activemq-core/src/test/java/org/apache/activemq/TestSupport.java

@@ -17,6 +17,8 @@
 package org.apache.activemq;
 
 import java.io.File;
+import java.io.IOException;
+import java.util.Enumeration;
 import java.util.Map;
 
 import javax.jms.Connection;
@@ -34,6 +36,10 @@ import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQMessage;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.store.PersistenceAdapter;
+import org.apache.activemq.store.amq.AMQPersistenceAdapter;
+import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -42,18 +48,11 @@ import org.apache.commons.logging.LogFactory;
  * 
  * @version $Revision: 1.5 $
  */
-public class TestSupport extends TestCase {
+public abstract class TestSupport extends CombinationTestSupport {
 
     protected ActiveMQConnectionFactory connectionFactory;
     protected boolean topic = true;
-
-    public TestSupport() {
-        super();
-    }
-
-    public TestSupport(String name) {
-        super(name);
-    }
+    public PersistenceAdapterChoice defaultPersistenceAdapter = PersistenceAdapterChoice.KahaDB;
 
     protected ActiveMQMessage createMessage() {
         return new ActiveMQMessage();
@@ -173,4 +172,28 @@ public class TestSupport extends TestCase {
                     regionBroker.getQueueRegion().getDestinationMap() :
                         regionBroker.getTopicRegion().getDestinationMap();
     }
+
+    public static enum PersistenceAdapterChoice {KahaDB, AMQ, JDBC };
+
+    public PersistenceAdapter setDefaultPersistenceAdapter(BrokerService broker) throws IOException {
+        return setPersistenceAdapter(broker, defaultPersistenceAdapter);
+    }
+    
+    public PersistenceAdapter setPersistenceAdapter(BrokerService broker, PersistenceAdapterChoice choice) throws IOException {
+        PersistenceAdapter adapter = null;
+        switch (choice) {
+        case AMQ:
+            adapter = new AMQPersistenceAdapter();
+            break;
+        case JDBC:
+            adapter = new JDBCPersistenceAdapter();
+            break;
+        case KahaDB:
+            adapter = new KahaDBPersistenceAdapter();
+            break;
+        }
+        broker.setPersistenceAdapter(adapter);
+        return adapter;
+    }
+
 }

activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2580Test.java

@@ -0,0 +1,202 @@
+/**
+ * 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.bugs;
+
+import junit.framework.Test;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.ActiveMQPrefetchPolicy;      
+import org.apache.activemq.TestSupport;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+import javax.jms.TopicConnection;
+import javax.jms.TopicSession;
+
+public class AMQ2580Test extends TestSupport {
+
+    private static final Log LOG = LogFactory.getLog(AMQ2580Test.class);
+
+    private static final String TOPIC_NAME = "topicName";
+    private static final String CLIENT_ID = "client_id";
+    private static final String textOfSelectedMsg = "good_message";
+
+    protected TopicConnection connection;
+
+    private Topic topic;
+    private Session session;
+    private MessageProducer producer;
+    private ConnectionFactory connectionFactory;
+    private TopicConnection topicConnection;
+    private BrokerService service;
+
+    public static Test suite() {
+        return suite(AMQ2580Test.class);
+    }
+
+    protected void setUp() throws Exception {
+        super.setUp();
+        initDurableBroker();
+        initConnectionFactory();
+        initTopic();
+    }
+
+    protected void tearDown() throws Exception {
+        shutdownClient();
+        service.stop();
+        super.tearDown();
+    }
+
+    private void initConnection() throws JMSException {
+        if (connection == null) {
+            LOG.info("Initializing connection");
+
+            connection = (TopicConnection) connectionFactory.createConnection();
+            connection.start();
+        }
+    }
+
+    public void initCombosForTestTopicIsDurableSmokeTest() throws Exception {
+        addCombinationValues("defaultPersistenceAdapter", PersistenceAdapterChoice.values());
+    }
+
+    public void testTopicIsDurableSmokeTest() throws Exception {
+
+        initClient();
+        MessageConsumer consumer = createMessageConsumer();
+        LOG.info("Consuming message");
+        assertNull(consumer.receive(1));
+        shutdownClient();
+        consumer.close();
+
+        sendMessages();
+        shutdownClient();
+
+        initClient();
+        consumer = createMessageConsumer();
+
+        LOG.info("Consuming message");
+        TextMessage answer1 = (TextMessage) consumer.receive(1000);
+        assertNotNull("we got our message", answer1);
+
+        consumer.close();
+    }
+
+    private MessageConsumer createMessageConsumer() throws JMSException {
+        LOG.info("creating durable subscriber");
+        return session.createDurableSubscriber(topic,
+                TOPIC_NAME,
+                "name='value'",
+                false);
+    }
+
+    private void initClient() throws JMSException {
+        LOG.info("Initializing client");
+
+        initConnection();
+        initSession();
+    }
+
+    private void shutdownClient()
+            throws JMSException {
+        LOG.info("Closing session and connection");
+        session.close();
+        connection.close();
+        session = null;
+        connection = null;
+    }
+
+    private void sendMessages()
+            throws JMSException {
+        initConnection();
+
+        initSession();
+
+        LOG.info("Creating producer");
+        producer = session.createProducer(topic);
+
+        sendMessageThatFailsSelection();
+
+        sendMessage(textOfSelectedMsg, "value");
+    }
+
+    private void initSession() throws JMSException {
+        LOG.info("Initializing session");
+        session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
+    }
+
+    private void sendMessageThatFailsSelection() throws JMSException {
+        for (int i = 0; i < 5; i++) {
+            String textOfNotSelectedMsg = "Msg_" + i;
+            sendMessage(textOfNotSelectedMsg, "not_value");
+            LOG.info("#");
+        }
+    }
+
+    private void sendMessage(
+            String msgText,
+            String propertyValue) throws JMSException {
+        LOG.info("Creating message: " + msgText);
+        TextMessage messageToSelect = session.createTextMessage(msgText);
+        messageToSelect.setStringProperty("name", propertyValue);
+        LOG.info("Sending message");
+        producer.send(messageToSelect);
+    }
+
+    protected void initConnectionFactory() throws Exception {
+        ActiveMQConnectionFactory activeMqConnectionFactory = createActiveMqConnectionFactory();
+        connectionFactory = activeMqConnectionFactory;
+    }
+
+
+    private ActiveMQConnectionFactory createActiveMqConnectionFactory() throws Exception {
+        ActiveMQConnectionFactory activeMqConnectionFactory = new ActiveMQConnectionFactory(
+                "failover:" + service.getTransportConnectors().get(0).getConnectUri().toString());
+        activeMqConnectionFactory.setWatchTopicAdvisories(false);
+        ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();
+        prefetchPolicy.setDurableTopicPrefetch(2);
+        prefetchPolicy.setOptimizeDurableTopicPrefetch(2);
+        activeMqConnectionFactory.setPrefetchPolicy(prefetchPolicy);
+        activeMqConnectionFactory.setClientID(CLIENT_ID);
+        return activeMqConnectionFactory;
+    }
+
+    private void initDurableBroker() throws Exception {
+        service = new BrokerService();
+        setDefaultPersistenceAdapter(service);
+        service.setDeleteAllMessagesOnStartup(true);
+        service.setAdvisorySupport(false);
+        service.setTransportConnectorURIs(new String[]{"tcp://localhost:0"});
+        service.setPersistent(true);
+        service.setUseJmx(false);
+        service.start();
+
+    }
+
+    private void initTopic() throws JMSException {
+        topicConnection = (TopicConnection) connectionFactory.createConnection();
+        TopicSession topicSession = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
+        topic = topicSession.createTopic(TOPIC_NAME);
+    }
+}