activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gtu...@apache.org
Subject svn commit: r932342 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/broker/jmx/ main/java/org/apache/activemq/broker/region/ main/java/org/apache/activemq/network/ test/java/org/apache/activemq/bugs/ test/java/org/apache/activemq/...
Date Fri, 09 Apr 2010 10:36:00 GMT
Author: gtully
Date: Fri Apr  9 10:36:00 2010
New Revision: 932342

URL: http://svn.apache.org/viewvc?rev=932342&view=rev
Log:
resolve https://issues.apache.org/activemq/browse/AMQ-2691 - ensure suppressed durable subs
are acked so that message can be removed and not recovered at a later date. Tidy up network
bridge creation so that durables are supported by default and allow dynamicOnly to determine
if durables are auto bridged. Avoid duplicate mbeans for durable subs on restart and allow
active to reflect status of a sub

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DurableConduitBridge.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/NetworkBridgeFactory.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2439Test.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/NetworkBrokerDetachTest.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java?rev=932342&r1=932341&r2=932342&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java
Fri Apr  9 10:36:00 2010
@@ -29,6 +29,7 @@ import org.apache.activemq.broker.region
 import org.apache.activemq.broker.region.RegionBroker;
 import org.apache.activemq.broker.region.Subscription;
 import org.apache.activemq.broker.region.Topic;
+import org.apache.activemq.broker.region.TopicRegion;
 import org.apache.activemq.broker.region.TopicSubscription;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQMessage;
@@ -338,14 +339,16 @@ public class ManagedRegionBroker extends
         if (destinations != null) {
             for (Iterator iter = destinations.iterator(); iter.hasNext();) {
                 ActiveMQDestination dest = (ActiveMQDestination)iter.next();
-                if (dest.isTopic()) {
+                if (dest.isTopic()) {                
                     SubscriptionInfo[] infos = destinationFactory.getAllDurableSubscriptions((ActiveMQTopic)dest);
                     if (infos != null) {
                         for (int i = 0; i < infos.length; i++) {
                             SubscriptionInfo info = infos[i];
-                            LOG.debug("Restoring durable subscription: " + info);
                             SubscriptionKey key = new SubscriptionKey(info);
-                            subscriptions.put(key, info);
+                            if (!alreadyKnown(key)) {
+                                LOG.debug("Restoring durable subscription mbean: " + info);
+                                subscriptions.put(key, info);
+                            }
                         }
                     }
                 }
@@ -359,6 +362,15 @@ public class ManagedRegionBroker extends
         }
     }
 
+    private boolean alreadyKnown(SubscriptionKey key) {
+        boolean known = false;
+        known = ((TopicRegion) getTopicRegion()).durableSubscriptionExists(key);
+        if (LOG.isTraceEnabled()) {
+            LOG.trace("Sub with key: " + key + ", " + (known ? "": "not") +  " already registered");
+        }
+        return known;
+    }
+
     protected void addInactiveSubscription(SubscriptionKey key, SubscriptionInfo info) {
         Hashtable map = brokerObjectName.getKeyPropertyList();
         try {

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java?rev=932342&r1=932341&r2=932342&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java
Fri Apr  9 10:36:00 2010
@@ -16,6 +16,7 @@
  */
 package org.apache.activemq.broker.jmx;
 
+import org.apache.activemq.broker.region.DurableTopicSubscription;
 import org.apache.activemq.broker.region.TopicSubscription;
 
 /**
@@ -58,5 +59,15 @@ public class TopicSubscriptionView exten
         }
     }
 
+    @Override
+    public boolean isActive() {
+        if (subscription instanceof DurableTopicSubscription) {
+            return ((DurableTopicSubscription)subscription).isActive();
+        } else {
+            return super.isActive();
+        }
+    }
+
+    
 
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java?rev=932342&r1=932341&r2=932342&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
Fri Apr  9 10:36:00 2010
@@ -283,6 +283,7 @@ public abstract class PrefetchSubscripti
                     MessageId messageId = node.getMessageId();
                     if (ack.getLastMessageId().equals(messageId)) {
                         // this should never be within a transaction
+                        dequeueCounter++;
                         node.getRegionDestination().getDestinationStatistics().getInflight().decrement();
                         destination = node.getRegionDestination();
                         acknowledge(context, ack, node);

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=932342&r1=932341&r2=932342&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
Fri Apr  9 10:36:00 2010
@@ -170,7 +170,7 @@ public class TopicRegion extends Abstrac
             for (int i = 0; i < infos.length; i++) {
 
                 SubscriptionInfo info = infos[i];
-                LOG.debug("Restoring durable subscription: " + infos);
+                LOG.debug("Restoring durable subscription: " + info);
                 SubscriptionKey key = new SubscriptionKey(info);
 
                 // A single durable sub may be subscribing to multiple topics.
@@ -313,4 +313,8 @@ public class TopicRegion extends Abstrac
         this.keepDurableSubsActive = keepDurableSubsActive;
     }
 
+    public boolean durableSubscriptionExists(SubscriptionKey key) {
+        return this.durableSubscriptions.containsKey(key);
+    }
+
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java?rev=932342&r1=932341&r2=932342&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
Fri Apr  9 10:36:00 2010
@@ -30,6 +30,7 @@ import java.util.concurrent.ThreadPoolEx
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+
 import org.apache.activemq.Service;
 import org.apache.activemq.advisory.AdvisorySupport;
 import org.apache.activemq.broker.BrokerService;
@@ -65,6 +66,7 @@ import org.apache.activemq.command.Sessi
 import org.apache.activemq.command.ShutdownInfo;
 import org.apache.activemq.command.WireFormatInfo;
 import org.apache.activemq.filter.DestinationFilter;
+import org.apache.activemq.filter.MessageEvaluationContext;
 import org.apache.activemq.transport.DefaultTransportListener;
 import org.apache.activemq.transport.FutureResponse;
 import org.apache.activemq.transport.ResponseCallback;
@@ -88,9 +90,9 @@ import org.apache.commons.logging.LogFac
  * @version $Revision$
  */
 public abstract class DemandForwardingBridgeSupport implements NetworkBridge, BrokerServiceAware
{
-
     private static final Log LOG = LogFactory.getLog(DemandForwardingBridge.class);
     private static final ThreadPoolExecutor ASYNC_TASKS;
+    protected static final String DURABLE_SUB_PREFIX = "NC-DS_";
     protected final Transport localBroker;
     protected final Transport remoteBroker;
     protected final IdGenerator idGenerator = new IdGenerator();
@@ -677,45 +679,41 @@ public abstract class DemandForwardingBr
                     final MessageDispatch md = (MessageDispatch) command;
                     final DemandSubscription sub = subscriptionMapByLocalId.get(md.getConsumerId());
                     if (sub != null && md.getMessage() != null && sub.incrementOutstandingResponses())
{
-                        // See if this consumer's brokerPath tells us it came from the broker
at the other end
-                        // of the bridge. I think we should be making this decision based
on the message's
-                        // broker bread crumbs and not the consumer's? However, the message's
broker bread
-                        // crumbs are null, which is another matter.   
-                        boolean cameFromRemote = false;
-                        Object consumerInfo = md.getMessage().getDataStructure();
-                        if (consumerInfo != null && (consumerInfo instanceof ConsumerInfo))
-                            cameFromRemote = contains(((ConsumerInfo) consumerInfo).getBrokerPath(),
remoteBrokerInfo.getBrokerId());
-
+                        
+                        if (originallyCameFromRemote(md, sub)) {
+                            if (LOG.isDebugEnabled()) {
+                                LOG.debug(configuration.getBrokerName() + " message not forwarded
to " + remoteBrokerName + " because message came from there or fails networkTTL: " + md.getMessage());
+                            }
+                            // still ack as it may be durable
+                            try {
+                                localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE,
1));
+                            } finally {
+                                sub.decrementOutstandingResponses();
+                            }
+                            return;
+                        }
+                        
                         Message message = configureMessage(md);
                         if (LOG.isDebugEnabled()) {
                             LOG.debug("bridging " + configuration.getBrokerName() + " ->
" + remoteBrokerName + ": " + message);
                         }
-
+                        
                         if (!message.isResponseRequired()) {
-
+                            
                             // If the message was originally sent using async
                             // send, we will preserve that QOS
                             // by bridging it using an async send (small chance
                             // of message loss).
-
                             try {
-                                // Don't send it off to the remote if it originally came
from the remote. 
-                                if (!cameFromRemote) {
-                                    remoteBroker.oneway(message);
-                                } else {
-                                    if (LOG.isDebugEnabled()) {
-                                        LOG.debug("Message not forwarded on to remote, because
message came from remote");
-                                    }
-                                }
-
+                                remoteBroker.oneway(message);
                                 localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE,
1));
                                 dequeueCounter.incrementAndGet();
                             } finally {
                                 sub.decrementOutstandingResponses();
                             }
-
+                            
                         } else {
-
+                            
                             // The message was not sent using async send, so we
                             // should only ack the local
                             // broker when we get confirmation that the remote
@@ -730,8 +728,7 @@ public abstract class DemandForwardingBr
                                         } else {
                                             localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE,
1));
                                             dequeueCounter.incrementAndGet();
-
-                                        }
+                                        }   
                                     } catch (IOException e) {
                                         serviceLocalException(e);
                                     } finally {
@@ -739,10 +736,10 @@ public abstract class DemandForwardingBr
                                     }
                                 }
                             };
-
+                            
                             remoteBroker.asyncRequest(message, callback);
+                            
                         }
-
                     } else {
                         if (LOG.isDebugEnabled()) {
                             LOG.debug("No subscription registered with this network bridge
for consumerId " + md.getConsumerId() + " for message: " + md.getMessage());
@@ -779,6 +776,27 @@ public abstract class DemandForwardingBr
         }
     }
 
+    private boolean originallyCameFromRemote(MessageDispatch md, DemandSubscription sub)
throws Exception {
+        // See if this consumer's brokerPath tells us it came from the broker at the other
end
+        // of the bridge. I think we should be making this decision based on the message's
+        // broker bread crumbs and not the consumer's? However, the message's broker bread
+        // crumbs are null, which is another matter.   
+        boolean cameFromRemote = false;
+        Object consumerInfo = md.getMessage().getDataStructure();
+        if (consumerInfo != null && (consumerInfo instanceof ConsumerInfo)) {
+            cameFromRemote = contains(((ConsumerInfo) consumerInfo).getBrokerPath(), remoteBrokerInfo.getBrokerId());
+        }
+        
+        // for durable subs, suppression via filter leaves dangling acks so we need to 
+        // check here and allow the ack irrespective
+        if (!cameFromRemote && sub.getLocalInfo().isDurable()) {
+            MessageEvaluationContext messageEvalContext = new MessageEvaluationContext();
+            messageEvalContext.setMessageReference(md.getMessage());
+            cameFromRemote = !createNetworkBridgeFilter(null).matches(messageEvalContext);
+        }  
+        return cameFromRemote;
+    }
+
     /**
      * @return Returns the dynamicallyIncludedDestinations.
      */
@@ -1130,9 +1148,14 @@ public abstract class DemandForwardingBr
         subscriptionMapByLocalId.put(sub.getLocalInfo().getConsumerId(), sub);
         subscriptionMapByRemoteId.put(sub.getRemoteInfo().getConsumerId(), sub);
 
-        // This works for now since we use a VM connection to the local broker.
-        // may need to change if we ever subscribe to a remote broker.
-        sub.getLocalInfo().setAdditionalPredicate(createNetworkBridgeFilter(info));
+        if (!info.isDurable()) {
+            // This works for now since we use a VM connection to the local broker.
+            // may need to change if we ever subscribe to a remote broker.
+            sub.getLocalInfo().setAdditionalPredicate(createNetworkBridgeFilter(info));
+        } else  {
+            // need to ack this message if it is ignored as it is durable so
+            // we check before we send. see: originallyCameFromRemote()
+        }
     }
 
     protected void removeDemandSubscription(ConsumerId id) throws IOException {

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DurableConduitBridge.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DurableConduitBridge.java?rev=932342&r1=932341&r2=932342&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DurableConduitBridge.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DurableConduitBridge.java
Fri Apr  9 10:36:00 2010
@@ -54,7 +54,7 @@ public class DurableConduitBridge extend
      */
     protected void setupStaticDestinations() {
         super.setupStaticDestinations();
-        ActiveMQDestination[] dests = durableDestinations;
+        ActiveMQDestination[] dests = configuration.isDynamicOnly() ? null : durableDestinations;
         if (dests != null) {
             for (int i = 0; i < dests.length; i++) {
                 ActiveMQDestination dest = dests[i];
@@ -96,7 +96,7 @@ public class DurableConduitBridge extend
     }
 
     protected String getSubscriberName(ActiveMQDestination dest) {
-        String subscriberName = configuration.getBrokerName() + "_" + dest.getPhysicalName();
+        String subscriberName = DURABLE_SUB_PREFIX + configuration.getBrokerName() + "_"
+ dest.getPhysicalName();
         return subscriberName;
     }
 

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/NetworkBridgeFactory.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/NetworkBridgeFactory.java?rev=932342&r1=932341&r2=932342&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/NetworkBridgeFactory.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/NetworkBridgeFactory.java
Fri Apr  9 10:36:00 2010
@@ -55,11 +55,8 @@ public final class NetworkBridgeFactory 
                                                       final NetworkBridgeListener listener)
{
         DemandForwardingBridge result = null;
         if (configuration.isConduitSubscriptions()) {
-            if (configuration.isDynamicOnly()) {
-                result = new ConduitBridge(configuration, localTransport, remoteTransport);
-            } else {
-                result = new DurableConduitBridge(configuration, localTransport, remoteTransport);
-            }
+            // dynamicOnly determines whether durables are auto bridged
+            result = new DurableConduitBridge(configuration, localTransport, remoteTransport);
         } else {
             result = new DemandForwardingBridge(configuration, localTransport, remoteTransport);
         }

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2439Test.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2439Test.java?rev=932342&r1=932341&r2=932342&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2439Test.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2439Test.java Fri
Apr  9 10:36:00 2010
@@ -28,8 +28,11 @@ import javax.jms.Session;
 import org.apache.activemq.JmsMultipleBrokersTestSupport;
 import org.apache.activemq.broker.jmx.BrokerView;
 import org.apache.activemq.util.Wait;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 public class AMQ2439Test extends JmsMultipleBrokersTestSupport {
+    private static final Log LOG = LogFactory.getLog(AMQ2439Test.class);
     Destination dest;
 
     
@@ -45,6 +48,7 @@ public class AMQ2439Test extends JmsMult
        
        assertTrue("dequeue is correct", Wait.waitFor(new Wait.Condition() {
            public boolean isSatisified() throws Exception {
+               LOG.info("dequeue count (want 1000), is : " + brokerView.getTotalDequeueCount());
                return 1000 == brokerView.getTotalDequeueCount();
            }
        }));

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/NetworkBrokerDetachTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/NetworkBrokerDetachTest.java?rev=932342&r1=932341&r2=932342&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/NetworkBrokerDetachTest.java
(original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/NetworkBrokerDetachTest.java
Fri Apr  9 10:36:00 2010
@@ -19,11 +19,18 @@ package org.apache.activemq.network;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assume.assumeNotNull;
 
+import java.io.File;
 import java.net.MalformedURLException;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageListener;
 import javax.jms.Session;
+import javax.jms.TopicSubscriber;
 import javax.management.InstanceNotFoundException;
 import javax.management.MBeanServerConnection;
 import javax.management.ObjectName;
@@ -35,6 +42,9 @@ import org.apache.activemq.ActiveMQConne
 import org.apache.activemq.ActiveMQPrefetchPolicy;
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
 import org.apache.activemq.util.Wait;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -44,28 +54,54 @@ public class NetworkBrokerDetachTest {
 
 	private final static String BROKER_NAME = "broker";
 	private final static String REM_BROKER_NAME = "networkedBroker";
-	private final static String QUEUE_NAME = "testQ";
+	private final static String DESTINATION_NAME = "testQ";
 	private final static int    NUM_CONSUMERS = 1;
 	
     protected static final Log LOG = LogFactory.getLog(NetworkBrokerDetachTest.class);
     protected final int numRestarts = 3;
+    protected final int networkTTL = 2;
+    protected final boolean dynamicOnly = false;
 
     protected BrokerService createBroker() throws Exception {
         BrokerService broker = new BrokerService();
         broker.setBrokerName(BROKER_NAME);
+        configureBroker(broker);
         broker.addConnector("tcp://localhost:61617");
         NetworkConnector networkConnector = broker.addNetworkConnector("static:(tcp://localhost:62617?wireFormat.maxInactivityDuration=500)?useExponentialBackOff=false");
-        networkConnector.setDuplex(false);
+        configureNetworkConnector(networkConnector);
         return broker;
     }
-    
+
     protected BrokerService createNetworkedBroker() throws Exception {
         BrokerService broker = new BrokerService();
         broker.setBrokerName(REM_BROKER_NAME);
+        configureBroker(broker);
+        broker.getManagementContext().setCreateConnector(false);
         broker.addConnector("tcp://localhost:62617");
+        NetworkConnector networkConnector = broker.addNetworkConnector("static:(tcp://localhost:61617?wireFormat.maxInactivityDuration=500)?useExponentialBackOff=false");
+        configureNetworkConnector(networkConnector);
         return broker;
     }
     
+    private void configureNetworkConnector(NetworkConnector networkConnector) {
+        networkConnector.setDuplex(false);
+        networkConnector.setNetworkTTL(networkTTL);
+        networkConnector.setDynamicOnly(dynamicOnly);
+    }
+    
+    // variants for each store....
+    private void configureBroker(BrokerService broker) throws Exception {
+        //KahaPersistenceAdapter persistenceAdapter = new KahaPersistenceAdapter();
+        //persistenceAdapter.setDirectory(new File("target/activemq-data/kaha/" + broker.getBrokerName()
+ "/NetworBrokerDetatchTest"));
+        //broker.setPersistenceAdapter(persistenceAdapter);        
+        
+        KahaDBPersistenceAdapter persistenceAdapter = new KahaDBPersistenceAdapter();
+        persistenceAdapter.setDirectory(new File("target/activemq-data/kahadb/NetworBrokerDetatchTest"));
+        broker.setPersistenceAdapter(persistenceAdapter);
+        
+        // default AMQ
+    }
+
     @Test
     public void testNetworkedBrokerDetach() throws Exception {
         BrokerService broker = createBroker();
@@ -79,29 +115,13 @@ public class NetworkBrokerDetachTest {
         ConnectionFactory consFactory = createConnectionFactory(networkedBroker);
         Connection consConn = consFactory.createConnection();
         Session consSession = consConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        
+        ActiveMQDestination destination = (ActiveMQDestination) consSession.createQueue(DESTINATION_NAME);
         for(int i=0; i<NUM_CONSUMERS; i++) {
-          consSession.createConsumer(consSession.createQueue(QUEUE_NAME));
+            consSession.createConsumer(destination);
         }
         
-        assertTrue("got expected consumer count from mbean within time limit", Wait.waitFor(new
Wait.Condition() {
-
-            public boolean isSatisified() throws Exception {
-                boolean result = false;
-                MBeanServerConnection mbsc = getMBeanServerConnection();
-                if (mbsc != null) {                
-                    // We should have 1 consumer for the queue on the local broker
-                    Object consumers = getAttribute(mbsc, "Queue", "Destination=" + QUEUE_NAME,
"ConsumerCount");
-                    if (consumers != null) {
-                        LOG.info("Consumers for " + QUEUE_NAME + " on " + BROKER_NAME + "
: " + consumers);
-                        if (1L == ((Long)consumers).longValue()) {
-                            result = true;
-                        }
-                    }
-                }
-                return result;
-            }      
-        }));
+        assertTrue("got expected consumer count from mbean within time limit", 
+                verifyConsumerCount(1, destination, BROKER_NAME));
         
         
         LOG.info("Stopping Consumer on the networked broker ...");
@@ -109,24 +129,7 @@ public class NetworkBrokerDetachTest {
         consConn.close();
         
         // We should have 0 consumer for the queue on the local broker
-        assertTrue("got expected 0 count from mbean within time limit", Wait.waitFor(new
Wait.Condition() {
-
-            public boolean isSatisified() throws Exception {
-                boolean result = false;
-                MBeanServerConnection mbsc = getMBeanServerConnection();
-                if (mbsc != null) {                
-                    // We should have 1 consumer for the queue on the local broker
-                    Object consumers = getAttribute(mbsc, "Queue", "Destination=" + QUEUE_NAME,
"ConsumerCount");
-                    if (consumers != null) {
-                        LOG.info("Consumers for " + QUEUE_NAME + " on " + BROKER_NAME + "
: " + consumers);
-                        if (0L == ((Long)consumers).longValue()) {
-                            result = true;
-                        }
-                    }
-                }
-                return result;
-            }      
-        }));
+        assertTrue("got expected 0 count from mbean within time limit", verifyConsumerCount(0,
destination, BROKER_NAME));
         
         networkedBroker.stop();
         networkedBroker.waitUntilStopped();
@@ -134,6 +137,106 @@ public class NetworkBrokerDetachTest {
         broker.waitUntilStopped();
     }
 
+    
+    @Test
+    public void testNetworkedBrokerDurableSubAfterRestart() throws Exception {        
+        BrokerService brokerOne = createBroker();
+        brokerOne.setDeleteAllMessagesOnStartup(true);
+        brokerOne.start();
+
+        BrokerService brokerTwo = createNetworkedBroker();
+        brokerTwo.setDeleteAllMessagesOnStartup(true);
+        brokerTwo.start();
+        
+        final AtomicInteger count = new AtomicInteger(0);
+        MessageListener counter = new MessageListener() {
+            public void onMessage(Message message) {
+                count.incrementAndGet();
+            }
+        };
+        
+        LOG.info("Creating durable consumer on each broker ...");
+        ActiveMQTopic destination = registerDurableConsumer(brokerTwo, counter);
+        registerDurableConsumer(brokerOne, counter);
+        
+        assertTrue("got expected consumer count from local broker mbean within time limit",
+                verifyConsumerCount(2, destination, BROKER_NAME));
+        
+        assertTrue("got expected consumer count from network broker mbean within time limit",
+                verifyConsumerCount(2, destination, REM_BROKER_NAME));
+        
+        sendMessageTo(destination, brokerOne);
+        
+        assertTrue("Got one message on each", verifyMessageCount(2, count));
+        
+        LOG.info("Stopping brokerTwo...");
+        brokerTwo.stop();
+        brokerTwo.waitUntilStopped();           
+        
+        LOG.info("restarting  broker Two...");
+        brokerTwo = createNetworkedBroker();
+        brokerTwo.start();
+   
+        LOG.info("Recreating durable Consumer on the broker after restart...");
+        registerDurableConsumer(brokerTwo, counter);
+        
+        // give advisories a chance to percolate
+        TimeUnit.SECONDS.sleep(5);
+        
+        sendMessageTo(destination, brokerOne);
+        
+        // expect similar after restart
+        assertTrue("got expected consumer count from local broker mbean within time limit",
+                verifyConsumerCount(2, destination, BROKER_NAME));
+ 
+        // a durable sub is auto bridged on restart unless dynamicOnly=true
+        assertTrue("got expected consumer count from network broker mbean within time limit",
+                verifyConsumerCount(2, destination, REM_BROKER_NAME));
+
+        assertTrue("got no inactive subs on broker", verifyDurableConsumerCount(0, BROKER_NAME));
+        assertTrue("got no inactive subs on other broker", verifyDurableConsumerCount(0,
REM_BROKER_NAME));
+
+        assertTrue("Got two more messages after restart", verifyMessageCount(4, count));
+        TimeUnit.SECONDS.sleep(1);
+        assertTrue("still Got just two more messages", verifyMessageCount(4, count));
+        
+        brokerTwo.stop();
+        brokerTwo.waitUntilStopped();
+        brokerOne.stop();
+        brokerOne.waitUntilStopped();
+    }
+
+    private boolean verifyMessageCount(final int i, final AtomicInteger count) throws Exception
{
+        return Wait.waitFor(new Wait.Condition() {
+            public boolean isSatisified() throws Exception {
+                return i == count.get();
+            }      
+        });
+    }
+
+    private ActiveMQTopic registerDurableConsumer(
+            BrokerService brokerService, MessageListener listener) throws Exception {
+        ConnectionFactory factory = createConnectionFactory(brokerService);
+        Connection connection = factory.createConnection();
+        connection.setClientID("DurableOne");
+        connection.start();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        ActiveMQTopic destination = (ActiveMQTopic) session.createTopic(DESTINATION_NAME);
+        // unique to a broker
+        TopicSubscriber sub = session.createDurableSubscriber(destination, "SubOne" + brokerService.getBrokerName());
+        sub.setMessageListener(listener);
+        return destination;
+    }
+
+    private void sendMessageTo(ActiveMQTopic destination, BrokerService brokerService) throws
Exception {
+        ConnectionFactory factory = createConnectionFactory(brokerService);
+        Connection conn = factory.createConnection();
+        conn.start();
+        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        session.createProducer(destination).send(session.createTextMessage("Hi"));
+        conn.close();
+    }
+    
     protected ConnectionFactory createConnectionFactory(final BrokerService broker) throws
Exception {
         
         String url = ((TransportConnector) broker.getTransportConnectors().get(0)).getServer().getConnectURI().toString();
@@ -154,6 +257,46 @@ public class NetworkBrokerDetachTest {
     }
     
     // JMX Helper Methods 
+    private boolean verifyConsumerCount(final long expectedCount, final ActiveMQDestination
destination, final String brokerName) throws Exception {
+        return Wait.waitFor(new Wait.Condition() {
+            public boolean isSatisified() throws Exception {
+                boolean result = false;
+                MBeanServerConnection mbsc = getMBeanServerConnection();
+                if (mbsc != null) {                
+                    // We should have 1 consumer for the queue on the local broker
+                    Object consumers = getAttribute(mbsc, brokerName, destination.isQueue()
? "Queue" : "Topic", "Destination=" + destination.getPhysicalName(), "ConsumerCount");
+                    if (consumers != null) {
+                        LOG.info("Consumers for " + destination.getPhysicalName() + " on
" + brokerName + " : " + consumers);
+                        if (expectedCount == ((Long)consumers).longValue()) {
+                            result = true;
+                        }
+                    }
+                }
+                return result;
+            }      
+        });
+    }
+    
+    
+    private boolean verifyDurableConsumerCount(final long expectedCount, final String brokerName)
throws Exception {
+        return Wait.waitFor(new Wait.Condition() {
+            public boolean isSatisified() throws Exception {
+                boolean result = false;
+                MBeanServerConnection mbsc = getMBeanServerConnection();
+                if (mbsc != null) {
+                    Set subs = getMbeans(mbsc, brokerName, "Subscription", "active=false,*");
+                    if (subs != null) {
+                        LOG.info("inactive durable subs on " + brokerName + " : " + subs);
+                        if (expectedCount == subs.size()) {
+                            result = true;
+                        }
+                    }
+                }
+                return result;
+            }      
+        });
+    }
+
     
     private MBeanServerConnection getMBeanServerConnection() throws MalformedURLException
{
         final JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi");
@@ -171,10 +314,21 @@ public class NetworkBrokerDetachTest {
         return mbsc;
     }
     
-    private Object getAttribute(MBeanServerConnection mbsc, String type, String pattern,
String attrName) throws Exception {
+    
+    private Set getMbeans(MBeanServerConnection mbsc, String brokerName, String type, String
pattern) throws Exception {
+        Set obj = null;
+        try {
+            obj = mbsc.queryMBeans(getObjectName(brokerName, type, pattern), null);
+        } catch (InstanceNotFoundException ignored) {
+            LOG.warn("getAttribute ex: " + ignored);
+        }
+        return obj;
+    }
+    
+    private Object getAttribute(MBeanServerConnection mbsc, String brokerName, String type,
String pattern, String attrName) throws Exception {
         Object obj = null;
         try {
-            obj = mbsc.getAttribute(getObjectName(BROKER_NAME, type, pattern), attrName);
+            obj = mbsc.getAttribute(getObjectName(brokerName, type, pattern), attrName);
         } catch (InstanceNotFoundException ignored) {
             LOG.warn("getAttribute ex: " + ignored);
         }



Mime
View raw message