diff --git a/activemq-core/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java b/activemq-core/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java index f86d65ec07..dc1e3dfa9e 100755 --- a/activemq-core/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java +++ b/activemq-core/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java @@ -17,6 +17,8 @@ */ package org.apache.activemq.broker.jmx; +import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicInteger; + import javax.management.ObjectName; import org.apache.activemq.broker.Broker; @@ -33,6 +35,7 @@ public class BrokerView implements BrokerViewMBean { final ManagedRegionBroker broker; private final BrokerService brokerService; + private final AtomicInteger sessionIdCounter = new AtomicInteger(0); public BrokerView(BrokerService brokerService, ManagedRegionBroker managedBroker) throws Exception { this.brokerService = brokerService; @@ -156,7 +159,7 @@ public class BrokerView implements BrokerViewMBean { ConsumerInfo info = new ConsumerInfo(); ConsumerId consumerId = new ConsumerId(); consumerId.setConnectionId(clientId); - consumerId.setSessionId(0); + consumerId.setSessionId(sessionIdCounter.incrementAndGet()); consumerId.setValue(0); info.setConsumerId(consumerId); info.setDestination(new ActiveMQTopic(topicName)); diff --git a/activemq-core/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java b/activemq-core/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java index c0ae11478e..071e373ee6 100755 --- a/activemq-core/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java +++ b/activemq-core/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java @@ -43,11 +43,11 @@ import org.apache.commons.logging.LogFactory; import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap; /** - * + * * @version $Revision: 1.14 $ */ abstract public class AbstractRegion implements Region { - + private static final Log log = LogFactory.getLog(AbstractRegion.class); protected final ConcurrentHashMap destinations = new ConcurrentHashMap(); @@ -60,7 +60,8 @@ abstract public class AbstractRegion implements Region { protected boolean autoCreateDestinations=true; protected final TaskRunnerFactory taskRunnerFactory; protected final Object destinationsMutex = new Object(); - + protected final Map consumerChangeMutexMap = new HashMap(); + public AbstractRegion(RegionBroker broker,DestinationStatistics destinationStatistics, UsageManager memoryManager, TaskRunnerFactory taskRunnerFactory, PersistenceAdapter persistenceAdapter) { this.broker = broker; this.destinationStatistics = destinationStatistics; @@ -111,7 +112,7 @@ abstract public class AbstractRegion implements Region { public void removeDestination(ConnectionContext context,ActiveMQDestination destination,long timeout) throws Exception{ - + // No timeout.. then try to shut down right way, fails if there are current subscribers. if( timeout == 0 ) { for(Iterator iter=subscriptions.values().iterator();iter.hasNext();){ @@ -121,19 +122,19 @@ abstract public class AbstractRegion implements Region { } } } - + if( timeout > 0 ) { - // TODO: implement a way to notify the subscribers that we want to take the down + // TODO: implement a way to notify the subscribers that we want to take the down // the destination and that they should un-subscribe.. Then wait up to timeout time before // dropping the subscription. - + } log.debug("Removing destination: "+destination); synchronized(destinationsMutex){ Destination dest=(Destination) destinations.remove(destination); if(dest!=null){ - + // timeout<0 or we timed out, we now force any remaining subscriptions to un-subscribe. for(Iterator iter=subscriptions.values().iterator();iter.hasNext();){ Subscription sub=(Subscription) iter.next(); @@ -141,20 +142,20 @@ abstract public class AbstractRegion implements Region { dest.removeSubscription(context, sub); } } - + destinationMap.removeAll(destination); dest.dispose(context); dest.stop(); - + }else{ log.debug("Destination doesn't exist: " + dest); } } } - + /** * Provide an exact or wildcard lookup of destinations in the region - * + * * @return a set of matching destination objects. */ public Set getDestinations(ActiveMQDestination destination) { @@ -162,7 +163,7 @@ abstract public class AbstractRegion implements Region { return destinationMap.get(destination); } } - + public Map getDestinationMap() { synchronized(destinationsMutex){ return new HashMap(destinations); @@ -177,43 +178,66 @@ abstract public class AbstractRegion implements Region { // lets auto-create the destination lookup(context, destination); } - - Subscription sub = createSubscription(context, info); - // We may need to add some destinations that are in persistent store but not active - // in the broker. - // - // TODO: think about this a little more. This is good cause destinations are not loaded into - // memory until a client needs to use the queue, but a management agent viewing the - // broker will not see a destination that exists in persistent store. We may want to - // eagerly load all destinations into the broker but have an inactive state for the - // destination which has reduced memory usage. - // - if( persistenceAdapter!=null ) { - Set inactiveDests = getInactiveDestinations(); - for (Iterator iter = inactiveDests.iterator(); iter.hasNext();) { - ActiveMQDestination dest = (ActiveMQDestination) iter.next(); - if( sub.matches(dest) ) { - context.getBroker().addDestination(context, dest); - } + Object addGuard; + synchronized(consumerChangeMutexMap) { + addGuard = consumerChangeMutexMap.get(info.getConsumerId()); + if (addGuard == null) { + addGuard = new Object(); + consumerChangeMutexMap.put(info.getConsumerId(), addGuard); } } - - subscriptions.put(info.getConsumerId(), sub); + synchronized (addGuard) { + Object o = subscriptions.get(info.getConsumerId()); + if (o != null) { + log.warn("A duplicate subscription was detected. Clients may be misbehaving. Later warnings you may see about subscription removal are a consequence of this."); + return (Subscription)o; + } - // Add the subscription to all the matching queues. - for (Iterator iter = destinationMap.get(info.getDestination()).iterator(); iter.hasNext();) { - Destination dest = (Destination) iter.next(); - dest.addSubscription(context, sub); - } + Subscription sub = createSubscription(context, info); - if( info.isBrowser() ) { - ((QueueBrowserSubscription)sub).browseDone(); + // We may need to add some destinations that are in persistent store but not active + // in the broker. + // + // TODO: think about this a little more. This is good cause destinations are not loaded into + // memory until a client needs to use the queue, but a management agent viewing the + // broker will not see a destination that exists in persistent store. We may want to + // eagerly load all destinations into the broker but have an inactive state for the + // destination which has reduced memory usage. + // + if( persistenceAdapter!=null ) { + Set inactiveDests = getInactiveDestinations(); + for (Iterator iter = inactiveDests.iterator(); iter.hasNext();) { + ActiveMQDestination dest = (ActiveMQDestination) iter.next(); + if( sub.matches(dest) ) { + context.getBroker().addDestination(context, dest); + } + } + } + + subscriptions.put(info.getConsumerId(), sub); + + // At this point we're done directly manipulating subscriptions, + // but we need to retain the synchronized block here. Consider + // otherwise what would happen if at this point a second + // thread added, then removed, as would be allowed with + // no mutex held. Remove is only essentially run once + // so everything after this point would be leaked. + + // Add the subscription to all the matching queues. + for (Iterator iter = destinationMap.get(info.getDestination()).iterator(); iter.hasNext();) { + Destination dest = (Destination) iter.next(); + dest.addSubscription(context, sub); + } + + if( info.isBrowser() ) { + ((QueueBrowserSubscription)sub).browseDone(); + } + + return sub; } - - return sub; } - + /** * Get all the Destinations that are in storage * @return Set of all stored destinations @@ -230,26 +254,29 @@ abstract public class AbstractRegion implements Region { inactiveDests.removeAll( destinations.keySet() ); return inactiveDests; } - + public void removeConsumer(ConnectionContext context, ConsumerInfo info) throws Exception { - + log.debug("Removing consumer: "+info.getConsumerId()); - + Subscription sub = (Subscription) subscriptions.remove(info.getConsumerId()); if( sub==null ) throw new IllegalArgumentException("The subscription does not exist: "+info.getConsumerId()); - + // remove the subscription from all the matching queues. for (Iterator iter = destinationMap.get(info.getDestination()).iterator(); iter.hasNext();) { Destination dest = (Destination) iter.next(); dest.removeSubscription(context, sub); } - + destroySubscription(sub); - + + synchronized (consumerChangeMutexMap) { + consumerChangeMutexMap.remove(info.getConsumerId()); + } } - protected void destroySubscription(Subscription sub) { + protected void destroySubscription(Subscription sub) { sub.destroy(); } @@ -262,7 +289,7 @@ abstract public class AbstractRegion implements Region { Destination dest = lookup(context, messageSend.getDestination()); dest.send(context, messageSend); } - + public void acknowledge(ConnectionContext context, MessageAck ack) throws Exception { Subscription sub = (Subscription) subscriptions.get(ack.getConsumerId()); if( sub==null ) @@ -295,7 +322,7 @@ abstract public class AbstractRegion implements Region { return dest; } } - + public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Exception{ Subscription sub = (Subscription) subscriptions.get(messageDispatchNotification.getConsumerId()); if (sub != null){ @@ -306,23 +333,23 @@ abstract public class AbstractRegion implements Region { for (Iterator iter = subscriptions.values().iterator(); iter.hasNext();) { Subscription sub = (Subscription) iter.next(); sub.gc(); - } + } for (Iterator iter = destinations.values() .iterator(); iter.hasNext();) { Destination dest = (Destination) iter.next(); dest.gc(); - } + } } protected abstract Subscription createSubscription(ConnectionContext context, ConsumerInfo info) throws Exception; abstract protected Destination createDestination(ConnectionContext context, ActiveMQDestination destination) throws Exception; - public boolean isAutoCreateDestinations() { + public boolean isAutoCreateDestinations() { return autoCreateDestinations; } public void setAutoCreateDestinations(boolean autoCreateDestinations) { this.autoCreateDestinations = autoCreateDestinations; } - + } diff --git a/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java b/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java index 54684a1bc9..5f92663325 100755 --- a/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java +++ b/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java @@ -105,6 +105,10 @@ public class TopicRegion extends AbstractRegion { else { super.addConsumer(context, info); sub = (DurableTopicSubscription) durableSubscriptions.get(key); + if (sub == null) { + throw new JMSException("Cannot use the same consumerId: " + info.getConsumerId() + " for two different durable subscriptions clientID: " + + key.getClientId() + " subscriberName: " + key.getSubscriptionName()); + } } sub.activate(context, info); diff --git a/activemq-core/src/test/java/org/apache/activemq/broker/DoubleSubscriptionTest.java b/activemq-core/src/test/java/org/apache/activemq/broker/DoubleSubscriptionTest.java new file mode 100644 index 0000000000..e03d77125b --- /dev/null +++ b/activemq-core/src/test/java/org/apache/activemq/broker/DoubleSubscriptionTest.java @@ -0,0 +1,118 @@ +/** + * + * 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.broker; + +import javax.jms.DeliveryMode; +import junit.framework.Test; +import org.apache.activemq.broker.StubConnection; +import org.apache.activemq.command.ActiveMQDestination; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.activemq.command.ConnectionInfo; +import org.apache.activemq.command.ConsumerInfo; +import org.apache.activemq.command.Message; +import org.apache.activemq.command.MessageAck; +import org.apache.activemq.command.ProducerInfo; +import org.apache.activemq.command.SessionInfo; +import org.apache.activemq.network.NetworkTestSupport; + +/** + * Pretend to be an abusive client that sends multiple + * identical ConsumerInfo commands and make sure the + * broker doesn't stall because of it. + */ + +public class DoubleSubscriptionTest extends NetworkTestSupport { + + public ActiveMQDestination destination; + public int deliveryMode; + + private String remoteURI = "tcp://localhost:0?wireFormat.tcpNoDelayEnabled=true"; + + public static Test suite() { + return suite(DoubleSubscriptionTest.class); + } + + public static void main(String[] args) { + junit.textui.TestRunner.run(suite()); + } + + public void initCombosForTestDoubleSubscription() { + addCombinationValues("destination", new Object[] { new ActiveMQQueue("TEST"), new ActiveMQQueue("TEST"), }); + } + public void testDoubleSubscription() throws Exception { + + // Start a normal consumer on the remote broker + StubConnection connection1 = createRemoteConnection(); + ConnectionInfo connectionInfo1 = createConnectionInfo(); + SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); + ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); + connection1.send(connectionInfo1); + connection1.send(sessionInfo1); + connection1.request(consumerInfo1); + + // Start a normal producer on a remote broker + StubConnection connection2 = createRemoteConnection(); + ConnectionInfo connectionInfo2 = createConnectionInfo(); + SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); + ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2); + connection2.send(connectionInfo2); + connection2.send(sessionInfo2); + connection2.request(producerInfo2); + + // Send a message to make sure the basics are working + connection2.request(createMessage(producerInfo2, destination, DeliveryMode.PERSISTENT)); + + Message m1 = receiveMessage(connection1); + assertNotNull(m1); + assertNoMessagesLeft(connection1); + + connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE)); + + // Send a message to sit on the broker while we mess with it + connection2.request(createMessage(producerInfo2, destination, DeliveryMode.PERSISTENT)); + + // Now we're going to resend the same consumer commands again and see if the broker + // can handle it. + connection1.send(connectionInfo1); + connection1.send(sessionInfo1); + connection1.request(consumerInfo1); + + // After this there should be 2 messages on the broker... + connection2.request(createMessage(producerInfo2, destination, DeliveryMode.PERSISTENT)); + + // ... let's start a fresh consumer... + connection1.stop(); + StubConnection connection3 = createRemoteConnection(); + ConnectionInfo connectionInfo3 = createConnectionInfo(); + SessionInfo sessionInfo3 = createSessionInfo(connectionInfo3); + ConsumerInfo consumerInfo3 = createConsumerInfo(sessionInfo3, destination); + connection3.send(connectionInfo3); + connection3.send(sessionInfo3); + connection3.request(consumerInfo3); + + // ... and then grab the 2 that should be there. + assertNotNull(receiveMessage(connection3)); + assertNotNull(receiveMessage(connection3)); + assertNoMessagesLeft(connection3); + } + + protected String getRemoteURI() { + return remoteURI; + } + +}