Return-Path: X-Original-To: apmail-activemq-commits-archive@www.apache.org Delivered-To: apmail-activemq-commits-archive@www.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id F1AA29CF6 for ; Tue, 24 Jan 2012 16:38:28 +0000 (UTC) Received: (qmail 59810 invoked by uid 500); 24 Jan 2012 16:38:28 -0000 Delivered-To: apmail-activemq-commits-archive@activemq.apache.org Received: (qmail 59734 invoked by uid 500); 24 Jan 2012 16:38:28 -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 59727 invoked by uid 99); 24 Jan 2012 16:38:28 -0000 Received: from nike.apache.org (HELO nike.apache.org) (192.87.106.230) by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 24 Jan 2012 16:38:28 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=5.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; Tue, 24 Jan 2012 16:38:25 +0000 Received: from eris.apache.org (localhost [127.0.0.1]) by eris.apache.org (Postfix) with ESMTP id 9138B23889CB for ; Tue, 24 Jan 2012 16:38:03 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r1235341 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/broker/region/TopicRegion.java test/java/org/apache/activemq/bugs/AMQ3674Test.java Date: Tue, 24 Jan 2012 16:38:03 -0000 To: commits@activemq.apache.org From: tabish@apache.org X-Mailer: svnmailer-1.0.8-patched Message-Id: <20120124163803.9138B23889CB@eris.apache.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: tabish Date: Tue Jan 24 16:38:03 2012 New Revision: 1235341 URL: http://svn.apache.org/viewvc?rev=1235341&view=rev Log: Add fix and test for: https://issues.apache.org/jira/browse/AMQ-3674 Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3674Test.java (with props) Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java?rev=1235341&r1=1235340&r2=1235341&view=diff ============================================================================== --- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java (original) +++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java Tue Jan 24 16:38:03 2012 @@ -172,12 +172,14 @@ public class TopicRegion extends Abstrac @Override public void removeSubscription(ConnectionContext context, RemoveSubscriptionInfo info) throws Exception { SubscriptionKey key = new SubscriptionKey(info.getClientId(), info.getSubscriptionName()); - DurableTopicSubscription sub = durableSubscriptions.remove(key); + DurableTopicSubscription sub = durableSubscriptions.get(key); if (sub == null) { throw new InvalidDestinationException("No durable subscription exists for: " + info.getSubscriptionName()); } if (sub.isActive()) { throw new JMSException("Durable consumer is in use"); + } else { + durableSubscriptions.get(key); } destinationsLock.readLock().lock(); Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3674Test.java URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3674Test.java?rev=1235341&view=auto ============================================================================== --- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3674Test.java (added) +++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3674Test.java Tue Jan 24 16:38:03 2012 @@ -0,0 +1,121 @@ +/** + * 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.*; + +import javax.jms.Connection; +import javax.jms.DeliveryMode; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.TopicSubscriber; +import javax.management.ObjectName; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.broker.TransportConnector; +import org.apache.activemq.broker.jmx.BrokerView; +import org.apache.activemq.command.ActiveMQTopic; +import org.apache.activemq.util.Wait; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class AMQ3674Test { + + private static Logger LOG = LoggerFactory.getLogger(AMQ3674Test.class); + + private final static int deliveryMode = DeliveryMode.NON_PERSISTENT; + private final static ActiveMQTopic destination = new ActiveMQTopic("XYZ"); + + private ActiveMQConnectionFactory factory; + private BrokerService broker; + + @Test + public void removeSubscription() throws Exception { + + final Connection producerConnection = factory.createConnection(); + producerConnection.start(); + final Connection consumerConnection = factory.createConnection(); + + consumerConnection.setClientID("subscriber1"); + Session consumerMQSession = consumerConnection.createSession(false, Session.CLIENT_ACKNOWLEDGE); + TopicSubscriber activeConsumer = (TopicSubscriber) consumerMQSession.createDurableSubscriber(destination, "myTopic"); + consumerConnection.start(); + + Session session = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageProducer producer = session.createProducer(destination); + producer.setDeliveryMode(deliveryMode); + + final BrokerView brokerView = broker.getAdminView(); + + assertEquals(1, brokerView.getDurableTopicSubscribers().length); + + LOG.info("Current Durable Topic Subscriptions: " + brokerView.getDurableTopicSubscribers().length); + + try { + brokerView.destroyDurableSubscriber("subscriber1", "myTopic"); + fail("Expected Exception for Durable consumer is in use"); + } catch(Exception e) { + LOG.info("Recieved expected exception: " + e.getMessage()); + } + + LOG.info("Current Durable Topic Subscriptions: " + brokerView.getDurableTopicSubscribers().length); + + assertEquals(1, brokerView.getDurableTopicSubscribers().length); + + activeConsumer.close(); + consumerConnection.stop(); + + assertTrue("The subscription should be in the inactive state.", Wait.waitFor(new Wait.Condition() { + + @Override + public boolean isSatisified() throws Exception { + return brokerView.getInactiveDurableTopicSubscribers().length == 1; + } + })); + + try { + brokerView.destroyDurableSubscriber("subscriber1", "myTopic"); + } finally { + producer.close(); + producerConnection.close(); + } + } + + @Before + public void setUp() throws Exception { + broker = new BrokerService(); + broker.setPersistent(false); + broker.setUseJmx(true); + broker.setDeleteAllMessagesOnStartup(true); + TransportConnector connector = broker.addConnector("tcp://localhost:0"); + broker.start(); + + factory = new ActiveMQConnectionFactory(connector.getPublishableConnectString()); + factory.setAlwaysSyncSend(true); + factory.setDispatchAsync(false); + } + + @After + public void tearDown() throws Exception { + broker.stop(); + } +} Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3674Test.java ------------------------------------------------------------------------------ svn:eol-style = native