/**
 * 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.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import javax.jms.Destination;
import javax.jms.MessageConsumer;

import org.apache.activemq.JmsMultipleBrokersTestSupport;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.command.Command;
import org.apache.activemq.network.DemandForwardingBridge;
import org.apache.activemq.network.NetworkBridgeConfiguration;
import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.util.MessageIdList;

/**
 * @version $Revision: 1.1.1.1 $
 */
public class TwoBrokerMessageNotSentToRemoteWhenNoConsumerTest extends JmsMultipleBrokersTestSupport {
    protected static final int MESSAGE_COUNT = 10;

    protected List<DemandForwardingBridge> bridges;
    protected AtomicInteger msgDispatchCount;

    /**
     * BrokerA -> BrokerB
     */
    public void testRemoteBrokerHasConsumer() throws Exception {
        // Setup broker networks
        bridgeBrokers("BrokerA", "BrokerB");

        startAllBrokers();

        // Setup destination
        Destination dest = createDestination("TEST.FOO", true);

        // Setup consumers
        MessageConsumer clientA = createConsumer("BrokerA", dest);
        MessageConsumer clientB = createConsumer("BrokerB", dest);

        // Send messages
        sendMessages("BrokerA", dest, MESSAGE_COUNT);

        // Get message count
        MessageIdList msgsA = getConsumerMessages("BrokerA", clientA);
        MessageIdList msgsB = getConsumerMessages("BrokerB", clientB);

        msgsA.waitForMessagesToArrive(MESSAGE_COUNT);
        msgsB.waitForMessagesToArrive(MESSAGE_COUNT);

        assertEquals(MESSAGE_COUNT, msgsA.getMessageCount());
        assertEquals(MESSAGE_COUNT, msgsB.getMessageCount());

        // Check that 10 message dispatch commands are send over the network
        assertEquals(MESSAGE_COUNT, msgDispatchCount.get());
    }

    /**
     * BrokerA -> BrokerB
     */
    public void testRemoteBrokerHasNoConsumer() throws Exception {
        // Setup broker networks
        bridgeBrokers("BrokerA", "BrokerB");

        startAllBrokers();

        // Setup destination
        Destination dest = createDestination("TEST.FOO", true);

        // Setup consumers
        MessageConsumer clientA = createConsumer("BrokerA", dest);

        // Send messages
        sendMessages("BrokerA", dest, MESSAGE_COUNT);

        // Get message count
        MessageIdList msgsA = getConsumerMessages("BrokerA", clientA);

        msgsA.waitForMessagesToArrive(MESSAGE_COUNT);

        assertEquals(MESSAGE_COUNT, msgsA.getMessageCount());

        // Check that no message dispatch commands are send over the network
        assertEquals(0, msgDispatchCount.get());
    }

    protected void bridgeBrokers(BrokerService localBroker, BrokerService remoteBroker) throws Exception {
        List remoteTransports = remoteBroker.getTransportConnectors();
        List localTransports = localBroker.getTransportConnectors();

        URI remoteURI;
        URI localURI;
        if (!remoteTransports.isEmpty() && !localTransports.isEmpty()) {
            remoteURI = ((TransportConnector)remoteTransports.get(0)).getConnectUri();
            localURI = ((TransportConnector)localTransports.get(0)).getConnectUri();

            // Ensure that we are connecting using tcp
            if (remoteURI.toString().startsWith("tcp:") && localURI.toString().startsWith("tcp:")) {
                NetworkBridgeConfiguration config = new NetworkBridgeConfiguration();
                config.setBrokerName(localBroker.getBrokerName());
                DemandForwardingBridge bridge = new DemandForwardingBridge(config, TransportFactory.connect(localURI), TransportFactory.connect(remoteURI)) {
                    protected void serviceLocalCommand(Command command) {
                        if (command.isMessageDispatch()) {
                            // Keep track of the number of message dispatches
                            // through the bridge
                            msgDispatchCount.incrementAndGet();
                        }

                        super.serviceLocalCommand(command);
                    }
                };
                bridges.add(bridge);

                bridge.start();
            } else {
                throw new Exception("Remote broker or local broker is not using tcp connectors");
            }
        } else {
            throw new Exception("Remote broker or local broker has no registered connectors.");
        }

        maxSetupTime = 2000;
    }

    public void setUp() throws Exception {
        super.setAutoFail(true);
        super.setUp();
        createBroker(new URI("broker:(tcp://localhost:61616)/BrokerA?persistent=false&useJmx=false"));
        createBroker(new URI("broker:(tcp://localhost:61617)/BrokerB?persistent=false&useJmx=false"));

        bridges = new ArrayList<DemandForwardingBridge>();
        msgDispatchCount = new AtomicInteger(0);
    }
}

@@ -1347,24 +1347,77 @@
 figuration;%0A
+import org.apache.activemq.network.NetworkConnector;%0A
 import org.a
@@ -1704,107 +1704,9 @@
 = 10
-;%0A%0A    protected List%3CDemandForwardingBridge%3E bridges;%0A    protected AtomicInteger msgDispatchCount
+0
 ;%0A%0A 
@@ -2166,32 +2166,60 @@
 okerB%22, dest);%0A%0A
+        Thread.sleep(2000);%0A
         // Send 
@@ -2692,146 +2692,8 @@
 );%0A%0A
-        // Check that 10 message dispatch commands are send over the network%0A        assertEquals(MESSAGE_COUNT, msgDispatchCount.get());%0A
     
@@ -3407,1963 +3407,15 @@
     
+%7D%0A%0A
     
-// Check that no message dispatch commands are send over the network%0A        assertEquals(0, msgDispatchCount.get());%0A    %7D%0A%0A    protected void bridgeBrokers(BrokerService localBroker, BrokerService remoteBroker) throws Exception %7B%0A        List remoteTransports = remoteBroker.getTransportConnectors();%0A        List localTransports = localBroker.getTransportConnectors();%0A%0A        URI remoteURI;%0A        URI localURI;%0A        if (!remoteTransports.isEmpty() && !localTransports.isEmpty()) %7B%0A            remoteURI = ((TransportConnector)remoteTransports.get(0)).getConnectUri();%0A            localURI = ((TransportConnector)localTransports.get(0)).getConnectUri();%0A%0A            // Ensure that we are connecting using tcp%0A            if (remoteURI.toString().startsWith(%22tcp:%22) && localURI.toString().startsWith(%22tcp:%22)) %7B%0A                NetworkBridgeConfiguration config = new NetworkBridgeConfiguration();%0A                config.setBrokerName(localBroker.getBrokerName());%0A                DemandForwardingBridge bridge = new DemandForwardingBridge(config, TransportFactory.connect(localURI), TransportFactory.connect(remoteURI)) %7B%0A                    protected void serviceLocalCommand(Command command) %7B%0A                        if (command.isMessageDispatch()) %7B%0A                            // Keep track of the number of message dispatches%0A                            // through the bridge%0A                            msgDispatchCount.incrementAndGet();%0A                        %7D%0A%0A                        super.serviceLocalCommand(command);%0A                    %7D%0A                %7D;%0A                bridges.add(bridge);%0A%0A                bridge.start();%0A            %7D else %7B%0A                throw new Exception(%22Remote broker or local broker is not using tcp connectors%22);%0A            %7D%0A        %7D else %7B%0A            throw new Exception(%22Remote broker or local broker has no registered connectors.%22);%0A        %7D%0A%0A        maxSetupTime = 2000;%0A    %7D%0A%0A    
 publ
@@ -3526,32 +3526,49 @@
 eBroker(new URI(
+%0A                
 %22broker:(tcp://l
@@ -3654,16 +3654,33 @@
 new URI(
+%0A                
 %22broker:
@@ -3748,117 +3748,8 @@
 %22));
-%0A%0A        bridges = new ArrayList%3CDemandForwardingBridge%3E();%0A        msgDispatchCount = new AtomicInteger(0);
 %0A   

Correct test case