https://issues.apache.org/jira/browse/AMQ-3253 - Support Temporary Destinations in a network without advisories. Support hub and spoke, such that two spokes can communicate. Revisit https://issues.apache.org/jira/browse/AMQ-2571 to make creating temp destinations on command configurable broker setAllowTempAutoCreationOnSend=true and allow temp destinations to be gced via gcInactiveDestinations policy

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1087912 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java

@@ -198,6 +198,7 @@ public class BrokerService implements Service {
     private int schedulePeriodForDestinationPurge=5000;
     private BrokerContext brokerContext;
     private boolean networkConnectorStartAsync = false;
+    private boolean allowTempAutoCreationOnSend;
 
 	static {
         String localHostName = "localhost";
@@ -1833,6 +1834,7 @@ public class BrokerService implements Service {
         regionBroker.setKeepDurableSubsActive(keepDurableSubsActive);
         regionBroker.setBrokerName(getBrokerName());
         regionBroker.getDestinationStatistics().setEnabled(enableStatistics);
+        regionBroker.setAllowTempAutoCreationOnSend(isAllowTempAutoCreationOnSend());
         if (brokerId != null) {
             regionBroker.setBrokerId(brokerId);
         }
@@ -2412,4 +2414,20 @@ public class BrokerService implements Service {
     public void setNetworkConnectorStartAsync(boolean networkConnectorStartAsync) {
         this.networkConnectorStartAsync = networkConnectorStartAsync;
     }
+
+    public boolean isAllowTempAutoCreationOnSend() {
+        return allowTempAutoCreationOnSend;
+    }
+
+    /**
+     * enable if temp destinations need to be propagated through a network when
+     * advisorySupport==false. This is used in conjunction with the policy
+     * gcInactiveDestinations for matching temps so they can get removed
+     * when inactive
+     *
+     * @param allowTempAutoCreationOnSend
+     */
+    public void setAllowTempAutoCreationOnSend(boolean allowTempAutoCreationOnSend) {
+        this.allowTempAutoCreationOnSend = allowTempAutoCreationOnSend;
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java

@@ -17,6 +17,7 @@
 package org.apache.activemq.broker.region;
 
 import java.io.IOException;
+import java.util.Collection;
 import javax.jms.ResourceAllocationException;
 import org.apache.activemq.advisory.AdvisorySupport;
 import org.apache.activemq.broker.Broker;
@@ -92,7 +93,7 @@ public abstract class BaseDestination implements Destination {
     private boolean reduceMemoryFootprint = false;
 
     /**
-     * @param broker
+     * @param brokerService
      * @param store
      * @param destination
      * @param parentStats
@@ -241,7 +242,7 @@ public abstract class BaseDestination implements Destination {
         return store;
     }
 
-    public final boolean isActive() {
+    public boolean isActive() {
         return destinationStatistics.getConsumers().getCount() != 0 || destinationStatistics.getProducers().getCount() != 0;
     }
 
@@ -674,4 +675,15 @@ public abstract class BaseDestination implements Destination {
     protected boolean isReduceMemoryFootprint() {
         return this.reduceMemoryFootprint;
     }
+
+   protected boolean hasRegularConsumers(Collection<Subscription> consumers) {
+        boolean hasRegularConsumers = false;
+        for (Subscription subscription: consumers) {
+            if (!subscription.getConsumerInfo().isNetworkSubscription()) {
+                hasRegularConsumers = true;
+                break;
+            }
+        }
+        return hasRegularConsumers;
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java

@@ -104,7 +104,7 @@ public class RegionBroker extends EmptyBroker {
     private ConnectionContext adminConnectionContext;
     private final Scheduler scheduler;
     private final ThreadPoolExecutor executor;
-    
+    private boolean allowTempAutoCreationOnSend;
     private final Runnable purgeInactiveDestinationsTask = new Runnable() {
         public void run() {
             purgeInactiveDestinations();
@@ -499,7 +499,7 @@ public class RegionBroker extends EmptyBroker {
                 || (producerExchange.getRegion() != null && producerExchange.getRegion().getDestinationMap().get(message.getDestination()) == null)) {
             ActiveMQDestination destination = message.getDestination();
             // ensure the destination is registered with the RegionBroker
-            producerExchange.getConnectionContext().getBroker().addDestination(producerExchange.getConnectionContext(), destination,false);
+            producerExchange.getConnectionContext().getBroker().addDestination(producerExchange.getConnectionContext(), destination, isAllowTempAutoCreationOnSend());
             Region region;
             switch (destination.getDestinationType()) {
             case ActiveMQDestination.QUEUE_TYPE:
@@ -935,6 +935,10 @@ public class RegionBroker extends EmptyBroker {
         synchronized (purgeInactiveDestinationsTask) {
             List<BaseDestination> list = new ArrayList<BaseDestination>();
             Map<ActiveMQDestination, Destination> map = getDestinationMap();
+            if (isAllowTempAutoCreationOnSend()) {
+                map.putAll(tempQueueRegion.getDestinationMap());
+                map.putAll(tempTopicRegion.getDestinationMap());
+            }
             long timeStamp = System.currentTimeMillis();
             for (Destination d : map.values()) {
                 if (d instanceof BaseDestination) {
@@ -956,7 +960,7 @@ public class RegionBroker extends EmptyBroker {
                             dest.getName() + " Inactive for longer than " + dest.getInactiveTimoutBeforeGC()
                                     + " ms - removing ...");
                     try {
-                        getRoot().removeDestination(context, dest.getActiveMQDestination(), 0);
+                        getRoot().removeDestination(context, dest.getActiveMQDestination(), isAllowTempAutoCreationOnSend() ? 1 : 0);
                     } catch (Exception e) {
                         LOG.error("Failed to remove inactive destination " + dest, e);
                     }
@@ -964,4 +968,12 @@ public class RegionBroker extends EmptyBroker {
             }
         }
     }
+
+    public boolean isAllowTempAutoCreationOnSend() {
+        return allowTempAutoCreationOnSend;
+    }
+
+    public void setAllowTempAutoCreationOnSend(boolean allowTempAutoCreationOnSend) {
+        this.allowTempAutoCreationOnSend = allowTempAutoCreationOnSend;
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/TempQueue.java

@@ -90,4 +90,15 @@ public class TempQueue extends Queue{
         }
         super.dispose(context);
     }
+
+    @Override
+    public boolean isActive() {
+        boolean isActive = super.isActive();
+        if (isActive && brokerService.isAllowTempAutoCreationOnSend()) {
+            synchronized (consumers) {
+                isActive = hasRegularConsumers(consumers);
+            }
+        }
+        return isActive;
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/TempTopic.java

@@ -69,4 +69,14 @@ public class TempTopic  extends Topic  implements Task{
     public void initialize() {
     }
 
+    @Override
+    public boolean isActive() {
+        boolean isActive = super.isActive();
+        if (isActive && brokerService.isAllowTempAutoCreationOnSend()) {
+            synchronized (consumers) {
+                isActive = hasRegularConsumers(consumers);
+            }
+        }
+        return isActive;
+    }
 }

activemq-core/src/test/java/org/apache/activemq/usecases/NetworkAsyncStartTest.java

@@ -54,7 +54,7 @@ public class NetworkAsyncStartTest extends JmsMultipleBrokersTestSupport {
         LOG.info("starting B transport connector");
         BrokerService brokerB = brokers.get("BrokerB").broker;
         brokerB.addConnector(brokerBUri);
-        brokerC.start();
+        brokerB.start();
 
         assertTrue("got bridge to B", waitForBridgeFormation(brokerA, 1, 0));
         assertTrue("got bridge to B&C", waitForBridgeFormation(brokerA, 1, 1));

activemq-core/src/test/java/org/apache/activemq/usecases/RequestReplyNoAdvisoryNetworkTest.java

@@ -26,6 +26,7 @@ import java.net.URLConnection;
 import java.net.URLStreamHandler;
 import java.net.URLStreamHandlerFactory;
 import java.util.Map;
+import java.util.Vector;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
@@ -51,6 +52,7 @@ import org.slf4j.LoggerFactory;
 public class RequestReplyNoAdvisoryNetworkTest extends JmsMultipleBrokersTestSupport {
     private static final transient Logger LOG = LoggerFactory.getLogger(RequestReplyNoAdvisoryNetworkTest.class);
 
+    Vector<BrokerService> brokers = new Vector<BrokerService>();
     BrokerService a, b;
     ActiveMQQueue sendQ = new ActiveMQQueue("sendQ");
     static final String connectionIdMarker = "ID:marker.";
@@ -118,6 +120,8 @@ public class RequestReplyNoAdvisoryNetworkTest extends JmsMultipleBrokersTestSup
         });
         a = new XBeanBrokerFactory().createBroker(new URI("xbean:" + localProtocolScheme + ":A"));
         b = new XBeanBrokerFactory().createBroker(new URI("xbean:" + localProtocolScheme + ":B"));
+        brokers.add(a);
+        brokers.add(b);
 
         doTestNonAdvisoryNetworkRequestReply();
     }
@@ -127,6 +131,25 @@ public class RequestReplyNoAdvisoryNetworkTest extends JmsMultipleBrokersTestSup
         doTestNonAdvisoryNetworkRequestReply();
     }
 
+    public void testNonAdvisoryNetworkRequestReplyWithPIM() throws Exception {
+        a = configureBroker("A");
+        b = configureBroker("B");
+        BrokerService hub = configureBroker("M");
+        hub.setAllowTempAutoCreationOnSend(true);
+        configureForPiggyInTheMiddle(bridge(a, hub));
+        configureForPiggyInTheMiddle(bridge(b, hub));
+
+        startBrokers();
+
+        waitForBridgeFormation(hub, 2, 0);
+        doTestNonAdvisoryNetworkRequestReply();
+    }
+
+    private void configureForPiggyInTheMiddle(NetworkConnector bridge) {
+        bridge.setDuplex(true);
+        bridge.setNetworkTTL(2);
+    }
+
     public void doTestNonAdvisoryNetworkRequestReply() throws Exception {
 
         waitForBridgeFormation(a, 1, 0);
@@ -141,6 +164,7 @@ public class RequestReplyNoAdvisoryNetworkTest extends JmsMultipleBrokersTestSup
         TextMessage message = sendSession.createTextMessage("1");
         message.setJMSReplyTo(realReplyQ);
         producer.send(message);
+        LOG.info("request sent");
 
         // responder
         ActiveMQConnectionFactory consumerFactory = createConnectionFactory(b);
@@ -149,7 +173,7 @@ public class RequestReplyNoAdvisoryNetworkTest extends JmsMultipleBrokersTestSup
         ActiveMQSession consumerSession = (ActiveMQSession)consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         MessageConsumer consumer = consumerSession.createConsumer(sendQ);
         TextMessage received = (TextMessage) consumer.receive(receiveTimeout);
-        assertNotNull(received);
+        assertNotNull("got request from sender ok", received);
 
         LOG.info("got request, sending reply");
 
@@ -166,16 +190,20 @@ public class RequestReplyNoAdvisoryNetworkTest extends JmsMultipleBrokersTestSup
         assertEquals("text is as expected", "got 1", reply.getText());
         sendConnection.close();
 
-        verifyAllTempQueuesAreGone();
+        LOG.info("checking for dangling temp destinations");
-    }
+        // ensure all temp dests get cleaned up on all brokers
-
+        for (BrokerService brokerService : brokers) {
-    private void verifyAllTempQueuesAreGone() throws Exception {
+            final RegionBroker regionBroker = (RegionBroker) brokerService.getRegionBroker();
-        for (BrokerService brokerService : new BrokerService[]{a, b}) {
+            assertTrue("all temps are gone on " + regionBroker.getBrokerName(), Wait.waitFor(new Wait.Condition(){
-            RegionBroker regionBroker = (RegionBroker) brokerService.getRegionBroker();
+                @Override
-            Map temps = regionBroker.getTempTopicRegion().getDestinationMap();
+                public boolean isSatisified() throws Exception {
-            assertTrue("no temp topics on " + brokerService + ", " + temps, temps.isEmpty());
+                    Map tempTopics = regionBroker.getTempTopicRegion().getDestinationMap();
-            temps = regionBroker.getTempQueueRegion().getDestinationMap();
+                    LOG.info("temp topics on " + regionBroker.getBrokerName() + ", " + tempTopics);
-            assertTrue("no temp queues on " + brokerService + ", " + temps, temps.isEmpty());
+                    Map tempQ = regionBroker.getTempQueueRegion().getDestinationMap();
+                    LOG.info("temp queues on " + regionBroker.getBrokerName() + ", " + tempQ);
+                    return tempQ.isEmpty() && tempTopics.isEmpty();
+                }
+            }));
         }
     }
 
@@ -199,27 +227,30 @@ public class RequestReplyNoAdvisoryNetworkTest extends JmsMultipleBrokersTestSup
         b = configureBroker("B");
         bridge(a, b);
         bridge(b, a);
-        a.start();
+        startBrokers();
-        b.start();
     }
 
-    public void tearDown() throws Exception {
+    private void startBrokers() throws Exception {
-        stop(a);
+        for (BrokerService broker: brokers) {
-        stop(b);
+            broker.start();
-    }
-
-    private void stop(BrokerService broker) throws Exception {
-        if (broker != null) {
-            broker.stop();
         }
     }
 
-    private void bridge(BrokerService from, BrokerService to) throws Exception {
+    public void tearDown() throws Exception {
-        TransportConnector toConnector = to.addConnector("tcp://localhost:0");
+        for (BrokerService broker: brokers) {
+            broker.stop();
+        }
+        brokers.clear();
+    }
+
+
+    private NetworkConnector bridge(BrokerService from, BrokerService to) throws Exception {
+        TransportConnector toConnector = to.getTransportConnectors().get(0);
         NetworkConnector bridge =
                 from.addNetworkConnector("static://" + toConnector.getPublishableConnectString());
         bridge.addStaticallyIncludedDestination(sendQ);
         bridge.addStaticallyIncludedDestination(replyQWildcard);
+        return bridge;
     }
 
     private BrokerService configureBroker(String brokerName) throws Exception {
@@ -232,8 +263,13 @@ public class RequestReplyNoAdvisoryNetworkTest extends JmsMultipleBrokersTestSup
         PolicyMap map = new PolicyMap();
         PolicyEntry tempReplyQPolicy = new PolicyEntry();
         tempReplyQPolicy.setOptimizedDispatch(true);
+        tempReplyQPolicy.setGcInactiveDestinations(true);
+        tempReplyQPolicy.setInactiveTimoutBeforeGC(10*1000);
         map.put(replyQWildcard, tempReplyQPolicy);
         broker.setDestinationPolicy(map);
+
+        broker.addConnector("tcp://localhost:0");
+        brokers.add(broker);
         return broker;
     }
 }