fix for: https://issues.apache.org/jira/browse/AMQ-4062

Durable consumer should be configured with the policyEntry configuration on activate. 

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

activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java

@@ -18,7 +18,6 @@ package org.apache.activemq.broker.region;
 
 import java.io.IOException;
 import java.util.Collections;
-import java.util.Iterator;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -30,6 +29,7 @@ import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.cursors.PendingMessageCursor;
 import org.apache.activemq.broker.region.cursors.StoreDurableSubscriberCursor;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ConsumerInfo;
 import org.apache.activemq.command.Message;
@@ -55,9 +55,9 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
     private AtomicLong offlineTimestamp = new AtomicLong(-1);
 
     public DurableTopicSubscription(Broker broker, SystemUsage usageManager, ConnectionContext context, ConsumerInfo info, boolean keepDurableSubsActive)
-        throws JMSException {
+            throws JMSException {
-        super(broker,usageManager, context, info);
+        super(broker, usageManager, context, info);
-        this.pending = new StoreDurableSubscriberCursor(broker,context.getClientId(), info.getSubscriptionName(), info.getPrefetchSize(), this);
+        this.pending = new StoreDurableSubscriberCursor(broker, context.getClientId(), info.getSubscriptionName(), info.getPrefetchSize(), this);
         this.pending.setSystemUsage(usageManager);
         this.pending.setMemoryUsageHighWaterMark(getCursorMemoryHighWaterMark());
         this.keepDurableSubsActive = keepDurableSubsActive;
@@ -80,8 +80,8 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
     }
 
     /**
-     * store will have a pending ack for all durables, irrespective of the selector
+     * store will have a pending ack for all durables, irrespective of the
-     * so we need to ack if node is un-matched
+     * selector so we need to ack if node is un-matched
      */
     public void unmatched(MessageReference node) throws IOException {
         MessageAck ack = new MessageAck();
@@ -101,23 +101,23 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
         }
         // do it just once per destination
         if (durableDestinations.containsKey(destination.getActiveMQDestination())) {
-             return;
+            return;
         }
         durableDestinations.put(destination.getActiveMQDestination(), destination);
 
         if (active.get() || keepDurableSubsActive) {
-            Topic topic = (Topic)destination;
+            Topic topic = (Topic) destination;
             topic.activate(context, this);
             if (pending.isEmpty(topic)) {
                 topic.recoverRetroactiveMessages(context, this);
             }
-            this.enqueueCounter+=pending.size();
+            this.enqueueCounter += pending.size();
         } else if (destination.getMessageStore() != null) {
-            TopicMessageStore store = (TopicMessageStore)destination.getMessageStore();
+            TopicMessageStore store = (TopicMessageStore) destination.getMessageStore();
             try {
-                this.enqueueCounter+=store.getMessageCount(subscriptionKey.getClientId(),subscriptionKey.getSubscriptionName());
+                this.enqueueCounter += store.getMessageCount(subscriptionKey.getClientId(), subscriptionKey.getSubscriptionName());
             } catch (IOException e) {
-                JMSException jmsEx = new JMSException("Failed to retrieve enqueueCount from store "+ e);
+                JMSException jmsEx = new JMSException("Failed to retrieve enqueueCount from store " + e);
                 jmsEx.setLinkedException(e);
                 throw jmsEx;
             }
@@ -125,16 +125,24 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
         dispatchPending();
     }
 
-    public void activate(SystemUsage memoryManager, ConnectionContext context,
+    public void activate(SystemUsage memoryManager, ConnectionContext context, ConsumerInfo info, RegionBroker regionBroker) throws Exception {
-            ConsumerInfo info) throws Exception {
         if (!active.get()) {
             this.context = context;
             this.info = info;
+
+            // On Activation we should update the configuration based on our new consumer info.
+            ActiveMQDestination dest = this.info.getDestination();
+            if (dest != null && regionBroker.getDestinationPolicy() != null) {
+                PolicyEntry entry = regionBroker.getDestinationPolicy().getEntryFor(dest);
+                if (entry != null) {
+                    entry.configure(broker, usageManager, this);
+                }
+            }
+
             LOG.debug("Activating " + this);
             if (!keepDurableSubsActive) {
-                for (Iterator<Destination> iter = durableDestinations.values()
+                for (Destination destination : durableDestinations.values()) {
-                        .iterator(); iter.hasNext();) {
+                    Topic topic = (Topic) destination;
-                    Topic topic = (Topic) iter.next();
                     add(context, topic);
                     topic.activate(context, this);
                 }
@@ -148,9 +156,8 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
                 // If nothing was in the persistent store, then try to use the
                 // recovery policy.
                 if (pending.isEmpty()) {
-                    for (Iterator<Destination> iter = durableDestinations.values()
+                    for (Destination destination : durableDestinations.values()) {
-                            .iterator(); iter.hasNext();) {
+                        Topic topic = (Topic) destination;
-                        Topic topic = (Topic) iter.next();
                         topic.recoverRetroactiveMessages(context, this);
                     }
                 }
@@ -171,8 +178,8 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
             pending.stop();
 
             synchronized (dispatchLock) {
-                for (Iterator<Destination> iter = durableDestinations.values().iterator(); iter.hasNext();) {
+                for (Destination destination : durableDestinations.values()) {
-                    Topic topic = (Topic)iter.next();
+                    Topic topic = (Topic) destination;
                     if (!keepDurableSubsActive) {
                         topic.deactivate(context, this);
                     } else {
@@ -247,7 +254,7 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
     }
 
     protected void doAddRecoveredMessage(MessageReference message) throws Exception {
-        synchronized(pending) {
+        synchronized (pending) {
             pending.addRecoveredMessage(message);
         }
     }
@@ -275,8 +282,9 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
     }
 
     public synchronized String toString() {
-        return "DurableTopicSubscription-" + getSubscriptionKey() + ", id=" + info.getConsumerId() + ", active=" + isActive() + ", destinations=" + durableDestinations.size() + ", total=" + enqueueCounter + ", pending="
+        return "DurableTopicSubscription-" + getSubscriptionKey() + ", id=" + info.getConsumerId() + ", active=" + isActive() + ", destinations="
-               + getPendingQueueSize() + ", dispatched=" + dispatchCounter + ", inflight=" + dispatched.size() + ", prefetchExtension=" + getPrefetchExtension();
+                + durableDestinations.size() + ", total=" + enqueueCounter + ", pending=" + getPendingQueueSize() + ", dispatched=" + dispatchCounter
+                + ", inflight=" + dispatched.size() + ", prefetchExtension=" + getPrefetchExtension();
     }
 
     public SubscriptionKey getSubscriptionKey() {
@@ -289,7 +297,6 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
     public void destroy() {
         synchronized (pendingLock) {
             try {
-
                 pending.reset();
                 while (pending.hasNext()) {
                     MessageReference node = pending.next();
@@ -301,7 +308,7 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
                 pending.clear();
             }
         }
-        synchronized  (dispatchLock) {
+        synchronized (dispatchLock) {
             for (MessageReference node : dispatched) {
                 node.decrementReferenceCount();
             }
@@ -321,7 +328,7 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
     }
 
     protected boolean isDropped(MessageReference node) {
-       return false;
+        return false;
     }
 
     public boolean isKeepDurableSubsActive() {

activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java

@@ -16,10 +16,18 @@
  */
 package org.apache.activemq.broker.region;
 
-import java.util.*;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
 import java.util.concurrent.ConcurrentHashMap;
+
 import javax.jms.InvalidDestinationException;
 import javax.jms.JMSException;
+
 import org.apache.activemq.advisory.AdvisorySupport;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.policy.PolicyEntry;
@@ -147,7 +155,7 @@ public class TopicRegion extends AbstractRegion {
                                            + " subscriberName: " + key.getSubscriptionName());
                 }
             }
-            sub.activate(usageManager, context, info);
+            sub.activate(usageManager, context, info, broker);
             return sub;
         } else {
             return super.addConsumer(context, info);
@@ -332,8 +340,6 @@ public class TopicRegion extends AbstractRegion {
         }
     }
 
-    /**
-     */
     private boolean hasDurableSubChanged(ConsumerInfo info1, ConsumerInfo info2) {
         if (info1.getSelector() != null ^ info2.getSelector() != null) {
             return true;
@@ -367,5 +373,4 @@ public class TopicRegion extends AbstractRegion {
     public boolean durableSubscriptionExists(SubscriptionKey key) {
         return this.durableSubscriptions.containsKey(key);
     }
-
 }

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

@@ -0,0 +1,273 @@
+/**
+ * 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 static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.Iterator;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.ActiveMQSession;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.DurableTopicSubscription;
+import org.apache.activemq.broker.region.RegionBroker;
+import org.apache.activemq.broker.region.Subscription;
+import org.apache.activemq.broker.region.TopicRegion;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ConsumerInfo;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
+import org.apache.activemq.util.SubscriptionKey;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class AMQ4062Test {
+
+    private BrokerService service;
+    private PolicyEntry policy;
+    private ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription> durableSubscriptions;
+
+    private static final int PREFETCH_SIZE_5=5;
+    private String connectionUri;
+
+    @Before
+    public void startBroker() throws IOException, Exception {
+        service=new BrokerService();
+        service.setPersistent(true);
+        service.setDeleteAllMessagesOnStartup(true);
+        service.setUseJmx(false);
+
+        KahaDBPersistenceAdapter pa=new KahaDBPersistenceAdapter();
+        File dataFile=new File("createData");
+        pa.setDirectory(dataFile);
+        pa.setJournalMaxFileLength(1024*1024*32);
+
+        service.setPersistenceAdapter(pa);
+
+        policy = new PolicyEntry();
+        policy.setTopic(">");
+        policy.setDurableTopicPrefetch(PREFETCH_SIZE_5);
+        PolicyMap pMap = new PolicyMap();
+        pMap.setDefaultEntry(policy);
+
+        service.setDestinationPolicy(pMap);
+
+        service.addConnector("tcp://localhost:0");
+
+        service.start();
+        service.waitUntilStarted();
+
+        connectionUri = service.getTransportConnectors().get(0).getPublishableConnectString();
+    }
+
+    public void restartBroker() throws IOException, Exception {
+        service=new BrokerService();
+        service.setPersistent(true);
+        service.setUseJmx(false);
+
+        KahaDBPersistenceAdapter pa=new KahaDBPersistenceAdapter();
+        File dataFile=new File("createData");
+        pa.setDirectory(dataFile);
+        pa.setJournalMaxFileLength(1024*1024*32);
+
+        service.setPersistenceAdapter(pa);
+
+        policy = new PolicyEntry();
+        policy.setTopic(">");
+        policy.setDurableTopicPrefetch(PREFETCH_SIZE_5);
+        PolicyMap pMap = new PolicyMap();
+        pMap.setDefaultEntry(policy);
+
+        service.setDestinationPolicy(pMap);
+        service.addConnector("tcp://localhost:0");
+        service.start();
+        service.waitUntilStarted();
+
+        connectionUri = service.getTransportConnectors().get(0).getPublishableConnectString();
+    }
+
+    @After
+    public void stopBroker() throws Exception {
+        service.stop();
+        service.waitUntilStopped();
+        service = null;
+    }
+
+    @Test
+    public void testDirableSubPrefetchRecovered() throws Exception{
+
+        PrefetchConsumer consumer=new PrefetchConsumer(true, connectionUri);
+        consumer.recieve();
+        durableSubscriptions=getDurableSubscriptions();
+        ConsumerInfo info=getConsumerInfo(durableSubscriptions);
+
+        //check if the prefetchSize equals to the size we set in the PolicyEntry
+        assertEquals(PREFETCH_SIZE_5, info.getPrefetchSize());
+
+        consumer.a.countDown();
+        Producer p=new Producer(connectionUri);
+        p.send();
+        p = null;
+
+        service.stop();
+        service.waitUntilStopped();
+        durableSubscriptions=null;
+
+        consumer = null;
+        stopBroker();
+
+        restartBroker();
+
+        getDurableSubscriptions();
+        info=null;
+        info = getConsumerInfo(durableSubscriptions);
+
+        //check if the prefetchSize equals to 0 after persistent storage recovered
+        //assertEquals(0, info.getPrefetchSize());
+
+        consumer=new PrefetchConsumer(false, connectionUri);
+        consumer.recieve();
+        consumer.a.countDown();
+
+        info=null;
+        info = getConsumerInfo(durableSubscriptions);
+
+        //check if the prefetchSize is the default size for durable consumer and the PolicyEntry
+        //we set earlier take no effect
+        //assertEquals(100, info.getPrefetchSize());
+        //info.getPrefetchSize() is 100,it should be 5,because I set the PolicyEntry as follows,
+        //policy.setDurableTopicPrefetch(PREFETCH_SIZE_5);
+        assertEquals(5, info.getPrefetchSize());
+    }
+
+    private ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription> getDurableSubscriptions() throws NoSuchFieldException, IllegalAccessException {
+        if(durableSubscriptions!=null) return durableSubscriptions;
+        RegionBroker regionBroker=(RegionBroker)service.getRegionBroker();
+        TopicRegion region=(TopicRegion)regionBroker.getTopicRegion();
+        Field field=TopicRegion.class.getDeclaredField("durableSubscriptions");
+        field.setAccessible(true);
+        durableSubscriptions=(ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription>)field.get(region);
+        return durableSubscriptions;
+    }
+
+    private ConsumerInfo getConsumerInfo(ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription> durableSubscriptions) {
+        ConsumerInfo info=null;
+        for(Iterator<DurableTopicSubscription> it=durableSubscriptions.values().iterator();it.hasNext();){
+            Subscription sub = it.next();
+            info=sub.getConsumerInfo();
+            if(info.getSubscriptionName().equals(PrefetchConsumer.SUBSCRIPTION_NAME)){
+                return info;
+            }
+        }
+        return null;
+    }
+
+    public class PrefetchConsumer implements MessageListener{
+        public static final String SUBSCRIPTION_NAME = "A_NAME_ABC_DEF";
+        private String user = ActiveMQConnection.DEFAULT_USER;
+        private String password = ActiveMQConnection.DEFAULT_PASSWORD;
+        private final String uri;
+        private boolean transacted;
+        ActiveMQConnection connection;
+        Session session;
+        MessageConsumer consumer;
+        private boolean needAck=false;
+        CountDownLatch a=new CountDownLatch(1);
+
+        public PrefetchConsumer(boolean needAck, String uri){
+            this.needAck=needAck;
+            this.uri = uri;
+        }
+
+        public void recieve() throws Exception{
+            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, uri);
+            connection = (ActiveMQConnection)connectionFactory.createConnection();
+            connection.setClientID("3");
+            connection.start();
+
+            session = connection.createSession(transacted, Session.CLIENT_ACKNOWLEDGE);
+            Destination destination = session.createTopic("topic2");
+            consumer = session.createDurableSubscriber((Topic)destination,SUBSCRIPTION_NAME);
+            consumer.setMessageListener(this);
+        }
+
+        public void onMessage(Message message) {
+            try {
+                a.await();
+            } catch (InterruptedException e1) {
+            }
+            if(needAck){
+                try {
+                    message.acknowledge();
+                    consumer.close();
+                    session.close();
+                    connection.close();
+                } catch (JMSException e) {
+                }
+            }
+        }
+    }
+
+    public class Producer {
+
+        protected final String user = ActiveMQConnection.DEFAULT_USER;
+
+        private String password = ActiveMQConnection.DEFAULT_PASSWORD;
+        private final String uri;
+        private boolean transacted;
+
+        public Producer(String uri) {
+            this.uri = uri;
+        }
+
+        public void send() throws Exception{
+            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, uri);
+            ActiveMQConnection connection = (ActiveMQConnection)connectionFactory.createConnection();
+            connection.start();
+
+            ActiveMQSession session = (ActiveMQSession)connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
+            Destination destination = session.createTopic("topic2");
+            MessageProducer producer = session.createProducer(destination);
+            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+            for(int i=0;i<100;i++){
+                TextMessage om=session.createTextMessage("hello from producer");
+                producer.send(om);
+            }
+            producer.close();
+            session.close();
+            connection.close();
+        }
+    }
+}