fix and test for: https://issues.apache.org/jira/browse/AMQ-3705

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

activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java

@@ -40,8 +40,6 @@ import org.slf4j.LoggerFactory;
 /**
  * This broker filter handles tracking the state of the broker for purposes of
  * publishing advisory messages to advisory consumers.
- *
- *
  */
 public class AdvisoryBroker extends BrokerFilter {
 
@@ -197,16 +195,14 @@ public class AdvisoryBroker extends BrokerFilter {
             info.setOperationType(DestinationInfo.REMOVE_OPERATION_TYPE);
             ActiveMQTopic topic = AdvisorySupport.getDestinationAdvisoryTopic(destination);
             fireAdvisory(context, topic, info);
-            try {
+            ActiveMQTopic[] advisoryDestinations = AdvisorySupport.getAllDestinationAdvisoryTopics(destination);
-                next.removeDestination(context, AdvisorySupport.getConsumerAdvisoryTopic(info.getDestination()), -1);
+            for(ActiveMQTopic advisoryDestination : advisoryDestinations) {
-            } catch (Exception expectedIfDestinationDidNotExistYet) {
+                try {
-            }
+                    next.removeDestination(context, advisoryDestination, -1);
-            try {
+                } catch (Exception expectedIfDestinationDidNotExistYet) {
-                next.removeDestination(context, AdvisorySupport.getProducerAdvisoryTopic(info.getDestination()), -1);
+                }
-            } catch (Exception expectedIfDestinationDidNotExistYet) {
             }
         }
-
     }
 
     @Override
@@ -220,16 +216,14 @@ public class AdvisoryBroker extends BrokerFilter {
             info.setOperationType(DestinationInfo.REMOVE_OPERATION_TYPE);
             ActiveMQTopic topic = AdvisorySupport.getDestinationAdvisoryTopic(destInfo.getDestination());
             fireAdvisory(context, topic, info);
-            try {
+            ActiveMQTopic[] advisoryDestinations = AdvisorySupport.getAllDestinationAdvisoryTopics(destInfo.getDestination());
-                next.removeDestination(context, AdvisorySupport.getConsumerAdvisoryTopic(info.getDestination()), -1);
+            for(ActiveMQTopic advisoryDestination : advisoryDestinations) {
-            } catch (Exception expectedIfDestinationDidNotExistYet) {
+                try {
-            }
+                    next.removeDestination(context, advisoryDestination, -1);
-            try {
+                } catch (Exception expectedIfDestinationDidNotExistYet) {
-                next.removeDestination(context, AdvisorySupport.getProducerAdvisoryTopic(info.getDestination()), -1);
+                }
-            } catch (Exception expectedIfDestinationDidNotExistYet) {
             }
         }
-
     }
 
     @Override

activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java

@@ -16,6 +16,8 @@
  */
 package org.apache.activemq.advisory;
 
+import java.util.ArrayList;
+
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import org.apache.activemq.ActiveMQMessageTransformation;
@@ -74,6 +76,28 @@ public final class AdvisorySupport {
         return CONNECTION_ADVISORY_TOPIC;
     }
 
+    public static ActiveMQTopic[] getAllDestinationAdvisoryTopics(Destination destination) throws JMSException {
+        return getAllDestinationAdvisoryTopics(ActiveMQMessageTransformation.transformDestination(destination));
+    }
+
+    public static ActiveMQTopic[] getAllDestinationAdvisoryTopics(ActiveMQDestination destination) throws JMSException {
+        ArrayList<ActiveMQTopic> result = new ArrayList<ActiveMQTopic>();
+
+        result.add(getConsumerAdvisoryTopic(destination));
+        result.add(getProducerAdvisoryTopic(destination));
+        result.add(getExpiredMessageTopic(destination));
+        result.add(getNoConsumersAdvisoryTopic(destination));
+        result.add(getSlowConsumerAdvisoryTopic(destination));
+        result.add(getFastProducerAdvisoryTopic(destination));
+        result.add(getMessageDiscardedAdvisoryTopic(destination));
+        result.add(getMessageDeliveredAdvisoryTopic(destination));
+        result.add(getMessageConsumedAdvisoryTopic(destination));
+        result.add(getMessageDLQdAdvisoryTopic(destination));
+        result.add(getFullAdvisoryTopic(destination));
+
+        return result.toArray(new ActiveMQTopic[0]);
+    }
+
     public static ActiveMQTopic getConsumerAdvisoryTopic(Destination destination) throws JMSException {
         return getConsumerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
     }
@@ -123,6 +147,17 @@ public final class AdvisorySupport {
         return new ActiveMQTopic(name);
     }
 
+    public static ActiveMQTopic getNoConsumersAdvisoryTopic(Destination destination) throws JMSException {
+        return getExpiredMessageTopic(ActiveMQMessageTransformation.transformDestination(destination));
+    }
+
+    public static ActiveMQTopic getNoConsumersAdvisoryTopic(ActiveMQDestination destination) {
+        if (destination.isQueue()) {
+            return getNoQueueConsumersAdvisoryTopic(destination);
+        }
+        return getNoTopicConsumersAdvisoryTopic(destination);
+    }
+
     public static ActiveMQTopic getNoTopicConsumersAdvisoryTopic(Destination destination) throws JMSException {
         return getNoTopicConsumersAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
     }

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

@@ -0,0 +1,162 @@
+/**
+ * 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 static org.junit.Assert.*;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.*;
+
+public class AdvisoryTopicCleanUpTest {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AdvisoryTopicCleanUpTest.class);
+
+    private BrokerService broker;
+    private String connectionUri;
+
+    protected ActiveMQConnectionFactory createConnectionFactory() throws Exception {
+        return new ActiveMQConnectionFactory(connectionUri + "?jms.redeliveryPolicy.maximumRedeliveries=2");
+    }
+
+    @Before
+    public void setUp() throws Exception {
+        createBroker();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        destroyBroker();
+    }
+
+    private void createBroker() throws Exception {
+        broker = new BrokerService();
+        broker.setPersistent(false);
+        broker.setDeleteAllMessagesOnStartup(true);
+        broker.setUseJmx(true);
+        connectionUri = broker.addConnector("tcp://localhost:0").getPublishableConnectString();
+
+        PolicyEntry policy = new PolicyEntry();
+        policy.setAdvisdoryForFastProducers(true);
+        policy.setAdvisoryForConsumed(true);
+        policy.setAdvisoryForDelivery(true);
+        policy.setAdvisoryForDiscardingMessages(true);
+        policy.setAdvisoryForSlowConsumers(true);
+        policy.setAdvisoryWhenFull(true);
+        policy.setProducerFlowControl(false);
+        PolicyMap pMap = new PolicyMap();
+        pMap.setDefaultEntry(policy);
+        broker.setDestinationPolicy(pMap);
+
+        broker.start();
+    }
+
+    protected Connection createConnection() throws Exception {
+        Connection con = createConnectionFactory().createConnection();
+        con.start();
+        return con;
+    }
+
+    private void destroyBroker() throws Exception {
+        if (broker != null)
+            broker.stop();
+    }
+
+    @Test
+    public void testAdvisoryTopic() throws Exception {
+        Connection connection = createConnection();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        ActiveMQDestination queue = (ActiveMQDestination) session.createQueue("AdvisoryTopicCleanUpTestQueue");
+        MessageProducer prod = session.createProducer(queue);
+        Message message = session.createMessage();
+        prod.send(message);
+        message = session.createMessage();
+        prod.send(message);
+        message = session.createMessage();
+        prod.send(message, Message.DEFAULT_DELIVERY_MODE, Message.DEFAULT_PRIORITY, 1000);
+        connection.close();
+        connection = createConnection();
+
+        session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        MessageConsumer consumer = session.createConsumer(queue);
+        message = consumer.receive(60 * 1000);
+        message.acknowledge();
+        connection.close();
+        connection = null;
+
+        for (int i = 0; i < 2; i++) {
+            connection = createConnection();
+            session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+            consumer = session.createConsumer(queue);
+            message = consumer.receive(60 * 1000);
+            session.rollback();
+            connection.close();
+            connection = null;
+        }
+
+        Thread.sleep(2 * 1000);
+
+        connection = createConnection();
+        session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        consumer = session.createConsumer(queue);
+        message = consumer.receive(1000);
+        if (message != null)
+            message.acknowledge();
+        connection.close();
+        connection = null;
+
+        TimeUnit.SECONDS.sleep(1);
+
+        ActiveMQDestination dests[] = broker.getRegionBroker().getDestinations();
+
+        for (ActiveMQDestination destination: dests) {
+            String name = destination.getPhysicalName();
+            if (name.contains(queue.getPhysicalName())) {
+                LOG.debug("Destination on Broker before removing the Queue: " + name);
+            }
+        }
+
+        dests = broker.getRegionBroker().getDestinations();
+        if (dests == null) {
+            fail("Should have Destination for: " + queue.getPhysicalName());
+        }
+
+        broker.getAdminView().removeQueue(queue.getPhysicalName());
+
+        dests = broker.getRegionBroker().getDestinations();
+        if (dests != null)
+        {
+            for (ActiveMQDestination destination: dests) {
+                String name = destination.getPhysicalName();
+                LOG.debug("Destination on broker after removing the Queue: " + name);
+                assertFalse("Advisory topic should not exist. " + name,
+                            name.startsWith("ActiveMQ.Advisory") && name.contains(queue.getPhysicalName()));
+            }
+        }
+    }
+}