From commits-return-13758-apmail-activemq-commits-archive=activemq.apache.org@activemq.apache.org Fri Jun 11 14:47:27 2010 Return-Path: Delivered-To: apmail-activemq-commits-archive@www.apache.org Received: (qmail 51403 invoked from network); 11 Jun 2010 14:47:27 -0000 Received: from unknown (HELO mail.apache.org) (140.211.11.3) by 140.211.11.9 with SMTP; 11 Jun 2010 14:47:27 -0000 Received: (qmail 52054 invoked by uid 500); 11 Jun 2010 14:47:27 -0000 Delivered-To: apmail-activemq-commits-archive@activemq.apache.org Received: (qmail 52008 invoked by uid 500); 11 Jun 2010 14:47:26 -0000 Mailing-List: contact commits-help@activemq.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@activemq.apache.org Delivered-To: mailing list commits@activemq.apache.org Received: (qmail 52001 invoked by uid 99); 11 Jun 2010 14:47:26 -0000 Received: from nike.apache.org (HELO nike.apache.org) (192.87.106.230) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 11 Jun 2010 14:47:26 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [140.211.11.4] (HELO eris.apache.org) (140.211.11.4) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 11 Jun 2010 14:47:24 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 0D8432388A60; Fri, 11 Jun 2010 14:46:41 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r953708 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/broker/region/Topic.java test/java/org/apache/activemq/usecases/DurableUnsubscribeTest.java Date: Fri, 11 Jun 2010 14:46:40 -0000 To: commits@activemq.apache.org From: dejanb@apache.org X-Mailer: svnmailer-1.0.8 Message-Id: <20100611144641.0D8432388A60@eris.apache.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: dejanb Date: Fri Jun 11 14:46:40 2010 New Revision: 953708 URL: http://svn.apache.org/viewvc?rev=953708&view=rev Log: https://issues.apache.org/activemq/browse/AMQ-2769 - durable sub unsubscribe Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DurableUnsubscribeTest.java Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java?rev=953708&r1=953707&r2=953708&view=diff ============================================================================== --- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java (original) +++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java Fri Jun 11 14:46:40 2010 @@ -157,13 +157,16 @@ public class Topic extends BaseDestinati sub.remove(context, this); } - public void deleteSubscription(ConnectionContext context, SubscriptionKey key) throws IOException { + public void deleteSubscription(ConnectionContext context, SubscriptionKey key) throws Exception { if (topicStore != null) { topicStore.deleteSubscription(key.clientId, key.subscriptionName); - Object removed = durableSubcribers.remove(key); + DurableTopicSubscription removed = durableSubcribers.remove(key); if (removed != null) { destinationStatistics.getConsumers().decrement(); } + // deactivate and remove + removed.deactivate(false); + consumers.remove(removed); } } Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DurableUnsubscribeTest.java URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DurableUnsubscribeTest.java?rev=953708&view=auto ============================================================================== --- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DurableUnsubscribeTest.java (added) +++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DurableUnsubscribeTest.java Fri Jun 11 14:46:40 2010 @@ -0,0 +1,109 @@ +/** + * 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 java.lang.management.ManagementFactory; + +import javax.jms.Connection; +import javax.jms.Session; +import javax.management.MBeanServer; +import javax.management.ObjectName; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerFactory; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.broker.region.Destination; +import org.apache.activemq.command.ActiveMQTopic; + +public class DurableUnsubscribeTest extends org.apache.activemq.TestSupport { + + private BrokerService broker; + private Connection connection; + private ActiveMQTopic topic; + + public void testUnsubscribe() throws Exception { + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + session.createDurableSubscriber(topic, "SubsId"); + session.close(); + + Destination d = broker.getDestination(topic); + assertEquals("Subscription is missing.", 1, d.getConsumers().size()); + + session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + session.unsubscribe("SubsId"); + session.close(); + + assertEquals("Subscription exists.", 0, d.getConsumers().size()); + } + + public void testDestroy() throws Exception { + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + session.createDurableSubscriber(topic, "SubsId2"); + session.close(); + + connection.close(); + connection = null; + Thread.sleep(1000); + + Destination d = broker.getDestination(topic); + assertEquals("Subscription is missing.", 1, d.getConsumers().size()); + + MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); + ObjectName[] subNames = broker.getAdminView().getDurableTopicSubscribers(); + mbs.invoke(subNames[0], "destroy", new Object[0], new String[0]); + + assertEquals("Subscription exists.", 0, d.getConsumers().size()); + } + + protected ActiveMQConnectionFactory createConnectionFactory() throws Exception { + return new ActiveMQConnectionFactory("vm://" + getName()); + } + + protected Connection createConnection() throws Exception { + Connection rc = super.createConnection(); + rc.setClientID(getName()); + return rc; + } + + protected void setUp() throws Exception { + topic = (ActiveMQTopic) createDestination(); + createBroker(); + 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.setUseJmx(true); + broker.setBrokerName(getName()); + broker.start(); + + connection = createConnection(); + } + + private void destroyBroker() throws Exception { + if (connection != null) + connection.close(); + if (broker != null) + broker.stop(); + } +}