/*
 * 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 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 javax.jms.Topic;

import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.config.FederationConfiguration;
import org.apache.activemq.artemis.core.config.federation.FederationAddressPolicyConfiguration;
import org.apache.activemq.artemis.core.postoffice.QueueBinding;
import org.apache.activemq.artemis.core.server.ComponentConfigurationRoutingType;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.server.transformer.Transformer;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class FederatedAddressTest extends FederatedTestBase {

   @Override
   @BeforeEach
   public void setUp() throws Exception {
      super.setUp();
   }

   protected ConnectionFactory getCF(int i) throws Exception {
      return new ActiveMQConnectionFactory("vm://" + i);
   }

   @Test
   public void testDownstreamFederatedAddressReplication() throws Exception {
      String address = getName();

      FederationConfiguration federationConfiguration = FederatedTestUtil.createAddressDownstreamFederationConfiguration("server1", address,
          getServer(0).getConfiguration().getTransportConfigurations("server0")[0]);
      getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
      getServer(0).getFederationManager().deploy();

      FederationConfiguration federationConfiguration2 = FederatedTestUtil.createAddressDownstreamFederationConfiguration("server0", address,
          getServer(1).getConfiguration().getTransportConfigurations("server1")[0]);
      getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration2);
      getServer(1).getFederationManager().deploy();

      testFederatedAddressReplication(address);
   }

   @Test
   public void testDownstreamFederatedAddressReplicationRef() throws Exception {
      String address = getName();

      FederationConfiguration federationConfiguration = FederatedTestUtil.createAddressDownstreamFederationConfiguration("server1", address,
         "server0");
      getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
      getServer(0).getFederationManager().deploy();

      FederationConfiguration federationConfiguration2 = FederatedTestUtil.createAddressDownstreamFederationConfiguration("server0", address,
          "server1");
      getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration2);
      getServer(1).getFederationManager().deploy();

      testFederatedAddressReplication(address);
   }

   @Test
   public void testDownstreamFederatedAddressReplicationRefOneWay() throws Exception {
      String address = getName();

      FederationConfiguration federationConfiguration2 = FederatedTestUtil.createAddressDownstreamFederationConfiguration("server0", address,
          "server1");
      getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration2);
      getServer(1).getFederationManager().deploy();

      testFederatedAddressReplication(address);
   }

   @Test
   public void testUpstreamFederatedAddressReplication() throws Exception {
      String address = getName();

      FederationConfiguration federationConfiguration = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server1", address);
      getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
      getServer(0).getFederationManager().deploy();

      FederationConfiguration federationConfiguration2 = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server0", address);
      getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration2);
      getServer(1).getFederationManager().deploy();

      testFederatedAddressReplication(address);
   }

   @Test
   public void testDownstreamFederatedAddressReplicationRefOneWayTransformer() throws Exception {
      String address = getName();

      FederationConfiguration federationConfiguration2 = FederatedTestUtil.createAddressDownstreamFederationConfiguration("server0", address, "server1");
      FederatedTestUtil.addAddressTransformerConfiguration(federationConfiguration2, address);
      getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration2);
      getServer(1).getFederationManager().deploy();

      verifyTransformer(address);
   }

   private void verifyTransformer(String address) throws Exception {
      ConnectionFactory cf1 = getCF(1);
      ConnectionFactory cf0 = getCF(0);
      try (Connection connection1 = cf1.createConnection();
           Connection connection0 = cf0.createConnection()) {
         connection1.start();
         connection0.start();

         Session session1 = connection1.createSession();
         Topic topic1 = session1.createTopic(address);
         MessageProducer producer1 = session1.createProducer(topic1);

         Session session0 = connection0.createSession();
         Topic topic0 = session0.createTopic(address);
         MessageConsumer consumer0 = session0.createConsumer(topic0);

         assertTrue(Wait.waitFor(() -> getServer(1).getPostOffice().getBindingsForAddress(
            SimpleString.of(address)).getBindings().size() == 1));

         producer1.send(session1.createTextMessage("hello"));
         Message message = consumer0.receive(1000);
         assertNotNull(message);
         assertTrue(message.getBooleanProperty(FederatedQueueTest.TestTransformer.TEST_PROPERTY));
      }
   }

   @Test
   public void testUpstreamFederatedAddressReplicationOneWay() throws Exception {
      String address = getName();

      FederationConfiguration federationConfiguration = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server1", address);
      getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
      getServer(0).getFederationManager().deploy();

      testFederatedAddressReplication(address);
   }

   @Test
   public void testUpstreamFederatedAddressReplicationOneWayTransformer() throws Exception {
      String address = getName();

      FederationConfiguration federationConfiguration = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server1", address);
      FederatedTestUtil.addAddressTransformerConfiguration(federationConfiguration, address);
      getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
      getServer(0).getFederationManager().deploy();

      verifyTransformer(address);
   }

   /**
    * Test diverts for downstream configurations
    */
   //Test creating address first followed by divert
   //Test creating divert before consumer
   //Test destroy divert at end
   @Test
   public void testDownstreamDivertAddressFirstAndDivertFirstDestroyDivert() throws Exception {
      testFederatedAddressDivert(true, true, true, true);
   }

   //Test creating address first followed by divert
   //Test creating divert before consumer
   //Test destroy queue at end
   @Test
   public void testDownstreamDivertAddressFirstAndDivertFirstDestroyQueue() throws Exception {
      testFederatedAddressDivert(true, true, true, false);
   }

   //Test creating divert first followed by address
   //Test creating divert before consumer
   //Test destroy divert at end
   @Test
   public void testDownstreamDivertAddressSecondDivertFirstDestroyDivert() throws Exception {
      testFederatedAddressDivert(true, false, true, true);
   }

   //Test creating divert first followed by address
   //Test creating divert before consumer
   //Test destroy divert at end
   @Test
   public void testDownstreamDivertAddressSecondDivertFirstDestroyQueue() throws Exception {
      testFederatedAddressDivert(true, false, true, false);
   }

   //Test creating address first followed by divert
   //Test creating consumer before divert
   //Test destroy divert at end
   @Test
   public void testDownstreamDivertAddressFirstDivertSecondDestroyDivert() throws Exception {
      testFederatedAddressDivert(true, true, false, true);
   }

   //Test creating address first followed by divert
   //Test creating consumer before divert
   //Test destroy queue at end
   @Test
   public void testDownstreamDivertAddressFirstDivertSecondDestroyQueue() throws Exception {
      testFederatedAddressDivert(true, true, false, false);
   }

   //Test creating divert first followed by address
   //Test creating consumer before divert
   //Test destroy divert at end
   @Test
   public void testDownstreamDivertAddressAndDivertSecondDestroyDivert() throws Exception {
      testFederatedAddressDivert(true, false, false, true);
   }

   //Test creating divert first followed by address
   //Test creating consumer before divert
   //Test destroy queue at end
   @Test
   public void testDownstreamDivertAddressAndDivertSecondDestroyQueue() throws Exception {
      testFederatedAddressDivert(true, false, false, false);
   }

   /**
    * Test diverts for upstream configurations
    */
   //Test creating address first followed by divert
   //Test creating divert before consumer
   //Test destroy divert at end
   @Test
   public void testUpstreamDivertAddressAndDivertFirstDestroyDivert() throws Exception {
      testFederatedAddressDivert(false, true, true, true);
   }

   //Test creating address first followed by divert
   //Test creating divert before consumer
   //Test destroy queue at end
   @Test
   public void testUpstreamDivertAddressAndDivertFirstDestroyQueue() throws Exception {
      testFederatedAddressDivert(false, true, true, false);
   }

   //Test creating divert first followed by address
   //Test creating divert before consumer
   //Test destroy divert at end
   @Test
   public void testUpstreamDivertAddressSecondDivertFirstDestroyDivert() throws Exception {
      testFederatedAddressDivert(false, false, true, true);
   }

   //Test creating divert first followed by address
   //Test creating divert before consumer
   //Test destroy queue at end
   @Test
   public void testUpstreamDivertAddressSecondDivertFirstDestroyQueue() throws Exception {
      testFederatedAddressDivert(false, false, true, false);
   }

   //Test creating address first followed by divert
   //Test creating consumer before divert
   //Test destroy divert at end
   @Test
   public void testUpstreamDivertAddressFirstDivertSecondDestroyDivert() throws Exception {
      testFederatedAddressDivert(false, true, false, true);
   }

   //Test creating address first followed by divert
   //Test creating consumer before divert
   //Test destroy queue at end
   @Test
   public void testUpstreamDivertAddressFirstDivertSecondDestroyQueue() throws Exception {
      testFederatedAddressDivert(false, true, false, false);
   }

   //Test creating divert first followed by address
   //Test creating consumer before divert
   //Test destroy divert at end
   @Test
   public void testUpstreamsDivertAddressAndDivertSecondDestroyDivert() throws Exception {
      testFederatedAddressDivert(false, false, false, true);
   }

   //Test creating divert first followed by address
   //Test creating consumer before divert
   //Test destroy queue at end
   @Test
   public void testUpstreamDivertAddressAndDivertSecondDestroyQueue() throws Exception {
      testFederatedAddressDivert(false, false, false, false);
   }

   protected void testFederatedAddressDivert(boolean downstream, boolean addressFirst, boolean divertBeforeConsumer,
                                             boolean destroyDivert) throws Exception {
      String address = getName();
      String address2 = "fedOneWayDivertTest";

      if (addressFirst) {
         getServer(0).addAddressInfo(new AddressInfo(SimpleString.of(address), RoutingType.MULTICAST));
      }

      final FederationConfiguration federationConfiguration;
      final int deployServer;
      if (downstream) {
         federationConfiguration = FederatedTestUtil.createAddressDownstreamFederationConfiguration(
            "server0", address, getServer(1).getConfiguration().getTransportConfigurations("server1")[0]);
         deployServer = 1;
      } else {
         federationConfiguration = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server1", address);
         deployServer = 0;
      }

      FederationAddressPolicyConfiguration policy = (FederationAddressPolicyConfiguration) federationConfiguration.getFederationPolicyMap().get("AddressPolicy" + address);
      //enable listening for divert bindings
      policy.setEnableDivertBindings(true);
      getServer(deployServer).getConfiguration().getFederationConfigurations().add(federationConfiguration);
      getServer(deployServer).getFederationManager().deploy();

      ConnectionFactory cf1 = getCF(1);
      ConnectionFactory cf0 = getCF(0);
      try (Connection connection1 = cf1.createConnection();
           Connection connection0 = cf0.createConnection()) {
         connection1.start();
         connection0.start();

         Session session1 = connection1.createSession();
         Topic topic1 = session1.createTopic(address);
         MessageProducer producer1 = session1.createProducer(topic1);

         if (divertBeforeConsumer) {
            getServer(0).deployDivert(new DivertConfiguration().setName(address + ":" + address2)
                   .setAddress(address).setExclusive(true).setForwardingAddress(address2)
                   .setRoutingType(ComponentConfigurationRoutingType.ANYCAST));
         }

         Session session0 = connection0.createSession();
         Queue queue0 = session0.createQueue(address2);
         MessageConsumer consumer0 = session0.createConsumer(queue0);

         if (!addressFirst) {
            getServer(0).addAddressInfo(new AddressInfo(SimpleString.of(address), RoutingType.MULTICAST));
         }

         if (!divertBeforeConsumer) {
            getServer(0).deployDivert(new DivertConfiguration().setName(address + ":" + address2)
                                         .setAddress(address).setExclusive(true).setForwardingAddress(address2)
                                         .setRoutingType(ComponentConfigurationRoutingType.ANYCAST));
         }

         Wait.assertEquals(1L, () -> getServer(1).getPostOffice().getBindingsForAddress(SimpleString.of(address)).getBindings().size(), 10_000, 100);
         final QueueBinding remoteQueueBinding = (QueueBinding) getServer(1).getPostOffice().getBindingsForAddress(SimpleString.of(address))
            .getBindings().iterator().next();
         Wait.assertEquals(1, () -> remoteQueueBinding.getQueue().getConsumerCount());

         producer1.send(session1.createTextMessage("hello"));
         assertNotNull(consumer0.receive(1000));

         //Test consumer is cleaned up after divert destroyed
         if (destroyDivert) {
            getServer(0).destroyDivert(SimpleString.of(address + ":" + address2));
         //Test consumer is cleaned up after queue destroyed
         } else {
            getServer(0).destroyQueue(SimpleString.of(address2), null, false);
         }
         assertTrue(Wait.waitFor(() -> remoteQueueBinding.getQueue().getConsumerCount() == 0, 2000, 100));
      }
   }

   private void testFederatedAddressReplication(String address) throws Exception {

      ConnectionFactory cf1 = getCF(1);
      ConnectionFactory cf0 = getCF(0);
      try (Connection connection1 = cf1.createConnection();
           Connection connection0 = cf0.createConnection()) {
         connection1.start();
         connection0.start();

         Session session1 = connection1.createSession();
         Topic topic1 = session1.createTopic(address);
         MessageProducer producer = session1.createProducer(topic1);
         producer.send(session1.createTextMessage("hello"));


         Session session0 = connection0.createSession();
         Topic topic0 = session0.createTopic(address);
         MessageConsumer consumer0 = session0.createConsumer(topic0);

         assertTrue(Wait.waitFor(() -> getServer(1).getPostOffice().getBindingsForAddress(
            SimpleString.of(address)).getBindings().size() == 1, 2000, 100));

         producer.send(session1.createTextMessage("hello"));

         assertNotNull(consumer0.receive(1000));


         producer.send(session1.createTextMessage("hello"));

         assertNotNull(consumer0.receive(1000));

         MessageConsumer consumer1 = session1.createConsumer(topic1);

         producer.send(session1.createTextMessage("hello"));

         assertNotNull(consumer1.receive(1000));
         assertNotNull(consumer0.receive(1000));
         consumer1.close();

         //Groups
         producer.send(session1.createTextMessage("hello"));
         assertNotNull(consumer0.receive(1000));

         producer.send(createTextMessage(session1, "groupA"));

         assertNotNull(consumer0.receive(1000));
         consumer1 = session1.createConsumer(topic1);

         producer.send(createTextMessage(session1, "groupA"));
         assertNotNull(consumer1.receive(1000));
         assertNotNull(consumer0.receive(1000));

      }

   }

   @Test
   public void testFederatedAddressDeployAfterQueuesExist() throws Exception {
      String address = getName();

      ConnectionFactory cf1 = getCF(1);
      ConnectionFactory cf0 = getCF(0);
      try (Connection connection1 = cf1.createConnection();
           Connection connection0 = cf0.createConnection()) {
         connection1.start();
         connection0.start();

         Session session1 = connection1.createSession();
         Topic topic1 = session1.createTopic(address);
         MessageProducer producer = session1.createProducer(topic1);
         producer.send(session1.createTextMessage("hello"));


         Session session0 = connection0.createSession();
         Topic topic0 = session0.createTopic(address);
         MessageConsumer consumer0 = session0.createConsumer(topic0);


         producer.send(session1.createTextMessage("hello"));
         assertNull(consumer0.receive(100));

         FederationConfiguration federationConfiguration = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server1", address);
         getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
         getServer(0).getFederationManager().deploy();

         Wait.waitFor(() -> getServer(1).getPostOffice().getBindingsForAddress(
            SimpleString.of(address)).getBindings().size() == 1, 2000, 100);

         producer.send(session1.createTextMessage("hello"));
         assertNotNull(consumer0.receive(1000));
      }
   }

   @Test
   public void testFederatedAddressRemoteBrokerRestart() throws Exception {
      String address = getName();

      FederationConfiguration federationConfiguration = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server1", address);
      getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
      getServer(0).getFederationManager().deploy();

      ConnectionFactory cf1 = getCF(1);
      ConnectionFactory cf0 = getCF(0);
      try (Connection connection1 = cf1.createConnection();
           Connection connection0 = cf0.createConnection()) {
         connection1.start();
         connection0.start();

         Session session1 = connection1.createSession();
         Topic topic1 = session1.createTopic(address);
         MessageProducer producer = session1.createProducer(topic1);
         producer.send(session1.createTextMessage("hello"));


         Session session0 = connection0.createSession();
         Topic topic0 = session0.createTopic(address);
         MessageConsumer consumer0 = session0.createConsumer(topic0);

         Wait.waitFor(() -> getServer(1).getPostOffice().getBindingsForAddress(SimpleString.of(address)).getBindings().size() == 1);

         producer.send(session1.createTextMessage("hello"));

         assertNotNull(consumer0.receive(10000));


         producer.send(session1.createTextMessage("hello"));

         assertNotNull(consumer0.receive(1000));

         connection1.close();
         getServer(1).stop();
         Wait.waitFor(() -> !getServer(1).isStarted());

         assertNull(consumer0.receive(100));

         getServer(1).start();
         Wait.waitFor(() -> getServer(1).isActive());

         Connection c1 = cf1.createConnection();
         c1.start();
         Wait.waitFor(() -> getServer(1).isStarted());
         session1 = c1.createSession();
         topic1 = session1.createTopic(address);
         producer = session1.createProducer(topic1);

         Wait.waitFor(() -> getServer(1).getPostOffice().getBindingsForAddress(SimpleString.of(address)).getBindings().size() == 1);
         producer.send(session1.createTextMessage("hello"));

         assertNotNull(consumer0.receive(1000));
      }
   }

   @Test
   public void testFederatedAddressLocalBrokerRestart() throws Exception {
      String address = getName();

      FederationConfiguration federationConfiguration = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server1", address);
      getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
      getServer(0).getFederationManager().deploy();

      ConnectionFactory cf1 = getCF(1);
      ConnectionFactory cf0 = getCF(0);
      try (Connection connection1 = cf1.createConnection();
           Connection connection0 = cf0.createConnection()) {
         connection1.start();
         connection0.start();

         Session session1 = connection1.createSession();
         Topic topic1 = session1.createTopic(address);
         MessageProducer producer = session1.createProducer(topic1);
         producer.send(session1.createTextMessage("hello"));


         Session session0 = connection0.createSession();
         Topic topic0 = session0.createTopic(address);
         MessageConsumer consumer0 = session0.createConsumer(topic0);

         Wait.waitFor(() -> getServer(1).getPostOffice().getBindingsForAddress(SimpleString.of(address)).getBindings().size() == 1);

         producer.send(session1.createTextMessage("hello"));

         assertNotNull(consumer0.receive(10000));


         producer.send(session1.createTextMessage("hello"));

         assertNotNull(consumer0.receive(1000));

         connection0.close();
         getServer(0).stop();
         Wait.waitFor(() -> !getServer(0).isStarted());

         producer.send(session1.createTextMessage("hello"));

         getServer(0).start();
         Wait.waitFor(() -> getServer(0).isActive());

         Connection newConnection = getCF(0).createConnection();
         newConnection.start();
         session0 = newConnection.createSession();
         topic0 =  session0.createTopic(address);
         consumer0 = session0.createConsumer(topic0);

         Wait.waitFor(() -> getServer(1).getPostOffice().getBindingsForAddress(SimpleString.of(address)).getBindings().size() == 1);
         producer.send(session1.createTextMessage("hello"));

         assertNotNull(consumer0.receive(1000));

         newConnection.close();
      }
   }

   @Test
   public void testFederatedAddressChainOfBrokers() throws Exception {
      String address = getName();

      //Set queue up on all three brokers
//      for (int i = 0; i < 3; i++) {
//         getServer(i).createQueue(QueueConfiguration.of(queueName).setRoutingType(RoutingType.ANYCAST));
//      }

      //Connect broker 0 (consumer will be here at end of chain) to broker 1
      FederationConfiguration federationConfiguration0 = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server1", address, 2);
      getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration0);
      getServer(0).getFederationManager().deploy();

      //Connect broker 1 (middle of chain) to broker 2
      FederationConfiguration federationConfiguration1 = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server2", address, 2);
      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();
         Topic topic0 = session0.createTopic(address);

         connection2.start();
         Session session2 = connection2.createSession();
         Topic topic2 = session2.createTopic(address);

         MessageProducer producer2 = session2.createProducer(topic2);
         MessageConsumer consumer0 = session0.createConsumer(topic0);

         assertTrue(Wait.waitFor(() -> getServer(1).getPostOffice().getBindingsForAddress(SimpleString.of(address)).getBindings().size() == 1));
         assertTrue(Wait.waitFor(() -> getServer(2).getPostOffice().getBindingsForAddress(SimpleString.of(address)).getBindings().size() == 1));

         //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 testUpstreamFederatedAddressWithCompressedMessage() throws Exception {
      final String DATA_REGULAR = RandomUtil.randomAlphaNumericString(10);
      final String DATA_COMPRESSED_REGULAR = "Compresses easily".repeat(500 * 1024);
      final String DATA_COMPRESSED_LARGE = RandomUtil.randomAlphaNumericString(1024 * 1024);
      final String address = getName();

      FederationConfiguration federationConfiguration = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server1", address);
      getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
      getServer(0).getFederationManager().deploy();

      FederationConfiguration federationConfiguration2 = FederatedTestUtil.createAddressUpstreamFederationConfiguration("server0", address);
      getServer(1).getConfiguration().getFederationConfigurations().add(federationConfiguration2);
      getServer(1).getFederationManager().deploy();

      ActiveMQConnectionFactory cf0 = new ActiveMQConnectionFactory("vm://" + 0);
      ActiveMQConnectionFactory cf1 = new ActiveMQConnectionFactory("vm://" + 1);
      cf0.setCompressLargeMessage(true);
      cf1.setCompressLargeMessage(true);

      try (Connection connection1 = cf1.createConnection();
           Connection connection0 = cf0.createConnection()) {
         connection1.start();
         connection0.start();

         Session session0 = connection0.createSession();
         Session session1 = connection1.createSession();
         Topic topic0 = session0.createTopic(address);
         Topic topic1 = session1.createTopic(address);
         MessageConsumer consumer0 = session0.createConsumer(topic0);
         MessageConsumer consumer1 = session1.createConsumer(topic1);

         MessageProducer producer = session0.createProducer(topic0);

         Wait.assertEquals(2, () -> getServer(0).getPostOffice().getBindingsForAddress(SimpleString.of(address)).getBindings().size(), 5000, 100);
         Wait.assertEquals(2, () -> getServer(1).getPostOffice().getBindingsForAddress(SimpleString.of(address)).getBindings().size(), 5000, 100);

         sendAndConsume(producer, session0, DATA_REGULAR, "regular", consumer0, consumer1);
         sendAndConsume(producer, session0, DATA_COMPRESSED_LARGE, "compressedLarge", consumer0, consumer1);
         sendAndConsume(producer, session0, DATA_COMPRESSED_REGULAR, "compressedRegular", consumer0, consumer1);
      }
   }

   private void sendAndConsume(MessageProducer producer,
                          Session session0,
                          String data,
                          String identification,
                          MessageConsumer consumer0,
                          MessageConsumer consumer1) throws Exception {
      {
         TextMessage message = session0.createTextMessage(data);
         message.setStringProperty("identification", identification);
         producer.send(message);
      }
      TextMessage message0 = (TextMessage) consumer0.receive(10_000);
      TextMessage message1 = (TextMessage) consumer1.receive(10_000);
      assertNotNull(message0);
      assertNotNull(message1);
      assertEquals(data, message0.getText());
      assertEquals(data, message1.getText());
   }

   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;
      }
   }

}
