diff --git a/activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java b/activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java index 56392d7ccb..213af6814b 100755 --- a/activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java +++ b/activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java @@ -29,6 +29,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ThreadPoolExecutor; import javax.jms.InvalidClientIDException; import javax.jms.JMSException; +import org.apache.activemq.advisory.AdvisorySupport; import org.apache.activemq.broker.Broker; import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.Connection; @@ -320,9 +321,11 @@ public class RegionBroker extends EmptyBroker { switch (destination.getDestinationType()) { case ActiveMQDestination.QUEUE_TYPE: queueRegion.removeDestination(context, destination, timeout); + removeAdvisoryTopics("Queue.", context, destination, timeout); break; case ActiveMQDestination.TOPIC_TYPE: topicRegion.removeDestination(context, destination, timeout); + removeAdvisoryTopics("Topic.", context, destination, timeout); break; case ActiveMQDestination.TEMP_QUEUE_TYPE: tempQueueRegion.removeDestination(context, destination, timeout); @@ -334,10 +337,30 @@ public class RegionBroker extends EmptyBroker { throw createUnknownDestinationTypeException(destination); } destinations.remove(destination); + } } + public void removeAdvisoryTopics(String destinationType, ConnectionContext context, ActiveMQDestination destination, long timeout) throws Exception { + if (this.brokerService.isAdvisorySupport()) { + String producerAdvisoryTopic = AdvisorySupport.PRODUCER_ADVISORY_TOPIC_PREFIX + destinationType + destination.getPhysicalName(); + String consumerAdvisoryTopic = AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX + destinationType + destination.getPhysicalName(); + + ActiveMQDestination dests[] = getDestinations(); + for (ActiveMQDestination dest: dests) { + String name = dest.getPhysicalName(); + if ( name.equals(producerAdvisoryTopic) || name.equals(consumerAdvisoryTopic) ) { + try { + removeDestination(context, dest, timeout); + } catch (JMSException ignore) { + // at least ignore the Unknown Destination Type JMSException + } + } + } + } + } + @Override public void addDestinationInfo(ConnectionContext context, DestinationInfo info) throws Exception { addDestination(context, info.getDestination(),true); diff --git a/activemq-core/src/test/java/org/apache/activemq/usecases/AdvisoryTopicDeletionTest.java b/activemq-core/src/test/java/org/apache/activemq/usecases/AdvisoryTopicDeletionTest.java new file mode 100644 index 0000000000..41edd60c5e --- /dev/null +++ b/activemq-core/src/test/java/org/apache/activemq/usecases/AdvisoryTopicDeletionTest.java @@ -0,0 +1,123 @@ +/** + * 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 javax.jms.Connection; +import javax.jms.Destination; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.MessageProducer; +import javax.jms.Session; + +import org.apache.activemq.TestSupport; +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerFactory; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.command.ActiveMQDestination; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +public class AdvisoryTopicDeletionTest extends TestSupport { + private static final Log LOG = LogFactory.getLog(AdvisoryTopicDeletionTest.class); + + private BrokerService broker; + private Connection connection; + + protected ActiveMQConnectionFactory createConnectionFactory() throws Exception { + return new ActiveMQConnectionFactory("vm://" + getName()); + } + + protected void setUp() throws Exception { + createBroker(); + topic = false; + super.setUp(); + } + + protected void tearDown() throws Exception { + super.tearDown(); + destroyBroker(); + } + + private void createBroker() throws Exception { + broker = BrokerFactory.createBroker("broker:(vm://localhost)"); + broker.setPersistent(false); + broker.setBrokerName(getName()); + broker.start(); + + connection = createConnection(); + } + + @Override + protected Connection createConnection() throws Exception { + Connection con = super.createConnection(); + con.start(); + return con; + } + + private void destroyBroker() throws Exception { + if (connection != null) + connection.close(); + if (broker != null) + broker.stop(); + } + + public void doTest() throws Exception { + Destination dest = createDestination(); + + Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + MessageConsumer consumer = consumerSession.createConsumer(dest); + + MessageProducer prod = producerSession.createProducer(dest); + Message message = producerSession.createMessage(); + prod.send(message); + + consumer.receive(60 * 1000); + connection.close(); + connection = null; + + if ( topic ) { + broker.getAdminView().removeTopic(((ActiveMQDestination)dest).getPhysicalName()); + } else { + broker.getAdminView().removeQueue(((ActiveMQDestination)dest).getPhysicalName()); + } + + ActiveMQDestination dests[] = broker.getRegionBroker().getDestinations(); + int matchingDestinations = 0; + for (ActiveMQDestination destination: dests) { + String name = destination.getPhysicalName(); + LOG.debug("Found destination " + name); + if (name.startsWith("ActiveMQ.Advisory") && name.contains(getDestinationString())) { + matchingDestinations++; + } + } + + assertEquals("No matching destinations should be found", 0, matchingDestinations); + } + + public void testTopic() throws Exception { + topic=true; + doTest(); + } + + public void testQueue() throws Exception { + topic=false; + doTest(); + } +}