/**
 * 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.network;

import java.io.File;
import java.lang.reflect.Field;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;

import jakarta.jms.Connection;
import jakarta.jms.MessageConsumer;
import jakarta.jms.Session;

import junit.framework.AssertionFailedError;
import junit.framework.Test;
import org.apache.activemq.JmsMultipleBrokersTestSupport;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnection;
import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.broker.region.Destination;
import org.apache.activemq.broker.region.DestinationFilter;
import org.apache.activemq.broker.region.DurableTopicSubscription;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.util.SubscriptionKey;
import org.apache.activemq.util.Wait;

/**
 * Test to make sure durable subscriptions propagate properly throughout network bridges
 * and that conduit subscriptions work properly
 */
public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSupport {

    private boolean duplex = true;
    private boolean deletePersistentMessagesOnStartup = true;

    @Override
    protected NetworkConnector bridgeBrokers(String localBrokerName, String remoteBrokerName) throws Exception {
        return bridgeBrokers(localBrokerName, remoteBrokerName, false, -1);
    }

    protected NetworkConnector bridgeBrokers(String localBrokerName, String remoteBrokerName,
                                             boolean dynamicOnly, int networkTTL) throws Exception {
        final NetworkConnector connector = super.bridgeBrokers(localBrokerName, remoteBrokerName);
        final ArrayList<ActiveMQDestination> includedDestinations = new ArrayList<>();
        includedDestinations.add(new ActiveMQTopic("TEST.FOO?forceDurable=true"));
        connector.setDynamicallyIncludedDestinations(includedDestinations);
        connector.setDuplex(duplex);
        connector.setDecreaseNetworkConsumerPriority(false);
        connector.setConduitSubscriptions(true);
        connector.setSyncDurableSubs(true);
        connector.setDynamicOnly(dynamicOnly);
        connector.setNetworkTTL(networkTTL);
        connector.setClientIdToken("|");
        return connector;
    }

    public void testDurablePropagationBrokerRestart() throws Exception {
        duplex = true;

        // Setup broker networks
        bridgeBrokers("Broker_A_A", "Broker_B_B");
        bridgeBrokers("Broker_B_B", "Broker_C_C");
        bridgeBrokers("Broker_C_C", "Broker_D_D");
        bridgeBrokers("Broker_D_D", "Broker_E_E");

        startAllBrokers();
        waitForBridgeFormation();

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

        // Setup consumers
        Connection conn = brokers.get("Broker_A_A").factory.createConnection();
        conn.setClientID("clientId1");
        conn.start();
        Session ses = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        final MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");
        final MessageConsumer clientA2 = ses.createDurableSubscriber(dest, "subA2");

        // let consumers propagate around the network
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);

        //bring online a consumer on the other side
        Connection conn2 = brokers.get("Broker_E_E").factory.createConnection();
        conn2.setClientID("clientId2");
        conn2.start();
        Session ses2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
        MessageConsumer clientE = ses2.createDurableSubscriber(dest, "subE");
        final MessageConsumer clientE2 = ses2.createDurableSubscriber(dest, "subE2");

        // let consumers propagate around the network
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);

        clientA.close();
        clientA2.close();
        clientE.close();
        clientE2.close();

        this.destroyAllBrokers();
        deletePersistentMessagesOnStartup = false;
        final String options = "?persistent=true&useJmx=false";
        createBroker(new URI("broker:(tcp://localhost:0)/Broker_A_A" + options));
        createBroker(new URI("broker:(tcp://localhost:0)/Broker_B_B" + options));
        createBroker(new URI("broker:(tcp://localhost:0)/Broker_C_C" + options));
        createBroker(new URI("broker:(tcp://localhost:0)/Broker_D_D" + options));
        createBroker(new URI("broker:(tcp://localhost:0)/Broker_E_E" + options));
        bridgeBrokers("Broker_A_A", "Broker_B_B");
        bridgeBrokers("Broker_B_B", "Broker_C_C");
        bridgeBrokers("Broker_C_C", "Broker_D_D");
        bridgeBrokers("Broker_D_D", "Broker_E_E");

        startAllBrokers();
        waitForBridgeFormation();

        // Wait for the async durable sync (syncDurableSubs=true) to complete across all bridges.
        // After restart with persistent data, NC durable subs must re-establish to match Phase 1
        // counts before we proceed with unsubscribes, otherwise the sync can re-create NC durable
        // subs AFTER the unsubscribe advisory has already propagated.
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);

        // Wait for all bridge sync executors to finish populating durableRemoteSubs.
        // setupStaticDestinations() creates DemandSubscriptions with empty durableRemoteSubs,
        // and the syncExecutor populates them asynchronously. Without this wait, unsubscribe
        // advisories may race with the sync executor: the advisory finds nothing to remove
        // (durableRemoteSubs is empty), then the sync populates it, leaving a stale NC durable sub.
        waitForAllBridgeSyncCompletion();

        conn = brokers.get("Broker_A_A").factory.createConnection();
        conn.setClientID("clientId1");
        conn.start();
        ses = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        conn2 = brokers.get("Broker_E_E").factory.createConnection();
        conn2.setClientID("clientId2");
        conn2.start();
        ses2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //bring one online and leave others offline to test mixed
        clientE = ses2.createDurableSubscriber(dest, "subE");
        clientE.close();

        ses.unsubscribe("subA");
        ses.unsubscribe("subA2");
        ses2.unsubscribe("subE");
        ses2.unsubscribe("subE2");

        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 0);
    }

    public void testDurablePropagationDuplex() throws Exception {
        duplex = true;
        testDurablePropagation();
    }

    public void testDurablePropagationOneWay() throws Exception {
        duplex = false;
        testDurablePropagation();
    }

    /**
     * Broker_A_A -> Broker_B_B -> Broker_C_C
     */
    protected void testDurablePropagation() throws Exception {
        // Setup broker networks
        bridgeBrokers("Broker_A_A", "Broker_B_B");
        bridgeBrokers("Broker_B_B", "Broker_C_C");
        if (!duplex) {
            bridgeBrokers("Broker_B_B", "Broker_A_A");
            bridgeBrokers("Broker_C_C", "Broker_B_B");
        }

        startAllBrokers();
        waitForBridgeFormation();

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

        // Setup consumers
        final Session ses = createSession("Broker_A_A");
        final MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");
        final MessageConsumer clientB = ses.createDurableSubscriber(dest, "subB");

        // let consumers propagate around the network
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);

        sendMessages("Broker_C_C", dest, 1);
        assertNotNull(clientA.receive(1000));
        assertNotNull(clientB.receive(1000));

        //bring online a consumer on the other side
        final Session ses2 = createSession("Broker_C_C");
        final MessageConsumer clientC = ses2.createDurableSubscriber(dest, "subC");
        //there will be 2 network durables, 1 for each direction of the bridge
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);

        clientA.close();
        clientB.close();
        clientC.close();
        ses.unsubscribe("subA");
        ses.unsubscribe("subB");
        ses2.unsubscribe("subC");

        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0);
    }

    public void testDurablePropagationConsumerAllBrokersDuplex() throws Exception {
        duplex = true;
        testDurablePropagationConsumerAllBrokers();
    }

    public void testDurablePropagationConsumerAllBrokersOneWay() throws Exception {
        duplex = false;
        testDurablePropagationConsumerAllBrokers();
    }

    protected void testDurablePropagationConsumerAllBrokers() throws Exception {
        // Setup broker networks
        bridgeBrokers("Broker_A_A", "Broker_B_B");
        bridgeBrokers("Broker_B_B", "Broker_C_C");
        if (!duplex) {
            bridgeBrokers("Broker_B_B", "Broker_A_A");
            bridgeBrokers("Broker_C_C", "Broker_B_B");
        }

        startAllBrokers();
        waitForBridgeFormation();

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

        // Setup consumers
        final Session ses = createSession("Broker_A_A");
        final MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");

        // let consumers propagate around the network
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);

        //bring online a consumer on the other side
        final Session ses2 = createSession("Broker_B_B");
        final MessageConsumer clientB = ses2.createDurableSubscriber(dest, "subB");

        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);

        final Session ses3 = createSession("Broker_C_C");
        final MessageConsumer clientC = ses3.createDurableSubscriber(dest, "subC");

        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);

        clientA.close();
        clientB.close();
        clientC.close();
        ses.unsubscribe("subA");
        ses2.unsubscribe("subB");
        ses3.unsubscribe("subC");

        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0);
    }

    public void testDurablePropagation5BrokerDuplex() throws Exception {
        duplex = true;
        testDurablePropagation5Broker();
    }

    public void testDurablePropagation5BrokerOneWay() throws Exception {
        duplex = false;
        testDurablePropagation5Broker();
    }

    protected void testDurablePropagation5Broker() throws Exception {
        // Setup broker networks
        bridgeBrokers("Broker_A_A", "Broker_B_B");
        bridgeBrokers("Broker_B_B", "Broker_C_C");
        bridgeBrokers("Broker_C_C", "Broker_D_D");
        bridgeBrokers("Broker_D_D", "Broker_E_E");
        if (!duplex) {
            bridgeBrokers("Broker_B_B", "Broker_A_A");
            bridgeBrokers("Broker_C_C", "Broker_B_B");
            bridgeBrokers("Broker_D_D", "Broker_C_C");
            bridgeBrokers("Broker_E_E", "Broker_D_D");
        }

        startAllBrokers();
        waitForBridgeFormation();

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

        // Setup consumers
        final Session ses = createSession("Broker_A_A");
        final MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");

        // let consumers propagate around the network
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);

        sendMessages("Broker_E_E", dest, 1);
        assertNotNull(clientA.receive(1000));

        //bring online a consumer on the other side
        final Session ses2 = createSession("Broker_E_E");
        final MessageConsumer clientE = ses2.createDurableSubscriber(dest, "subE");

        //there will be 2 network durables, 1 for each direction of the bridge
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);

        clientA.close();
        clientE.close();
        ses.unsubscribe("subA");
        ses2.unsubscribe("subE");

        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 0);
    }

    public void testDurablePropagationSpokeDuplex() throws Exception {
        duplex = true;
        testDurablePropagationSpoke();
    }

    public void testDurablePropagationSpokeOneWay() throws Exception {
        duplex = false;
        testDurablePropagationSpoke();
    }

    protected void testDurablePropagationSpoke() throws Exception {
        // Setup broker networks
        bridgeBrokers("Broker_A_A", "Broker_B_B");
        bridgeBrokers("Broker_B_B", "Broker_C_C");
        bridgeBrokers("Broker_B_B", "Broker_D_D");
        if (!duplex) {
            bridgeBrokers("Broker_B_B", "Broker_A_A");
            bridgeBrokers("Broker_C_C", "Broker_B_B");
            bridgeBrokers("Broker_D_D", "Broker_B_B");
        }

        startAllBrokers();
        waitForBridgeFormation();

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

        // Setup consumers
        final Session ses = createSession("Broker_A_A");
        final Session ses2 = createSession("Broker_B_B");
        final Session ses3 = createSession("Broker_C_C");
        final Session ses4 = createSession("Broker_D_D");

        final MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");
        final MessageConsumer clientAB = ses.createDurableSubscriber(dest, "subAB");

        // let consumers propagate around the network
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);

        final MessageConsumer clientD = ses4.createDurableSubscriber(dest, "subD");

        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);

        sendMessages("Broker_A_A", dest, 1);
        assertNotNull(clientD.receive(1000));
        sendMessages("Broker_C_C", dest, 1);
        assertNotNull(clientD.receive(1000));

        final MessageConsumer clientB = ses2.createDurableSubscriber(dest, "subB");
        final MessageConsumer clientC = ses3.createDurableSubscriber(dest, "subC");

        // let consumers propagate around the network (Wait.waitFor polls internally)
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 3);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);

        clientA.close();
        clientAB.close();
        clientB.close();
        clientC.close();
        clientD.close();

        ses.unsubscribe("subA");
        ses.unsubscribe("subAB");
        ses2.unsubscribe("subB");
        ses3.unsubscribe("subC");
        ses4.unsubscribe("subD");

        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 0);
    }

    public void testForceDurablePropagationDuplex() throws Exception {
        duplex = true;
        testForceDurablePropagation();
    }

    public void testForceDurablePropagationOneWay() throws Exception {
        duplex = false;
        testForceDurablePropagation();
    }

    protected void testForceDurablePropagation() throws Exception {
        // Setup broker networks
        bridgeBrokers("Broker_A_A", "Broker_B_B");
        bridgeBrokers("Broker_B_B", "Broker_C_C");
        if (!duplex) {
            bridgeBrokers("Broker_B_B", "Broker_A_A");
            bridgeBrokers("Broker_C_C", "Broker_B_B");
        }

        startAllBrokers();
        waitForBridgeFormation();

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

        // Setup consumers
        final Session ses = createSession("Broker_A_A");
        final MessageConsumer clientA = ses.createConsumer(dest);

        // let consumers propagate around the network (Wait.waitFor polls internally)
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);

        sendMessages("Broker_C_C", dest, 1);
        assertNotNull(clientA.receive(1000));

        final Session ses2 = createSession("Broker_C_C");
        final MessageConsumer clientC = ses2.createConsumer(dest);

        // let consumers propagate around the network (Wait.waitFor polls internally)
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);

        clientA.close();
        clientC.close();

        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0);
    }

    public void testDurablePropagationSyncDuplex() throws Exception {
        duplex = true;
        testDurablePropagationSync();
    }

    public void testDurablePropagationSyncOneWay() throws Exception {
        duplex = false;
        testDurablePropagationSync();
    }

    protected void testDurablePropagationSync() throws Exception {
        // Setup broker networks
        final NetworkConnector nc1 = bridgeBrokers("Broker_A_A", "Broker_B_B");
        final NetworkConnector nc2 = bridgeBrokers("Broker_B_B", "Broker_C_C");

        NetworkConnector nc3 = null;
        NetworkConnector nc4 = null;
        if (!duplex) {
            nc3 = bridgeBrokers("Broker_B_B", "Broker_A_A");
            nc4 = bridgeBrokers("Broker_C_C", "Broker_B_B");
        }

        startAllBrokers();

        nc1.stop();
        nc2.stop();

        if (!duplex) {
            nc3.stop();
            nc4.stop();
        }

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

        // Setup consumers -- no bridges are running so no NC subs should be created
        final Session ses = createSession("Broker_A_A");
        final Session ses2 = createSession("Broker_C_C");
        final MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");
        final MessageConsumer clientB = ses.createDurableSubscriber(dest, "subB");
        final MessageConsumer clientC = ses2.createDurableSubscriber(dest, "subC");

        // No bridges running, so no NC durable subs should exist
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0);

        nc1.start();
        nc2.start();
        if (!duplex) {
            nc3.start();
            nc4.start();
        }

        //there will be 2 network durables, 1 for each direction of the bridge
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);

        clientA.close();
        clientB.close();
        clientC.close();
    }

    public void testDurablePropagationMultipleBridgesDifferentDestinations() throws Exception {
        duplex = true;

        // Setup broker networks
        bridgeBrokers("Broker_A_A", "Broker_B_B");
        bridgeBrokers("Broker_B_B", "Broker_C_C");

        //Duplicate the bridges with different included destinations - valid use case
        final NetworkConnector nc3 = bridgeBrokers("Broker_A_A", "Broker_B_B");
        final NetworkConnector nc4 = bridgeBrokers("Broker_B_B", "Broker_C_C");
        nc3.setName("nc_3_3");
        nc4.setName("nc_4_4");
        final ArrayList<ActiveMQDestination> includedDestinations = new ArrayList<>();
        includedDestinations.add(new ActiveMQTopic("TEST.FOO2?forceDurable=true"));
        nc3.setDynamicallyIncludedDestinations(includedDestinations);
        nc4.setDynamicallyIncludedDestinations(includedDestinations);

        startAllBrokers();
        waitForBridgeFormation();

        // Setup destination
        final ActiveMQTopic dest = (ActiveMQTopic) createDestination("TEST.FOO", true);
        final ActiveMQTopic dest2 = (ActiveMQTopic) createDestination("TEST.FOO2", true);

        // Setup consumers
        final Session ses = createSession("Broker_A_A");
        final Session ses2 = createSession("Broker_C_C");
        final MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");
        final MessageConsumer clientAa = ses.createDurableSubscriber(dest2, "subAa");
        final MessageConsumer clientC = ses2.createDurableSubscriber(dest, "subC");
        final MessageConsumer clientCc = ses2.createDurableSubscriber(dest2, "subCc");

        //make sure network durables are online (Wait.waitFor polls internally)
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1);
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest2, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest2, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest2, 1);

        clientA.close();
        clientC.close();
        ses.unsubscribe("subA");
        ses2.unsubscribe("subC");

        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest2, 2);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest2, 1);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest2, 1);

        clientAa.close();
        clientCc.close();
        ses.unsubscribe("subAa");
        ses2.unsubscribe("subCc");

        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest2, 0);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest2, 0);
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest2, 0);
    }

    /*
     * The following tests should work for correct sync/propagation of durables even if
     * TTL is missing on broker restart. This is for TTL: 0, -1, 1, or 4 (same number
     * of network hops in the network)
     */
    public void testDurablePropagationSyncTtl0BrokerRestart() throws Exception {
        testDurablePropagation(0, false, true, List.of(0, 0, 0, 0, 0));
    }

    public void testDurablePropagationSyncTtl1BrokerRestart() throws Exception {
        testDurablePropagation(1, false, true, List.of(0, 1, 0, 1, 0));
    }

    public void testDurablePropagationSyncTtl4BrokerRestart() throws Exception {
        testDurablePropagation(4, false, true, List.of(1, 2, 2, 2, 1));
    }

    // This test also tests the fix in NetworkBridgeUtils.getBrokerSubscriptionInfo()
    // where the clientId was missing for the offline durable which could cause demand to be created
    // by mistake and create a loop. Without the clientId the sync could not tell the
    // network durable was for its direct bridge (was created because of a local consumer
    // on its broker) so a loop could be created on restart if the sync happened before the
    // real consumer connected first.
    public void testDurablePropagationSyncTtlNotSetBrokerRestart() throws Exception {
        testDurablePropagation(-1, false, true, List.of(1, 2, 2, 2, 1));
    }

    /*
     * The following test demonstrates the problem with missing TTL and is only solved
     * by making dynamicOnly true, on restart consumers have yet to come back online
     * for offline durables so we end up missing TTL and create extra demand on sync.
     * TTL is missing on broker restart. This is for TTL > 1 but less than same number
     * of network hops in the network
     */
    public void testDurablePropagationDynamicFalseTtl2BrokerRestartFail() throws Exception {
        try {
            testDurablePropagation(2, false, true, List.of(0, 1, 2, 1, 0));
            fail("Exepected to fail");
        } catch (AssertionFailedError e) {
            // expected
        }
    }

    public void testDurablePropagationDynamicFalseTtl3BrokerRestartFail() throws Exception {
        try {
            testDurablePropagation(2, false, true, List.of(0, 2, 2, 2, 0));
            fail("Exepected to fail");
        } catch (AssertionFailedError e) {
            // expected
        }
    }

    /*
     * The following tests make sure propagation works correctly with TTL set
     * when dynamicOnly is false or true, and durable sync enabled.
     * When false, durables get reactivated and a sync is done and TTL can be
     * missing for offline durables so this works correctly ONLY if the consumers
     * are online and have correct TTL info. When true, consumers drive reactivation
     * entirely and the sync is skipped.
     *
     * For dynamicOnly being false, these tests for TTL > 1 demonstrate the improvement in
     * createDemandSubscription() inside of DemandForwardingBridgeSupport
     * where we now include the full broker path (all TTL info) if the
     * consumer is already online that created the subscription. Before,
     * we just included the remote broker.
     *
     * If brokers are restarted/consumers offline, TTL can be missing so it's recommended
     * to set dynamicOnly to true if TTL is > 1 (TTL 1 and TTL -1 can still be handled)
     * to prevent propagation of durables that shouldn't exist and let consumers drive
     * the reactivation.
     *
     * The tests keep the consumers online and only restart the connectors and not
     * the brokers so the consumer info and TTL are preserved for the tests.
     */
    public void testDurablePropagationDynamicFalseTtl0() throws Exception {
        testDurablePropagation(0, false, List.of(0, 0, 0, 0, 0));
    }

    public void testDurablePropagationDynamicFalseTtl1() throws Exception {
        testDurablePropagation(1, false, List.of(0, 1, 0, 1, 0));
    }

    public void testDurablePropagationDynamicFalseTtl2() throws Exception {
        testDurablePropagation(2, false, List.of(0, 1, 2, 1, 0));
    }

    public void testDurablePropagationDynamicFalseTtl3() throws Exception {
        testDurablePropagation(3, false, List.of(0, 2, 2, 2, 0));
    }

    public void testDurablePropagationDynamicFalseTtl4() throws Exception {
        testDurablePropagation(4, false, List.of(1, 2, 2, 2, 1));
    }

    public void testDurablePropagationDynamicFalseTtlNotSet() throws Exception {
        testDurablePropagation(-1, false, List.of(1, 2, 2, 2, 1));
    }

    public void testDurablePropagationDynamicTrueTtl0() throws Exception {
        testDurablePropagation(0, true, List.of(0, 0, 0, 0, 0));
    }

    public void testDurablePropagationDynamicTrueTtl1() throws Exception {
        testDurablePropagation(1, true, List.of(0, 1, 0, 1, 0));
    }

    public void testDurablePropagationDynamicTrueTtl2() throws Exception {
        testDurablePropagation(2, true, List.of(0, 1, 2, 1, 0));
    }

    public void testDurablePropagationDynamicTrueTtl3() throws Exception {
        testDurablePropagation(3, true, List.of(0, 2, 2, 2, 0));
    }

    public void testDurablePropagationDynamicTrueTtl4() throws Exception {
        testDurablePropagation(4, true, List.of(1, 2, 2, 2, 1));
    }

    public void testDurablePropagationDynamicTrueTtlNotSet() throws Exception {
        testDurablePropagation(-1, true, List.of(1, 2, 2, 2, 1));
    }

    private void testDurablePropagation(final int ttl, final boolean dynamicOnly,
                                        final List<Integer> expected) throws Exception {
        testDurablePropagation(ttl, dynamicOnly, false, expected);
    }

    private void testDurablePropagation(final int ttl, final boolean dynamicOnly, final boolean restartBrokers,
                                         final List<Integer> expected) throws Exception {
        duplex = true;

        // Setup broker networks
        final NetworkConnector nc1 = bridgeBrokers("Broker_A_A", "Broker_B_B", dynamicOnly, ttl);
        final NetworkConnector nc2 = bridgeBrokers("Broker_B_B", "Broker_C_C", dynamicOnly, ttl);
        final NetworkConnector nc3 = bridgeBrokers("Broker_C_C", "Broker_D_D", dynamicOnly, ttl);
        final NetworkConnector nc4 = bridgeBrokers("Broker_D_D", "Broker_E_E", dynamicOnly, ttl);

        startAllBrokers();
        stopNetworkConnectors(nc1, nc2, nc3, nc4);

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

        // Setup consumers
        final Session ses = createSession("Broker_A_A");
        final Session ses2 = createSession("Broker_E_E");
        final MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");
        final MessageConsumer clientE = ses2.createDurableSubscriber(dest, "subE");

        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 0);
        assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 0);

        startNetworkConnectors(nc1, nc2, nc3, nc4);

        // Check that the correct network durables exist
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, expected.get(0));
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, expected.get(1));
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, expected.get(2));
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, expected.get(3));
        assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, expected.get(4));

        if (restartBrokers) {
            // go offline and restart to make sure sync works to re-enable and doesn't
            // propagate wrong demand
            clientA.close();
            clientE.close();
            destroyAllBrokers();
            setUp();
            brokers.values().forEach(bi -> bi.broker.setDeleteAllMessagesOnStartup(false));
            bridgeBrokers("Broker_A_A", "Broker_B_B", dynamicOnly, ttl);
            bridgeBrokers("Broker_B_B", "Broker_C_C", dynamicOnly, ttl);
            bridgeBrokers("Broker_C_C", "Broker_D_D", dynamicOnly, ttl);
            bridgeBrokers("Broker_D_D", "Broker_E_E", dynamicOnly, ttl);
            startAllBrokers();
            waitForBridgeFormation();
        } else {
            // restart just the network connectors but leave the consumers online
            // to test sync works ok. Things should work for all cases both dynamicOnly
            // false and true because TTL info still exits and consumers are online
            stopNetworkConnectors(nc1, nc2, nc3, nc4);
            startNetworkConnectors(nc1, nc2, nc3, nc4);
        }

        // after restarting the bridges, check sync/demand are correct
        assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, expected.get(0));
        assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, expected.get(1));
        assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, expected.get(2));
        assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, expected.get(3));
        assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, expected.get(4));
    }

    protected void assertNCDurableSubsCount(final BrokerService brokerService, final ActiveMQTopic dest,
            final int count) throws Exception {
        final boolean result = Wait.waitFor(() -> count == getNCDurableSubs(brokerService, dest).size(),
                TimeUnit.SECONDS.toMillis(30), 500);
        assertTrue("Expected " + count + " NC durable sub(s) on " + brokerService.getBrokerName()
                + " for " + dest.getTopicName() + ", but got "
                + getNCDurableSubs(brokerService, dest).size(), result);
    }

    protected List<DurableTopicSubscription> getNCDurableSubs(final BrokerService brokerService,
            final ActiveMQTopic dest) throws Exception {
        final List<DurableTopicSubscription> subs = new ArrayList<>();
        final Destination d = brokerService.getDestination(dest);
        final org.apache.activemq.broker.region.Topic destination;
        if (d instanceof DestinationFilter) {
            destination = ((DestinationFilter) d).getAdaptor(org.apache.activemq.broker.region.Topic.class);
        } else {
            destination = (org.apache.activemq.broker.region.Topic) d;
        }

        for (final SubscriptionKey key : destination.getDurableTopicSubs().keySet()) {
            if (key.getSubscriptionName().startsWith(DemandForwardingBridge.DURABLE_SUB_PREFIX)) {
                final DurableTopicSubscription sub = destination.getDurableTopicSubs().get(key);
                if (sub != null) {
                    subs.add(sub);
                }
            }
        }

        return subs;
    }

    @Override
    public void setUp() throws Exception {
        super.setAutoFail(true);
        super.setUp();
        deletePersistentMessagesOnStartup = true;
        final String options = "?persistent=true&useJmx=false";
        // Use ephemeral ports (0) to avoid port conflicts when tests run in parallel
        createBroker(new URI("broker:(tcp://localhost:0)/Broker_A_A" + options));
        createBroker(new URI("broker:(tcp://localhost:0)/Broker_B_B" + options));
        createBroker(new URI("broker:(tcp://localhost:0)/Broker_C_C" + options));
        createBroker(new URI("broker:(tcp://localhost:0)/Broker_D_D" + options));
        createBroker(new URI("broker:(tcp://localhost:0)/Broker_E_E" + options));
    }

    @Override
    protected void configureBroker(BrokerService broker) {
        broker.setBrokerId(broker.getBrokerName());
        broker.setDeleteAllMessagesOnStartup(deletePersistentMessagesOnStartup);
        broker.setDataDirectory("target" + File.separator + "test-data" + File.separator + "DurableFiveBrokerNetworkBridgeTest");
    }

    /**
     * Wait for all bridge sync executors (both initiator and duplex bridges) to complete
     * processing their BrokerSubscriptionInfo tasks on all brokers.
     * Uses reflection to access the private syncExecutor, following the same pattern
     * as {@link DynamicNetworkTestSupport#findDuplexBridge}.
     */
    private void waitForAllBridgeSyncCompletion() throws Exception {
        final Field syncExecutorField = DemandForwardingBridgeSupport.class.getDeclaredField("syncExecutor");
        syncExecutorField.setAccessible(true);
        final Field duplexBridgeField = TransportConnection.class.getDeclaredField("duplexBridge");
        duplexBridgeField.setAccessible(true);

        for (final BrokerItem item : brokers.values()) {
            final BrokerService broker = item.broker;
            // Initiator bridges (accessible via network connectors)
            for (final NetworkConnector nc : broker.getNetworkConnectors()) {
                for (final NetworkBridge bridge : nc.activeBridges()) {
                    if (bridge instanceof DemandForwardingBridgeSupport) {
                        flushSyncExecutor(syncExecutorField, (DemandForwardingBridgeSupport) bridge);
                    }
                }
            }
            // Duplex bridges (accessible via transport connections)
            for (final TransportConnector tc : broker.getTransportConnectors()) {
                for (final TransportConnection conn : tc.getConnections()) {
                    if (conn.getConnectionId() != null && conn.getConnectionId().startsWith("networkConnector_")) {
                        final DemandForwardingBridgeSupport duplexBridge =
                            (DemandForwardingBridgeSupport) duplexBridgeField.get(conn);
                        if (duplexBridge != null) {
                            flushSyncExecutor(syncExecutorField, duplexBridge);
                        }
                    }
                }
            }
        }
    }

    private void flushSyncExecutor(final Field syncExecutorField,
            final DemandForwardingBridgeSupport bridge) throws Exception {
        final ExecutorService syncExecutor = (ExecutorService) syncExecutorField.get(bridge);
        if (syncExecutor.isShutdown()) {
            return;
        }
        final CountDownLatch latch = new CountDownLatch(1);
        try {
            syncExecutor.execute(latch::countDown);
        } catch (final RejectedExecutionException e) {
            return;
        }
        assertTrue("Sync executor should complete on " + bridge,
            latch.await(30, TimeUnit.SECONDS));
    }

    protected void startNetworkConnectors(NetworkConnector... connectors) throws Exception {
        for (final NetworkConnector connector : connectors) {
            connector.start();
        }
    }

    protected void stopNetworkConnectors(NetworkConnector... connectors) throws Exception {
        for (final NetworkConnector connector : connectors) {
            connector.stop();
        }
    }

    protected Session createSession(String broker) throws Exception {
        final Connection con = createConnection(broker);
        con.start();
        return con.createSession(false, Session.AUTO_ACKNOWLEDGE);
    }

    public static Test suite() {
        return suite(DurableFiveBrokerNetworkBridgeTest.class);
    }
}
