content: /*
   * 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.artemis.tests.integration.federation;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  import javax.jms.Connection;
  import javax.jms.ConnectionFactory;
  import javax.jms.JMSException;
  import javax.jms.Message;
  import javax.jms.MessageConsumer;
  import javax.jms.MessageProducer;
  import javax.jms.Queue;
  import javax.jms.Session;
  import javax.jms.TextMessage;
  import java.util.Collections;
  
  import org.apache.activemq.artemis.api.core.QueueConfiguration;
  import org.apache.activemq.artemis.api.core.RoutingType;
  import org.apache.activemq.artemis.api.core.SimpleString;
  import org.apache.activemq.artemis.core.config.FederationConfiguration;
  import org.apache.activemq.artemis.core.config.federation.FederationQueuePolicyConfiguration;
  import org.apache.activemq.artemis.core.config.federation.FederationUpstreamConfiguration;
  import org.apache.activemq.artemis.core.postoffice.QueueBinding;
  import org.apache.activemq.artemis.core.server.ActiveMQServer;
  import org.apache.activemq.artemis.core.server.transformer.Transformer;
  import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
  import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
  import org.apache.activemq.artemis.tests.util.Wait;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  
  public class FederatedQueueTest extends FederatedTestBase {
  
     @Override
     @BeforeEach
     public void setUp() throws Exception {
        super.setUp();
     }
  
  
     @Override
     protected void configureQueues(ActiveMQServer server) throws Exception {
        server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoCreateAddresses(false).setAutoCreateQueues(false).setDefaultConsumerWindowSize(-1));
        createSimpleQueue(server, getName());
     }
  
     protected ConnectionFactory getCF(int i) throws Exception {
        return new ActiveMQConnectionFactory("vm://" + i);
     }
  
     @Test
     public void testFederatedQueueRemoteConsumeUpstream() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server1", queueName);
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
        getServer(0).getFederationManager().deploy();
  
        testFederatedQueueRemoteConsume(queueName);
     }
  
     @Test
     public void testMultipleFederatedQueueRemoteConsumersUpstream() throws Exception {
        String connector = "server1";
  
        getServer(0).getAddressSettingsRepository().getMatch("#").setAutoCreateAddresses(true).setAutoCreateQueues(true);
        getServer(1).getAddressSettingsRepository().getMatch("#").setAutoCreateAddresses(true).setAutoCreateQueues(true);
  
        getServer(1).createQueue(QueueConfiguration.of("Test.Q.1").setRoutingType(RoutingType.ANYCAST));
        getServer(1).createQueue(QueueConfiguration.of("Test.Q.2").setRoutingType(RoutingType.ANYCAST));
  
        getServer(0).getConfiguration().getFederationConfigurations().add(new FederationConfiguration()
                                                                             .setName("default")
                                                                             .addFederationPolicy(new FederationQueuePolicyConfiguration()
                                                                                                     .setName("myQueuePolicy")
                                                                                                     .addInclude(new FederationQueuePolicyConfiguration.Matcher()
                                                                                                                    .setQueueMatch("#")
                                                                                                                    .setAddressMatch("Test.#")))
                                                                             .addUpstreamConfiguration(new FederationUpstreamConfiguration()
                                                                                                          .setName("server1-upstream")
                                                                                                          .addPolicyRef("myQueuePolicy")
                                                                                                          .setStaticConnectors(Collections.singletonList(connector))));
        getServer(0).getFederationManager().deploy();
  
        ConnectionFactory cf1 = getCF(0);
        ConnectionFactory cf2 = getCF(0);
        ConnectionFactory cf3 = getCF(1);
        try (Connection consumer1Connection = cf1.createConnection();
             Connection consumer2Connection = cf2.createConnection();
             Connection producerConnection = cf3.createConnection()) {
           consumer1Connection.start();
           Session session1 = consumer1Connection.createSession();
           Queue queue1 = session1.createQueue("Test.Q.1");
           MessageConsumer consumer1 = session1.createConsumer(queue1);
  
           consumer2Connection.start();
           Session session2 = consumer2Connection.createSession();
           Queue queue2 = session2.createQueue("Test.Q.2");
           MessageConsumer consumer2 = session2.createConsumer(queue2);
  
           Session session3 = producerConnection.createSession();
           MessageProducer producer = session3.createProducer(queue2);
           producer.send(session3.createTextMessage("hello"));
  
           assertNotNull(consumer2.receive(1000));
  
           consumer1Connection.close();
  
           producer.send(session3.createTextMessage("hello"));
  
           assertNotNull(consumer2.receive(1000));
        }
     }
  
     @Test
     public void testFederatedQueueRemoteConsumeUpstreamPriorityAdjustment() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server1", queueName);
        FederationQueuePolicyConfiguration policy = (FederationQueuePolicyConfiguration) federationConfiguration.getFederationPolicyMap().get("QueuePolicy" + queueName);
        //Favor federated broker over local consumers
        policy.setPriorityAdjustment(1);
  
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
        getServer(0).getFederationManager().deploy();
  
        testFederatedQueueRemoteConsumeUpstreamPriorityAdjustment(queueName);
     }
  
     @Test
     public void testFederatedQueueRemoteConsumeDownstreamPriorityAdjustment() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueDownstreamFederationConfiguration("server0", queueName, "server1");
        FederationQueuePolicyConfiguration policy = (FederationQueuePolicyConfiguration) federationConfiguration.getFederationPolicyMap().get("QueuePolicy" + queueName);
        //Favor federated broker over local consumers
        policy.setPriorityAdjustment(1);
  
        getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration);
        getServer(1).getFederationManager().deploy();
  
        testFederatedQueueRemoteConsumeUpstreamPriorityAdjustment(queueName);
     }
  
     private void testFederatedQueueRemoteConsumeUpstreamPriorityAdjustment(final String queueName) throws Exception {
        ConnectionFactory cf1 = getCF(1);
        ConnectionFactory cf0 = getCF(0);
        try (Connection connection1 = cf1.createConnection();
             Connection connection0 = cf0.createConnection()) {
           connection0.start();
           connection1.start();
           Session session0 = connection0.createSession();
           Session session1 = connection1.createSession();
           Queue queue0 = session0.createQueue(queueName);
           Queue queue1 = session1.createQueue(queueName);
  
           MessageConsumer consumer1 = session1.createConsumer(queue1);
           Wait.waitFor(() -> getConsumerCount(getServer(1), queueName, 1));
  
           MessageConsumer consumer0 = session0.createConsumer(queue0);
           Wait.waitFor(() -> getConsumerCount(getServer(1), queueName, 2));
  
           MessageProducer producer1 = session1.createProducer(queue1);
           producer1.send(session1.createTextMessage("hello"));
  
           //Consumer 0 should receive the message over consumer because of adjusted priority
           //to favor the federated broker
           assertNull(consumer1.receiveNoWait());
           assertNotNull(consumer0.receive(1000));
  
           consumer0.close();
           consumer1.close();
        }
     }
  
     private void verifyTransformer(String queueName) throws Exception {
        ConnectionFactory cf1 = getCF(1);
        ConnectionFactory cf0 = getCF(0);
        try (Connection connection1 = cf1.createConnection();
             Connection connection0 = cf0.createConnection()) {
           connection1.start();
           Session session1 = connection1.createSession();
           Queue queue1 = session1.createQueue(queueName);
           MessageProducer producer1 = session1.createProducer(queue1);
           producer1.send(session1.createTextMessage("hello"));
  
           connection0.start();
           Session session0 = connection0.createSession();
           Queue queue0 = session0.createQueue(queueName);
           MessageConsumer consumer0 = session0.createConsumer(queue0);
  
           Message message = consumer0.receive(1000);
           assertNotNull(message);
           assertTrue(message.getBooleanProperty(TestTransformer.TEST_PROPERTY));
        }
     }
  
     @Test
     public void testFederatedQueueRemoteConsumeUpstreamTransformer() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server1", queueName);
        FederatedTestUtil.addQueueTransformerConfiguration(federationConfiguration, queueName);
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
        getServer(0).getFederationManager().deploy();
  
        verifyTransformer(queueName);
     }
  
     @Test
     public void testFederatedQueueRemoteConsumeDownstream() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueDownstreamFederationConfiguration("server0", queueName, "server1");
        getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration);
        getServer(1).getFederationManager().deploy();
  
        testFederatedQueueRemoteConsume(queueName);
     }
  
     @Test
     public void testFederatedQueueRemoteConsumeDownstreamTransformer() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueDownstreamFederationConfiguration("server0", queueName, "server1");
        FederatedTestUtil.addQueueTransformerConfiguration(federationConfiguration, queueName);
        getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration);
        getServer(1).getFederationManager().deploy();
  
        verifyTransformer(queueName);
     }
  
     private void testFederatedQueueRemoteConsume(final String queueName) throws Exception {
  
        ConnectionFactory cf1 = getCF(1);
        ConnectionFactory cf0 = getCF(0);
        try (Connection connection1 = cf1.createConnection();
             Connection connection0 = cf0.createConnection()) {
           connection1.start();
           Session session1 = connection1.createSession();
           Queue queue1 = session1.createQueue(queueName);
           MessageProducer producer = session1.createProducer(queue1);
           producer.send(session1.createTextMessage("hello"));
  
           connection0.start();
           Session session0 = connection0.createSession();
           Queue queue0 = session0.createQueue(queueName);
           MessageConsumer consumer0 = session0.createConsumer(queue0);
  
           assertNotNull(consumer0.receive(1000));
  
  
           producer.send(session1.createTextMessage("hello"));
  
           assertNotNull(consumer0.receive(1000));
  
           MessageConsumer consumer1 = session1.createConsumer(queue1);
  
           producer.send(session1.createTextMessage("hello"));
  
           assertNotNull(consumer1.receive(1000));
           assertNull(consumer0.receiveNoWait());
           consumer1.close();
  
           //Groups
           producer.send(session1.createTextMessage("hello"));
           assertNotNull(consumer0.receive(1000));
  
           producer.send(createTextMessage(session1, "groupA"));
  
           assertNotNull(consumer0.receive(1000));
           consumer1 = session1.createConsumer(queue1);
  
           producer.send(createTextMessage(session1, "groupA"));
           assertNull(consumer1.receiveNoWait());
           assertNotNull(consumer0.receive(1000));
        }
  
     }
  
     @Test
     public void testWithLargeMessage() throws Exception {
        internalTestWithLargeMessages(1);
     }
  
     @Test
     public void testWithMultipleLargeMessages() throws Exception {
        internalTestWithLargeMessages(5);
     }
  
     private void internalTestWithLargeMessages(int messageNumber) throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server1", queueName);
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
        getServer(0).getFederationManager().deploy();
  
        ConnectionFactory cf1 = getCF(1);
        ConnectionFactory cf0 = getCF(0);
        final String payload = new String(new byte[1 * 1024 * 1024]).replace('\0', '+');
        try (Connection connection1 = cf1.createConnection();
             Connection connection0 = cf0.createConnection()) {
           connection1.start();
           Session session1 = connection1.createSession();
           Queue queue1 = session1.createQueue(queueName);
           MessageProducer producer = session1.createProducer(queue1);
           for (int i = 0; i < messageNumber; i++) {
              producer.send(session1.createTextMessage(payload));
           }
  
           connection0.start();
           Session session0 = connection0.createSession();
           Queue queue0 = session0.createQueue(queueName);
           MessageConsumer consumer0 = session0.createConsumer(queue0);
  
           for (int i = 0; i < messageNumber; i++) {
              assertNotNull(consumer0.receive(1000));
           }
        }
     }
  
     @Test
     public void testFederatedQueueRemoteConsumeDeployAfterConsumersExist() throws Exception {
        String queueName = getName();
        ConnectionFactory cf0 = getCF(0);
  
        ConnectionFactory cf1 = getCF(1);
        try (Connection connection0 = cf0.createConnection();
             Connection connection1 = cf1.createConnection()) {
  
           connection1.start();
           Session session1 = connection1.createSession();
           Queue queue1 = session1.createQueue(queueName);
           MessageProducer producer = session1.createProducer(queue1);
           producer.send(session1.createTextMessage("hello"));
  
           connection0.start();
           Session session0 = connection0.createSession();
           Queue queue0 = session0.createQueue(queueName);
           MessageConsumer consumer0 = session0.createConsumer(queue0);
  
           assertNull(consumer0.receiveNoWait());
  
           FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server1", queueName);
           getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
           getServer(0).getFederationManager().deploy();
  
           producer.send(session1.createTextMessage("hello"));
  
           assertNotNull(consumer0.receive(10000));
        }
     }
  
     @Test
     public void testFederatedQueueBiDirectionalUpstream() throws Exception {
        String queueName = getName();
        FederationConfiguration federationConfiguration0 = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server1", queueName);
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration0);
        getServer(0).getFederationManager().deploy();
  
        FederationConfiguration federationConfiguration1 = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server0", queueName);
        getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration1);
        getServer(1).getFederationManager().deploy();
  
        testFederatedQueueBiDirectional(queueName, false);
     }
  
     @Test
     public void testFederatedQueueBiDirectionalDownstream() throws Exception {
        String queueName = getName();
        FederationConfiguration federationConfiguration0 = FederatedTestUtil.createQueueDownstreamFederationConfiguration("server1", queueName, "server0");
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration0);
        getServer(0).getFederationManager().deploy();
  
        FederationConfiguration federationConfiguration1 = FederatedTestUtil.createQueueDownstreamFederationConfiguration("server0", queueName, "server1");
        getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration1);
        getServer(1).getFederationManager().deploy();
  
        testFederatedQueueBiDirectional(queueName, false);
     }
  
     @Test
     public void testFederatedQueueBiDirectionalDownstreamUpstream() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration0 = FederatedTestUtil.createQueueDownstreamFederationConfiguration("server1-downstream",
            "server1", queueName, null, false, "server0");
        FederationUpstreamConfiguration upstreamConfig = FederatedTestUtil.createQueueFederationUpstream("server1", queueName);
        federationConfiguration0.addUpstreamConfiguration(upstreamConfig);
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration0);
        getServer(0).getFederationManager().deploy();
  
        testFederatedQueueBiDirectional(queueName, false);
     }
  
     @Test
     public void testFederatedQueueBiDirectionalDownstreamUpstreamSharedConnection() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration0 = FederatedTestUtil.createQueueDownstreamFederationConfiguration("server1-downstream",
            "server1", queueName, null, true, "server0");
        FederationUpstreamConfiguration upstreamConfig = FederatedTestUtil.createQueueFederationUpstream("server1", queueName);
        upstreamConfig.getConnectionConfiguration().setShareConnection(true);
        federationConfiguration0.addUpstreamConfiguration(upstreamConfig);
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration0);
        getServer(0).getFederationManager().deploy();
  
        testFederatedQueueBiDirectional(queueName, true);
     }
  
     @Test
     public void testFederatedQueueShareUpstreamConnectionFalse() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration0 = FederatedTestUtil.createQueueDownstreamFederationConfiguration("server1-downstream",
            "server1", queueName, null, false, "server0");
        federationConfiguration0.addUpstreamConfiguration(FederatedTestUtil.createQueueFederationUpstream("server1", queueName));
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration0);
        getServer(0).getFederationManager().deploy();
  
        testFederatedQueueShareUpstreamConnection(queueName, 2, 3);
     }
  
     @Test
     public void testFederatedQueueShareUpstreamConnectionTrue() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration0 = FederatedTestUtil.createQueueDownstreamFederationConfiguration("server1-downstream",
            "server1", queueName, null, true, "server0");
        FederationUpstreamConfiguration upstreamConfiguration = FederatedTestUtil.createQueueFederationUpstream("server1", queueName);
        upstreamConfiguration.getConnectionConfiguration().setShareConnection(true);
        federationConfiguration0.addUpstreamConfiguration(upstreamConfiguration);
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration0);
        getServer(0).getFederationManager().deploy();
  
        testFederatedQueueShareUpstreamConnection(queueName, 2, 2);
     }
  
     private void testFederatedQueueShareUpstreamConnection(String queueName, int server0Connections, int server1Connections) throws Exception {
        ConnectionFactory cf1 = getCF(1);
        ConnectionFactory cf0 = getCF(0);
        try (Connection connection1 = cf1.createConnection();
             Connection connection0 = cf0.createConnection()) {
           connection0.start();
           connection1.start();
           Session session0 = connection0.createSession();
           Session session1 = connection1.createSession();
  
           MessageConsumer consumer0 = session0.createConsumer(session0.createQueue(queueName));
           MessageConsumer consumer1 = session1.createConsumer(session1.createQueue(queueName));
  
           assertTrue(Wait.waitFor(() -> getServer(0).getConnectionCount() == server0Connections, 500, 100));
           assertTrue(Wait.waitFor(() -> getServer(1).getConnectionCount() == server1Connections, 500, 100));
           assertFalse(Wait.waitFor(() -> getServer(0).getConnectionCount() > server0Connections, 500, 100));
           assertFalse(Wait.waitFor(() -> getServer(1).getConnectionCount() > server1Connections, 500, 100));
        }
     }
  
     private void testFederatedQueueBiDirectional(String queueName, boolean shared) throws Exception {
        ConnectionFactory cf1 = getCF(1);
        ConnectionFactory cf0 = getCF(0);
        try (Connection connection1 = cf1.createConnection();
             Connection connection0 = cf0.createConnection()) {
           connection0.start();
           Session session0 = connection0.createSession();
           Queue queue0 = session0.createQueue(queueName);
           MessageProducer producer0 = session0.createProducer(queue0);
  
           connection1.start();
           Session session1 = connection1.createSession();
           Queue queue1 = session1.createQueue(queueName);
           MessageProducer producer1 = session1.createProducer(queue1);
           MessageConsumer consumer0 = session0.createConsumer(queue0);
  
           //Test producers being on broker 0 and broker 1 and consumer on broker 0.
           producer0.send(session1.createTextMessage("hello"));
           assertNotNull(consumer0.receive(1000));
  
           producer1.send(session1.createTextMessage("hello"));
           assertNotNull(consumer0.receive(1000));
  
           Wait.assertTrue(() -> getServer(0).getPostOffice().getBinding(SimpleString.of(queueName)) != null);
           Wait.assertTrue(() -> getServer(1).getPostOffice().getBinding(SimpleString.of(queueName)) != null);
           //Wait to see if extra consumers are created - this tests to make sure there is no loop and tests the FederatedQueue metaDataFilterString
           //is working properly - should only be 1 consumer on each (1 for the local consumer on broker0 and 1 for the federated consumer on broker1)
           assertFalse(Wait.waitFor(() -> getServer(0).locateQueue(SimpleString.of(queueName)).getConsumerCount() > 1, 500, 100));
           assertFalse(Wait.waitFor(() -> getServer(1).locateQueue(SimpleString.of(queueName)).getConsumerCount() > 1, 500, 100));
  
           //Test consumer move from broker 0, to broker 1
           final int server1ConsumerCount = getServer(1).getConnectionCount();
           consumer0.close();
           Wait.waitFor(() -> ((QueueBinding) getServer(0).getPostOffice().getBinding(SimpleString.of(queueName))).consumerCount() == 0, 1000);
  
           //Make sure we don't drop connection if shared
           if (shared) {
              assertFalse(Wait.waitFor(() -> getServer(1).getConnectionCount() == server1ConsumerCount - 1,
                      500, 100));
              assertEquals(server1ConsumerCount, getServer(1).getConnectionCount());
           }
  
           MessageConsumer consumer1 = session1.createConsumer(queue1);
  
           producer0.send(session1.createTextMessage("hello"));
           assertNotNull(consumer1.receive(1000));
  
           producer1.send(session1.createTextMessage("hello"));
           assertNotNull(consumer1.receive(1000));
  
           //Test consumers on both broker 0, and broker 1 that messages route to consumers on same broker
           consumer0 = session0.createConsumer(queue0);
  
           producer0.send(session1.createTextMessage("produce0"));
           producer1.send(session1.createTextMessage("produce1"));
  
           Message message0 = consumer0.receive(1000);
           assertNotNull(message0);
           assertEquals("produce0", ((TextMessage) message0).getText());
  
           Message message1 = consumer1.receive(1000);
           assertNotNull(message1);
           assertEquals("produce1", ((TextMessage) message1).getText());
        }
     }
  
     @Test
     public void testFederatedQueueChainOfBrokers() throws Exception {
        String queueName = getName();
  
        //Connect broker 0 (consumer will be here at end of chain) to broker 1
        FederationConfiguration federationConfiguration0 = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server1", queueName, true);
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration0);
        getServer(0).getFederationManager().deploy();
  
        //Connect broker 1 (middle of chain) to broker 2
        FederationConfiguration federationConfiguration1 = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server2", queueName, true);
        getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration1);
        getServer(1).getFederationManager().deploy();
        //Broker 2 we dont setup any federation as he is the upstream (head of the chain)
  
        //Now the test.
  
  
        ConnectionFactory cf2 = getCF(2);
        ConnectionFactory cf0 = getCF(0);
        try (Connection connection2 = cf2.createConnection();
             Connection connection0 = cf0.createConnection()) {
           connection0.start();
           Session session0 = connection0.createSession();
           Queue queue0 = session0.createQueue(queueName);
  
           connection2.start();
           Session session2 = connection2.createSession();
           Queue queue2 = session2.createQueue(queueName);
           MessageProducer producer2 = session2.createProducer(queue2);
           MessageConsumer consumer0 = session0.createConsumer(queue0);
  
  
           //Test producers being on broker 2 and consumer on broker 0, with broker 2 being in the middle of the chain.
           producer2.send(session2.createTextMessage("hello"));
           assertNotNull(consumer0.receive(1000));
        }
     }
  
     @Test
     public void testFederatedQueueRemoteBrokerRestart() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server1", queueName);
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
        getServer(0).getFederationManager().deploy();
  
        ConnectionFactory cf1 = getCF(1);
        Connection connection1 = cf1.createConnection();
        connection1.start();
        Session session1 = connection1.createSession();
        Queue queue1 =  session1.createQueue(queueName);
        MessageProducer producer = session1.createProducer(queue1);
        producer.send(session1.createTextMessage("hello"));
  
        ConnectionFactory cf0 = getCF(0);
        Connection connection0 = cf0.createConnection();
        connection0.start();
        Session session0 = connection0.createSession();
        Queue queue0 =  session0.createQueue(queueName);
        MessageConsumer consumer0 = session0.createConsumer(queue0);
  
        assertNotNull(consumer0.receive(1000));
  
        producer.send(session1.createTextMessage("hello"));
  
        assertNotNull(consumer0.receive(1000));
  
        connection1.close();
        getServer(1).stop();
  
        assertNull(consumer0.receiveNoWait());
  
        getServer(1).start();
        Wait.assertTrue(getServer(1)::isActive);
        createSimpleQueue(getServer(1), getName());
  
        connection1 = cf1.createConnection();
        connection1.start();
        session1 = connection1.createSession();
        queue1 =  session1.createQueue(queueName);
        producer = session1.createProducer(queue1);
        producer.send(session1.createTextMessage("hello"));
  
        Wait.waitFor(() -> getConsumerCount(getServer(1), queueName, 1));
  
        assertNotNull(consumer0.receive(1000));
     }
  
     private boolean getConsumerCount(ActiveMQServer server, String queueName, int count) {
        QueueBinding binding = (QueueBinding)server.getPostOffice().getBinding(SimpleString.of(queueName));
        if (binding == null) {
           return false;
        }
        if (binding.consumerCount() != count) {
           return false;
        }
  
        return true;
     }
  
     @Test
     public void testFederatedQueueLocalBrokerRestart() throws Exception {
        String queueName = getName();
  
        FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server1", queueName);
        getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
        getServer(0).getFederationManager().deploy();
  
        ConnectionFactory cf1 = getCF(1);
        Connection connection1 = cf1.createConnection();
        connection1.start();
        Session session1 = connection1.createSession();
        Queue queue1 =  session1.createQueue(queueName);
        MessageProducer producer = session1.createProducer(queue1);
        producer.send(session1.createTextMessage("hello"));
  
        ConnectionFactory cf0 = getCF(0);
        Connection connection0 = cf0.createConnection();
        connection0.start();
        Session session0 = connection0.createSession();
        Queue queue0 =  session0.createQueue(queueName);
        MessageConsumer consumer0 = session0.createConsumer(queue0);
  
        assertNotNull(consumer0.receive(1000));
  
        producer.send(session1.createTextMessage("hello"));
  
        assertNotNull(consumer0.receive(1000));
  
        connection0.close();
        getServer(0).stop();
  
        producer.send(session1.createTextMessage("hello"));
  
        getServer(0).start();
        Wait.waitFor(() -> getServer(0).isActive());
        createSimpleQueue(getServer(0), getName());
  
        connection0 = getCF(0).createConnection();
        connection0.start();
        session0 = connection0.createSession();
        queue0 =  session0.createQueue(queueName);
        consumer0 = session0.createConsumer(queue0);
        producer.send(session1.createTextMessage("hello"));
  
        Wait.assertTrue(() -> getServer(1).getPostOffice().getBinding(SimpleString.of(queueName)) != null);
        Wait.waitFor(() -> ((QueueBinding) getServer(1)
              .getPostOffice()
              .getBinding(SimpleString.of(queueName)))
              .consumerCount() == 1);
  
        assertNotNull(consumer0.receive(5000));
     }
  
     private Message createTextMessage(Session session1, String group) throws JMSException {
        Message message = session1.createTextMessage("hello");
        message.setStringProperty("JMSXGroupID", group);
        return message;
     }
  
     public static class TestTransformer implements Transformer {
  
        static String TEST_PROPERTY = "transformed";
  
        @Override
        public org.apache.activemq.artemis.api.core.Message transform(org.apache.activemq.artemis.api.core.Message message) {
           message.putBooleanProperty(TEST_PROPERTY, true);
           return message;
        }
     }
  }
  