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

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 static org.junit.jupiter.api.Assertions.fail;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.Interceptor;
import org.apache.activemq.artemis.api.core.JsonUtil;
import org.apache.activemq.artemis.api.core.Message;
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.api.core.TransportConfiguration;
import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
import org.apache.activemq.artemis.api.core.client.ClientConsumer;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.ClientProducer;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
import org.apache.activemq.artemis.api.core.client.ServerLocator;
import org.apache.activemq.artemis.api.core.management.BridgeControl;
import org.apache.activemq.artemis.core.client.impl.ServerLocatorImpl;
import org.apache.activemq.artemis.core.config.BridgeConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.TransformerConfiguration;
import org.apache.activemq.artemis.core.io.SequentialFileFactory;
import org.apache.activemq.artemis.core.io.nio.NIOSequentialFileFactory;
import org.apache.activemq.artemis.core.journal.PreparedTransactionInfo;
import org.apache.activemq.artemis.core.journal.RecordInfo;
import org.apache.activemq.artemis.core.journal.impl.JournalImpl;
import org.apache.activemq.artemis.core.management.impl.view.ConnectionField;
import org.apache.activemq.artemis.core.management.impl.view.predicate.ActiveMQFilterPredicate;
import org.apache.activemq.artemis.core.persistence.impl.journal.DescribeJournal;
import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
import org.apache.activemq.artemis.core.postoffice.DuplicateIDCache;
import org.apache.activemq.artemis.core.postoffice.impl.PostOfficeImpl;
import org.apache.activemq.artemis.core.protocol.core.Packet;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendMessage;
import org.apache.activemq.artemis.core.remoting.impl.invm.TransportConstants;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.cluster.Bridge;
import org.apache.activemq.artemis.core.server.cluster.impl.BridgeImpl;
import org.apache.activemq.artemis.core.server.cluster.impl.BridgeMetrics;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.ServiceRegistryImpl;
import org.apache.activemq.artemis.core.server.transformer.AddHeadersTransformer;
import org.apache.activemq.artemis.core.server.transformer.Transformer;
import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
import org.apache.activemq.artemis.json.JsonArray;
import org.apache.activemq.artemis.json.JsonObject;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.utils.ReusableLatch;
import org.apache.activemq.artemis.utils.collections.LinkedListIterator;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@ExtendWith(ParameterizedTestExtension.class)
public class BridgeTest extends ActiveMQTestBase {

   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

   private ActiveMQServer server0;
   private ActiveMQServer server1;
   private ServerLocator locator;

   private final boolean netty;

   @Parameters(name = "isNetty={0}")
   public static Collection getParameters() {
      return Arrays.asList(new Object[][]{{true}, {false}});
   }

   public BridgeTest(boolean isNetty) {
      this.netty = isNetty;
   }

   protected boolean isNetty() {
      return netty;
   }

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

   private String getConnector() {
      if (isNetty()) {
         return NETTY_CONNECTOR_FACTORY;
      } else {
         return INVM_CONNECTOR_FACTORY;
      }
   }

   @TestTemplate
   public void testSimpleBridge() throws Exception {
      internaltestSimpleBridge(false, false);
   }

   @TestTemplate
   public void testSimpleBridgeFiles() throws Exception {
      internaltestSimpleBridge(false, true);
   }

   @TestTemplate
   public void testSimpleBridgeLargeMessageNullPersistence() throws Exception {
      internaltestSimpleBridge(true, false);
   }

   @TestTemplate
   public void testSimpleBridgeLargeMessageFiles() throws Exception {
      internaltestSimpleBridge(true, true);
   }

   @TestTemplate
   public void testLargeMessageBridge() throws Exception {
      long time = System.currentTimeMillis();
      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, true, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "queue1";

      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);

      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      connectors.put(server1tc.getName(), server1tc);
      server0.getConfiguration().setConnectorConfigurations(connectors);

      final int messageSize = 1024 * 200;

      final int numMessages = 10;

      List<String> connectorConfig = new ArrayList<>();
      connectorConfig.add(server1tc.getName());
      BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(1000).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(false).setConfirmationWindowSize(numMessages * messageSize / 2).setStaticConnectors(connectorConfig);

      List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
      bridgeConfigs.add(bridgeConfiguration);
      server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

      QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
      queueConfigs0.add(queueConfig0);
      server0.getConfiguration().setQueueConfigs(queueConfigs0);

      QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(forwardAddress);
      List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
      queueConfigs1.add(queueConfig1);
      server1.getConfiguration().setQueueConfigs(queueConfigs1);

      server1.start();
      server0.start();
      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = addSessionFactory(locator.createSessionFactory(server0tc));

      ClientSessionFactory sf1 = addSessionFactory(locator.createSessionFactory(server1tc));

      ClientSession session0 = sf0.createSession(false, true, true);

      ClientSession session1 = sf1.createSession(false, true, true);

      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

      ClientConsumer consumer1 = session1.createConsumer(queueName1);

      session1.start();

      final byte[] bytes = new byte[messageSize];

      final SimpleString propKey = SimpleString.of("testkey");

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);

         message.putIntProperty(propKey, i);

         message.getBodyBuffer().writeBytes(bytes);

         producer0.send(message);
      }

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer1.receive(5000);

         assertNotNull(message);

         assertEquals(i, message.getObjectProperty(propKey));

         readLargeMessages(message, 10);

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      session0.close();

      session1.close();

      sf0.close();

      sf1.close();

      closeFields();
      if (server0.getConfiguration().isPersistenceEnabled()) {
         assertEquals(0, loadQueues(server0).size());
      }
      long timeTaken = System.currentTimeMillis() - time;
   }

   @TestTemplate
   public void testBlockedBridgeAndReconnect() throws Exception {
      long time = System.currentTimeMillis();
      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, true, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);
      server1.getAddressSettingsRepository().clear();
      server1.getAddressSettingsRepository().addMatch("#", new AddressSettings().setMaxSizeBytes(10124 * 10).setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK));

      server0.getAddressSettingsRepository().clear();
      server0.getAddressSettingsRepository().addMatch("#", new AddressSettings().setMaxSizeBytes(Long.MAX_VALUE).setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK));

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "queue1";

      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);

      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      connectors.put(server1tc.getName(), server1tc);
      server0.getConfiguration().setConnectorConfigurations(connectors);

      final int messageSize = 1024;

      final int numMessages = 1000;

      List<String> connectorConfig = new ArrayList<>();
      connectorConfig.add(server1tc.getName());
      BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(100).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(true).setConfirmationWindowSize(numMessages * messageSize / 2).setStaticConnectors(connectorConfig).setProducerWindowSize(1024);

      List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
      bridgeConfigs.add(bridgeConfiguration);
      server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

      QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
      queueConfigs0.add(queueConfig0);
      server0.getConfiguration().setQueueConfigs(queueConfigs0);

      QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(forwardAddress);
      List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
      queueConfigs1.add(queueConfig1);
      server1.getConfiguration().setQueueConfigs(queueConfigs1);

      server1.start();
      server0.start();
      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = addSessionFactory(locator.createSessionFactory(server0tc));

      ClientSessionFactory sf1 = addSessionFactory(locator.createSessionFactory(server1tc));

      ClientSession session0 = sf0.createSession(false, true, 0);
      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

      ClientSession session1 = sf1.createSession(true, true, 0);
      ClientConsumer consumer1 = session1.createConsumer(queueName1);

      session1.start();

      final byte[] bytes = new byte[messageSize];

      final SimpleString propKey = SimpleString.of("testkey");

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);

         message.putIntProperty(propKey, i);

         message.getBodyBuffer().writeBytes(bytes);

         producer0.send(message);

         if (i % 100 == 0) {
            session0.commit();
         }
      }
      session0.commit();

      for (int i = 0; i < numMessages / 2; i++) {
         ClientMessage message = consumer1.receive(5000);

         assertNotNull(message);

         assertEquals(i, message.getObjectProperty(propKey));

         message.acknowledge();
      }
      session1.commit();

      BridgeImpl bridge = (BridgeImpl) server0.getClusterManager().getBridges().get("bridge1");

      // stop in the middle. wait the bridge to block
      Wait.assertTrue("bridge is never blocked", bridge::isBlockedOnFlowControl);

      session1.close();
      sf1.close();

      // now restart the server.. the bridge should be reconnecting now
      server1.stop();
      server1.start();

      sf1 = addSessionFactory(locator.createSessionFactory(server1tc));
      session1 = sf1.createSession(true, true, 0);
      consumer1 = session1.createConsumer(queueName1);
      session1.start();

      // consume the rest of the messages
      for (int i = numMessages / 2; i < numMessages; i++) {
         ClientMessage message = consumer1.receive(5000);

         assertNotNull(message);

         assertEquals(i, message.getObjectProperty(propKey));

         message.acknowledge();
      }

      Wait.assertEquals(0, server0.locateQueue(SimpleString.of("queue0"))::getMessageCount);

      assertNull(consumer1.receiveImmediate());

      session0.close();

      session1.close();

      sf0.close();

      sf1.close();

      closeFields();
      if (server0.getConfiguration().isPersistenceEnabled()) {
         assertEquals(0, loadQueues(server0).size());
      }
      long timeTaken = System.currentTimeMillis() - time;
   }

   public void internaltestSimpleBridge(final boolean largeMessage, final boolean useFiles) throws Exception {
      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, useFiles, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, useFiles, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "queue1";

      // Map<String, TransportConfiguration> connectors = new HashMap<>();
      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);

      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      connectors.put(server1tc.getName(), server1tc);
      server0.getConfiguration().setConnectorConfigurations(connectors);

      final int messageSize = 1024;

      final int numMessages = 10;

      List<String> connectorConfig = new ArrayList<>();
      connectorConfig.add(server1tc.getName());
      BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(1000).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(false).setConfirmationWindowSize(numMessages * messageSize / 2).setStaticConnectors(connectorConfig);

      List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
      bridgeConfigs.add(bridgeConfiguration);
      server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

      QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
      queueConfigs0.add(queueConfig0);
      server0.getConfiguration().setQueueConfigs(queueConfigs0);

      QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(forwardAddress);
      List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
      queueConfigs1.add(queueConfig1);
      server1.getConfiguration().setQueueConfigs(queueConfigs1);

      server1.start();
      server0.start();
      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = addSessionFactory(locator.createSessionFactory(server0tc));

      ClientSessionFactory sf1 = addSessionFactory(locator.createSessionFactory(server1tc));

      ClientSession session0 = sf0.createSession(false, true, true);

      ClientSession session1 = sf1.createSession(false, true, true);

      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

      ClientConsumer consumer1 = session1.createConsumer(queueName1);

      session1.start();

      final byte[] bytes = new byte[messageSize];

      final SimpleString propKey = SimpleString.of("testkey");

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);

         if (largeMessage) {
            message.setBodyInputStream(ActiveMQTestBase.createFakeLargeStream(10 * 1024));
         }

         message.putIntProperty(propKey, i);

         message.getBodyBuffer().writeBytes(bytes);

         producer0.send(message);
      }

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer1.receive(5000);

         assertNotNull(message);

         assertEquals(i, message.getObjectProperty(propKey));

         if (largeMessage) {
            readLargeMessages(message, 10);
         }

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      session0.close();

      session1.close();

      sf0.close();

      sf1.close();

      assertEquals(1, server0.getClusterManager().getBridges().size());
      BridgeMetrics bridgeMetrics = server0.getClusterManager().getBridges().get("bridge1").getMetrics();
      assertEquals(10, bridgeMetrics.getMessagesPendingAcknowledgement());
      assertEquals(10, bridgeMetrics.getMessagesAcknowledged());

      closeFields();
      if (server0.getConfiguration().isPersistenceEnabled()) {
         assertEquals(0, loadQueues(server0).size());
      }
   }

   @TestTemplate
   public void testClientSessionFactoryLeak() throws Exception {
      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, true, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";

      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      connectors.put(server1tc.getName(), server1tc);
      server0.getConfiguration().setConnectorConfigurations(connectors);

      List<String> connectorConfig = List.of(server1tc.getName());
      // intentionally configure a bridge that will fail to connect and attempt to reconnect multiple times quickly
      server0.getConfiguration().setBridgeConfigurations(List.of(new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(0).setReconnectAttempts(-1).setStaticConnectors(connectorConfig)));
      server0.getConfiguration().setQueueConfigs(List.of(QueueConfiguration.of(queueName0).setAddress(testAddress)));

      server1.start();
      server0.start();

      ServerLocatorImpl serverLocator = (ServerLocatorImpl) ((BridgeImpl)server0.getClusterManager().getBridges().get("bridge1")).getServerLocator();
      Wait.waitFor(() -> serverLocator.getClientSessionFactoryCount() > 1, 500, 10);
      assertTrue(serverLocator.getClientSessionFactoryCount() <= 1);
   }

   @TestTemplate
   public void testBridgeClientID() throws Exception {
      final String clientId = RandomUtil.randomUUIDString();
      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, true, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "queue1";

      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      connectors.put(server1tc.getName(), server1tc);
      server0.getConfiguration().setConnectorConfigurations(connectors);

      List<String> connectorConfig = List.of(server1tc.getName());
      server0.getConfiguration().setBridgeConfigurations(List.of(new BridgeConfiguration().setName(getName()).setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(10).setReconnectAttempts(250).setStaticConnectors(connectorConfig).setClientId(clientId)));
      server0.getConfiguration().setQueueConfigs(List.of(QueueConfiguration.of(queueName0).setAddress(testAddress)));
      server1.getConfiguration().setQueueConfigs(List.of(QueueConfiguration.of(queueName1).setAddress(forwardAddress)));

      server1.start();
      server0.start();

      Wait.assertTrue(()-> server0.getClusterManager().getBridges().get(getName()).isConnected(), 2000, 25);

      String connectionsAsJsonString = server1.getActiveMQServerControl().listConnections(createJsonFilter(ConnectionField.CLIENT_ID.getName(), ActiveMQFilterPredicate.Operation.EQUALS.toString(), clientId), 1, 1);
      JsonObject connectionsAsJsonObject = JsonUtil.readJsonObject(connectionsAsJsonString);
      JsonArray array = (JsonArray) connectionsAsJsonObject.get("data");
      assertEquals(1, array.size(), "number of connections returned from query");
   }

   private void addTargetParameters(final Map<String, Object> server1Params) {
      if (isNetty()) {
         server1Params.put("port", org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_PORT + 1);
      } else {
         server1Params.put(TransportConstants.SERVER_ID_PROP_NAME, 1);
      }
   }

   private void readLargeMessages(final ClientMessage message, int kiloBlocks) {
      byte[] byteRead = new byte[1024];

      for (int j = 0; j < kiloBlocks; j++) {
         message.getBodyBuffer().readBytes(byteRead);
      }
   }

   @TestTemplate
   public void testWithFilter() throws Exception {
      internalTestWithFilter(false, false);
   }

   @TestTemplate
   public void testWithFilterFiles() throws Exception {
      internalTestWithFilter(false, true);
   }

   @TestTemplate
   public void testWithFilterLargeMessages() throws Exception {
      internalTestWithFilter(true, false);
   }

   @TestTemplate
   public void testWithFilterLargeMessagesFiles() throws Exception {
      internalTestWithFilter(true, true);
   }

   public void internalTestWithFilter(final boolean largeMessage, final boolean useFiles) throws Exception {

      final int numMessages = 10;

      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, useFiles, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, useFiles, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "queue1";

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);
      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);
      connectors.put(server1tc.getName(), server1tc);

      server0.getConfiguration().setConnectorConfigurations(connectors);

      final String filterString = "animal='goat'";

      List<String> staticConnectors = new ArrayList<>();
      staticConnectors.add(server1tc.getName());
      BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setFilterString(filterString).setRetryInterval(1000).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(false).setConfirmationWindowSize(0).setStaticConnectors(staticConnectors);

      List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
      bridgeConfigs.add(bridgeConfiguration);
      server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

      QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
      queueConfigs0.add(queueConfig0);
      server0.getConfiguration().setQueueConfigs(queueConfigs0);

      QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(forwardAddress);
      List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
      queueConfigs1.add(queueConfig1);
      server1.getConfiguration().setQueueConfigs(queueConfigs1);

      server1.start();
      server0.start();

      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = locator.createSessionFactory(server0tc);

      ClientSessionFactory sf1 = locator.createSessionFactory(server1tc);

      ClientSession session0 = sf0.createSession(false, true, true);

      ClientSession session1 = sf1.createSession(false, true, true);

      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

      ClientConsumer consumer1 = session1.createConsumer(queueName1);

      session1.start();

      final SimpleString propKey = SimpleString.of("testkey");

      final SimpleString selectorKey = SimpleString.of("animal");

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);

         message.putIntProperty(propKey, i);

         message.putStringProperty(selectorKey, SimpleString.of("monkey"));

         if (largeMessage) {
            message.setBodyInputStream(ActiveMQTestBase.createFakeLargeStream(10 * 1024));
         }

         producer0.send(message);
      }

      assertNull(consumer1.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);

         message.putIntProperty(propKey, i);

         message.putStringProperty(selectorKey, SimpleString.of("goat"));

         if (largeMessage) {
            message.setBodyInputStream(ActiveMQTestBase.createFakeLargeStream(10 * 1024));
         }

         producer0.send(message);
      }

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer1.receive(4000);

         assertNotNull(message);

         assertEquals("goat", message.getStringProperty(selectorKey));

         assertEquals(i, message.getObjectProperty(propKey));

         message.acknowledge();

         if (largeMessage) {
            readLargeMessages(message, 10);
         }
      }

      session0.commit();

      session1.commit();

      assertNull(consumer1.receiveImmediate());

      session0.close();

      session1.close();

      sf0.close();

      sf1.close();
      closeFields();
      if (useFiles) {
         Map<Long, AtomicInteger> counters = loadQueues(server0);
         assertEquals(1, counters.size());
         Long key = counters.keySet().iterator().next();

         AtomicInteger value = counters.get(key);
         assertNotNull(value);
         assertEquals(numMessages, counters.get(key).intValue());
      }

   }

   @TestTemplate
   public void testStartLater() throws Exception {
      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, true, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "forwardAddress";

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);
      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);
      connectors.put(server1tc.getName(), server1tc);

      server0.getConfiguration().setConnectorConfigurations(connectors);

      List<String> staticConnectors = new ArrayList<>();
      staticConnectors.add(server1tc.getName());
      BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(100).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(false).setConfirmationWindowSize(1024).setStaticConnectors(staticConnectors);

      List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
      bridgeConfigs.add(bridgeConfiguration);
      server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

      QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
      queueConfigs0.add(queueConfig0);
      server0.getConfiguration().setQueueConfigs(queueConfigs0);

      server0.start();

      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = locator.createSessionFactory(server0tc);

      ClientSession session0 = sf0.createSession(false, true, true);

      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

      final int numMessages = 100;

      final SimpleString propKey = SimpleString.of("testkey");

      final SimpleString selectorKey = SimpleString.of("animal");

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);

         message.getBodyBuffer().writeBytes(new byte[1024]);

         message.putIntProperty(propKey, i);

         message.putStringProperty(selectorKey, SimpleString.of("monkey" + i));

         producer0.send(message);
      }

      server1.start();

      Wait.assertTrue(server1::isActive);


      ClientSessionFactory sf1 = locator.createSessionFactory(server1tc);

      ClientSession session1 = sf1.createSession(false, true, true);

      try {
         session1.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress).setRoutingType(RoutingType.ANYCAST));
      } catch (Throwable ignored) {
         ignored.printStackTrace();
      }

      ClientConsumer consumer1 = session1.createConsumer(queueName1);

      session1.start();

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer1.receive(5000);
         assertNotNull(message);
         message.acknowledge();
      }

      session1.commit();

      assertNull(consumer1.receiveImmediate());

      consumer1.close();

      session1.deleteQueue(queueName1);

      session1.close();

      sf1.close();

      server1.stop();

      session0.close();

      sf0.close();
      closeFields();
      assertEquals(0, loadQueues(server0).size());

   }

   @TestTemplate
   public void testWithDuplicates() throws Exception {
      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, true, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String secondQueue = "queue1";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "forwardQueue";

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);
      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);
      connectors.put(server1tc.getName(), server1tc);

      server0.getConfiguration().setConnectorConfigurations(connectors);

      List<String> staticConnectors = new ArrayList<>();
      staticConnectors.add(server1tc.getName());
      BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(100).setReconnectAttemptsOnSameNode(-1).setConfirmationWindowSize(0).setStaticConnectors(staticConnectors);

      List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
      bridgeConfigs.add(bridgeConfiguration);
      server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

      QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
      queueConfigs0.add(queueConfig0);
      queueConfig0 = QueueConfiguration.of(secondQueue).setAddress(testAddress);
      queueConfigs0.add(queueConfig0);
      server0.getConfiguration().setQueueConfigs(queueConfigs0);

      server0.start();

      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = locator.createSessionFactory(server0tc);

      ClientSession session0 = sf0.createSession(false, true, true);

      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

      final int numMessages = 1000;

      final SimpleString propKey = SimpleString.of("testkey");

      final SimpleString selectorKey = SimpleString.of("animal");

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);

         message.getBodyBuffer().writeBytes(new byte[1024]);

         message.putIntProperty(propKey, i);

         message.putStringProperty(selectorKey, SimpleString.of("monkey" + i));

         producer0.send(message);
      }

      server1.start();

      // Inserting the duplicateIDs so the bridge will fail in a few
      {
         long[] ids = new long[100];

         Queue queue = server0.locateQueue(SimpleString.of(queueName0));
         LinkedListIterator<MessageReference> iterator = queue.iterator();

         for (int i = 0; i < 100; i++) {
            assertTrue(iterator.hasNext());
            ids[i] = iterator.next().getMessage().getMessageID();
         }

         iterator.close();

         DuplicateIDCache duplicateTargetCache = server1.getPostOffice().getDuplicateIDCache(PostOfficeImpl.BRIDGE_CACHE_STR.concat(forwardAddress));

         TransactionImpl tx = new TransactionImpl(server1.getStorageManager());
         for (long id : ids) {
            byte[] duplicateArray = BridgeImpl.getDuplicateBytes(server0.getNodeManager().getUUID(), id);
            duplicateTargetCache.addToCache(duplicateArray, tx);
         }
         tx.commit();
      }


      ClientSessionFactory sf1 = locator.createSessionFactory(server1tc);

      ClientSession session1 = sf1.createSession(false, true, true);

      try {
         session1.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress).setRoutingType(RoutingType.ANYCAST));
      } catch (Throwable ignored) {
         ignored.printStackTrace();
      }

      ClientConsumer consumer1 = session1.createConsumer(queueName1);

      session1.start();

      for (int i = 100; i < numMessages; i++) {
         ClientMessage message = consumer1.receive(5000);
         assertNotNull(message);
         assertEquals(i, message.getIntProperty(propKey).intValue());
         message.acknowledge();
      }

      session1.commit();

      assertNull(consumer1.receiveImmediate());

      ClientConsumer otherConsumer = session0.createConsumer(secondQueue);
      session0.start();
      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = otherConsumer.receive(5000);
         assertNotNull(message);
         // This is validating the Bridge is not messing up with the original message
         // and should make a copy of the message before sending it
         assertEquals(2, message.getPropertyNames().size());
         assertEquals(i, message.getIntProperty(propKey).intValue());
         assertEquals(SimpleString.of("monkey" + i), message.getSimpleStringProperty(selectorKey));
         message.acknowledge();

      }

      consumer1.close();

      session1.deleteQueue(queueName1);

      session1.close();

      sf1.close();

      SimpleString queueName1Str = SimpleString.of(queueName1);
      Wait.assertTrue(() -> server1.locateQueue(queueName1Str) == null);

      server1.stop();

      session0.close();

      sf0.close();

      closeFields();

      Wait.assertEquals(0, () -> loadQueues(server0).size());

   }

   private void closeFields() throws Exception {
      locator.close();
      server0.stop();
      server1.stop();
   }

   @TestTemplate
   public void testWithTransformer() throws Exception {
      internaltestWithTransformer(false);
   }

   @TestTemplate
   public void testWithTransformerFiles() throws Exception {
      internaltestWithTransformer(true);
   }

   private void internaltestWithTransformer(final boolean useFiles) throws Exception {
      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, false, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, false, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "queue1";

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);
      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);
      connectors.put(server1tc.getName(), server1tc);

      server0.getConfiguration().setConnectorConfigurations(connectors);

      List<String> staticConnectors = new ArrayList<>();
      staticConnectors.add(server1tc.getName());

      BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setTransformerConfiguration(new TransformerConfiguration(SimpleTransformer.class.getName())).setRetryInterval(1000).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(false).setConfirmationWindowSize(1024).setStaticConnectors(staticConnectors);

      List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
      bridgeConfigs.add(bridgeConfiguration);
      server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

      QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
      queueConfigs0.add(queueConfig0);
      server0.getConfiguration().setQueueConfigs(queueConfigs0);

      QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(forwardAddress);
      List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
      queueConfigs1.add(queueConfig1);
      server1.getConfiguration().setQueueConfigs(queueConfigs1);

      server1.start();
      server0.start();

      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = locator.createSessionFactory(server0tc);

      ClientSessionFactory sf1 = locator.createSessionFactory(server1tc);

      ClientSession session0 = sf0.createSession(false, true, true);

      ClientSession session1 = sf1.createSession(false, true, true);

      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

      ClientConsumer consumer1 = session1.createConsumer(queueName1);

      session1.start();

      final int numMessages = 10;

      final SimpleString propKey = SimpleString.of("wibble");

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);

         message.putStringProperty(propKey, SimpleString.of("bing"));

         message.getBodyBuffer().writeString("doo be doo be doo be doo");

         producer0.send(message);
      }

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer1.receive(200);

         assertNotNull(message);

         SimpleString val = (SimpleString) message.getObjectProperty(propKey);

         assertEquals(SimpleString.of("bong"), val);

         String sval = message.getBodyBuffer().readString();

         assertEquals("dee be dee be dee be dee", sval);

         message.acknowledge();

      }

      assertNull(consumer1.receiveImmediate());

      session0.close();

      session1.close();

      sf0.close();

      sf1.close();

      if (server0.getConfiguration().isPersistenceEnabled()) {
         assertEquals(0, loadQueues(server0).size());
      }
   }

   @TestTemplate
   public void testWithTransformerProperties() throws Exception {
      final String propKey = "bridged";
      final String propValue = "true";

      TransformerConfiguration transformerConfiguration = new TransformerConfiguration(AddHeadersTransformer.class.getName());
      transformerConfiguration.getProperties().put(propKey, propValue);

      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, false, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, false, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "queue1";

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);
      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);
      connectors.put(server1tc.getName(), server1tc);

      server0.getConfiguration().setConnectorConfigurations(connectors);

      List<String> staticConnectors = new ArrayList<>();
      staticConnectors.add(server1tc.getName());

      BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setTransformerConfiguration(transformerConfiguration).setRetryInterval(1000).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(false).setConfirmationWindowSize(1024).setStaticConnectors(staticConnectors);

      List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
      bridgeConfigs.add(bridgeConfiguration);
      server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

      QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
      queueConfigs0.add(queueConfig0);
      server0.getConfiguration().setQueueConfigs(queueConfigs0);

      QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(forwardAddress);
      List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
      queueConfigs1.add(queueConfig1);
      server1.getConfiguration().setQueueConfigs(queueConfigs1);

      server1.start();
      server0.start();

      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = locator.createSessionFactory(server0tc);

      ClientSessionFactory sf1 = locator.createSessionFactory(server1tc);

      ClientSession session0 = sf0.createSession(false, true, true);

      ClientSession session1 = sf1.createSession(false, true, true);

      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

      ClientConsumer consumer1 = session1.createConsumer(queueName1);

      session1.start();

      final int numMessages = 10;

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);

         message.getBodyBuffer().writeString("doo be doo be doo be doo");

         producer0.send(message);
      }

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer1.receive(200);

         assertNotNull(message);

         String messagePropVal = message.getStringProperty(propKey);

         assertEquals(propValue, messagePropVal);

         String sval = message.getBodyBuffer().readString();

         assertEquals("doo be doo be doo be doo", sval);

         message.acknowledge();

      }

      assertNull(consumer1.receiveImmediate());

      session0.close();

      session1.close();

      sf0.close();

      sf1.close();

      if (server0.getConfiguration().isPersistenceEnabled()) {
         assertEquals(0, loadQueues(server0).size());
      }
   }

   @TestTemplate
   public void testSawtoothLoad() throws Exception {
      Map<String, Object> server0Params = new HashMap<>();
      ActiveMQServer server0 = createClusteredServerWithParams(isNetty(), 0, true, server0Params);
      server0.getConfiguration().setThreadPoolMaxSize(10);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      ActiveMQServer server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);
      server1.getConfiguration().setThreadPoolMaxSize(10);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "queue1";

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      final TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);
      final TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);
      connectors.put(server1tc.getName(), server1tc);

      server0.getConfiguration().setConnectorConfigurations(connectors);

      List<String> staticConnectors = new ArrayList<>();
      staticConnectors.add(server1tc.getName());

      BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(1000).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(false).setConfirmationWindowSize(0).setStaticConnectors(staticConnectors).setProducerWindowSize(1);

      List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
      bridgeConfigs.add(bridgeConfiguration);
      server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

      QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
      queueConfigs0.add(queueConfig0);
      server0.getConfiguration().setQueueConfigs(queueConfigs0);

      QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(forwardAddress);
      List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
      queueConfigs1.add(queueConfig1);
      server1.getConfiguration().setQueueConfigs(queueConfigs1);

      try {
         server1.start();
         server0.start();

         final int numMessages = 300;

         final int totalrepeats = 3;

         final AtomicInteger errors = new AtomicInteger(0);

         // We shouldn't have more than 10K messages pending
         final Semaphore semop = new Semaphore(10000);

         class ConsumerThread extends Thread {

            @Override
            public void run() {
               try {
                  ServerLocator locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server1tc));

                  ClientSessionFactory sf = createSessionFactory(locator);

                  ClientSession session = sf.createSession(false, false);

                  session.start();

                  ClientConsumer consumer = session.createConsumer(queueName1);

                  for (int i = 0; i < numMessages; i++) {
                     ClientMessage message = consumer.receive(5000);

                     assertNotNull(message);

                     message.acknowledge();
                     semop.release();
                     if (i % 1000 == 0) {
                        session.commit();
                     }
                  }

                  session.commit();

                  session.close();
                  sf.close();
                  locator.close();

               } catch (Throwable e) {
                  e.printStackTrace();
                  errors.incrementAndGet();
               }
            }
         }

         class ProducerThread extends Thread {

            final int nmsg;

            ProducerThread(int nmsg) {
               this.nmsg = nmsg;
            }

            @Override
            public void run() {
               ServerLocator locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc));

               locator.setBlockOnDurableSend(false).setBlockOnNonDurableSend(false);

               ClientSessionFactory sf = null;

               ClientSession session = null;

               ClientProducer producer = null;

               try {
                  sf = createSessionFactory(locator);

                  session = sf.createSession(false, true, true);

                  producer = session.createProducer(SimpleString.of(testAddress));

                  for (int i = 0; i < nmsg; i++) {
                     assertEquals(0, errors.get());
                     ClientMessage message = session.createMessage(true);

                     message.putIntProperty("seq", i);

                     if (i % 100 == 0) {
                        message.setPriority((byte) (RandomUtil.randomPositiveInt() % 9));
                     } else {
                        message.setPriority((byte) 5);
                     }

                     message.getBodyBuffer().writeBytes(new byte[50]);

                     producer.send(message);
                     assertTrue(semop.tryAcquire(1, 10, TimeUnit.SECONDS));
                  }
               } catch (Throwable e) {
                  e.printStackTrace(System.out);
                  errors.incrementAndGet();
               } finally {
                  try {
                     session.close();
                     sf.close();
                     locator.close();
                  } catch (Exception ignored) {
                     errors.incrementAndGet();
                  }
               }
            }
         }

         for (int repeat = 0; repeat < totalrepeats; repeat++) {
            List<Thread> threads = new ArrayList<>();

            threads.add(new ConsumerThread());
            threads.add(new ProducerThread(numMessages / 2));
            threads.add(new ProducerThread(numMessages / 2));

            for (Thread t : threads) {
               t.start();
            }

            for (Thread t : threads) {
               t.join();
            }

            assertEquals(0, errors.get());
         }
      } finally {
         try {
            server0.stop();
         } catch (Exception ignored) {

         }

         try {
            server1.stop();
         } catch (Exception ignored) {

         }
      }

      assertEquals(0, loadQueues(server0).size());

   }

   @TestTemplate
   public void testBridgeWithPaging() throws Exception {
      ActiveMQServer server0 = null;
      ActiveMQServer server1 = null;

      final int PAGE_MAX = 10 * 1024;

      final int PAGE_SIZE = 1 * 1024;
      try {

         Map<String, Object> server0Params = new HashMap<>();
         server0 = createClusteredServerWithParams(isNetty(), 0, true, PAGE_SIZE, PAGE_MAX, server0Params);

         Map<String, Object> server1Params = new HashMap<>();
         addTargetParameters(server1Params);
         server1 = createClusteredServerWithParams(isNetty(), 1, true, PAGE_SIZE, -1, server1Params);
         server1.getConfiguration().setJournalBufferTimeout_AIO(10).setJournalBufferTimeout_NIO(10);

         final String testAddress = "testAddress";
         final String queueName0 = "queue0";
         final String forwardAddress = "forwardAddress";
         final String queueName1 = "queue1";

         Map<String, TransportConfiguration> connectors = new HashMap<>();
         TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);

         TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);
         connectors.put(server1tc.getName(), server1tc);

         server0.getConfiguration().setConnectorConfigurations(connectors);

         server0.getConfiguration().setIDCacheSize(20000).setJournalBufferTimeout_NIO(10).setJournalBufferTimeout_AIO(10);

         List<String> staticConnectors = new ArrayList<>();
         staticConnectors.add(server1tc.getName());

         BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(1).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(true).setConfirmationWindowSize(1).setStaticConnectors(staticConnectors);

         bridgeConfiguration.setCallTimeout(1000);

         List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
         bridgeConfigs.add(bridgeConfiguration);
         server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

         QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
         List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
         queueConfigs0.add(queueConfig0);
         server0.getConfiguration().setQueueConfigs(queueConfigs0);

         QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(forwardAddress);
         List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
         queueConfigs1.add(queueConfig1);
         server1.getConfiguration().setQueueConfigs(queueConfigs1);

         List<String> interceptorToStop = new ArrayList<>();
         interceptorToStop.add(StopInterceptor.class.getName());
         server1.getConfiguration().setIncomingInterceptorClassNames(interceptorToStop);

         StopInterceptor.serverToStop = server0;

         server1.start();
         server0.start();

         locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
         ClientSessionFactory sf0 = locator.createSessionFactory(server0tc);

         ClientSessionFactory sf1 = locator.createSessionFactory(server1tc);

         ClientSession session0 = sf0.createSession(false, false, true);

         ClientSession session1 = sf1.createSession(false, true, true);

         ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

         ClientConsumer consumer1 = session1.createConsumer(queueName1);

         session1.start();

         final int numMessages = 200;

         final SimpleString propKey = SimpleString.of("testkey");

         for (int i = 0; i < numMessages; i++) {
            ClientMessage message = session0.createMessage(true);
            message.getBodyBuffer().writeBytes(new byte[512]);

            message.putIntProperty(propKey, i);

            producer0.send(message);
         }

         session0.commit();

         assertTrue(StopInterceptor.latch.await(1, TimeUnit.HOURS));

         StopInterceptor.thread.join(15000);

         if (StopInterceptor.thread.isAlive()) {
            System.out.println(threadDump("Still alive, stop didn't work!!!"));
            fail("Thread that should restart the server still alive");
         }

         // Restarting the server
         server0.start();

         Map<Integer, AtomicInteger> receivedMsg = new HashMap<>();

         for (int i = 0; i < numMessages; i++) {
            ClientMessage message = consumer1.receive(5000);

            if (message == null) {
               break;
            }

            Integer msgKey = message.getIntProperty(propKey);

            AtomicInteger msgCount = receivedMsg.get(msgKey);

            if (msgKey.intValue() != i) {
               System.err.println("Message " + msgCount + " received out of order, expected to be " + i + " it's acceptable but not the ideal!");
            }

            if (msgCount == null) {
               msgCount = new AtomicInteger();
               receivedMsg.put(msgKey, msgCount);
            }

            msgCount.incrementAndGet();

            if (i % 500 == 0)
               logger.debug("received {}", i);
         }

         boolean failed = false;

         if (consumer1.receiveImmediate() != null) {
            System.err.println("Unexpected message received");
            failed = true;
         }

         for (int i = 0; i < numMessages; i++) {
            AtomicInteger msgCount = receivedMsg.get(i);
            if (msgCount == null) {
               System.err.println("Msg " + i + " wasn't received");
               failed = true;
            } else if (msgCount.get() > 1) {
               System.err.println("msg " + i + " was received " + msgCount.get() + " times");
               failed = true;
            }

         }

         assertFalse(failed, "Test failed");

         session0.close();

         session1.close();

         sf0.close();

         sf1.close();

      } finally {
         if (locator != null) {
            locator.close();
         }
         try {
            server0.stop();
         } catch (Throwable ignored) {
         }

         try {
            server1.stop();
         } catch (Throwable ignored) {
         }
      }

      assertEquals(0, loadQueues(server0).size());

   }

   // Stops a server after 100 messages received
   public static class StopInterceptor implements Interceptor {

      static ActiveMQServer serverToStop;

      static Thread thread;

      static final ReusableLatch latch = new ReusableLatch(0);

      public static void reset() {
         latch.setCount(1);
         serverToStop = null;
         count = 0;
         thread = null;
      }

      static int count = 0;

      @Override
      public synchronized boolean intercept(Packet packet, RemotingConnection connection) throws ActiveMQException {

         if (packet instanceof SessionSendMessage && ++count == 100) {
            try {
               thread = new Thread("***Server Restarter***") {

                  @Override
                  public void run() {
                     try {
                        serverToStop.fail(false);
                        latch.countDown();
                     } catch (Exception e) {
                        e.printStackTrace();
                     }
                  }
               };

               thread.start();

               latch.await();
               return true;
            } catch (Exception e) {
               e.printStackTrace();
            }
         }
         return true;
      }
   }

   @TestTemplate
   public void testBridgeWithLargeMessage() throws Exception {
      ActiveMQServer server0 = null;
      ActiveMQServer server1 = null;

      final int PAGE_MAX = 1024 * 1024;

      final int PAGE_SIZE = 10 * 1024;
      ServerLocator locator = null;
      try {

         Map<String, Object> server0Params = new HashMap<>();
         server0 = createClusteredServerWithParams(isNetty(), 0, true, PAGE_SIZE, PAGE_MAX, server0Params);

         Map<String, Object> server1Params = new HashMap<>();
         addTargetParameters(server1Params);
         server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);

         final String testAddress = "testAddress";
         final String queueName0 = "queue0";
         final String forwardAddress = "forwardAddress";
         final String queueName1 = "queue1";

         Map<String, TransportConfiguration> connectors = new HashMap<>();
         TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);

         TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);
         connectors.put(server1tc.getName(), server1tc);

         server0.getConfiguration().setConnectorConfigurations(connectors);

         List<String> staticConnectors = new ArrayList<>();
         staticConnectors.add(server1tc.getName());

         BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(1000).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(false).setConfirmationWindowSize(1024).setStaticConnectors(staticConnectors);

         List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
         bridgeConfigs.add(bridgeConfiguration);
         server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

         QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
         List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
         queueConfigs0.add(queueConfig0);
         server0.getConfiguration().setQueueConfigs(queueConfigs0);

         QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(forwardAddress);
         List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
         queueConfigs1.add(queueConfig1);
         server1.getConfiguration().setQueueConfigs(queueConfigs1);

         server1.start();
         server0.start();

         locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
         ClientSessionFactory sf0 = locator.createSessionFactory(server0tc);

         ClientSessionFactory sf1 = locator.createSessionFactory(server1tc);

         ClientSession session0 = sf0.createSession(false, true, true);

         ClientSession session1 = sf1.createSession(false, true, true);

         ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

         ClientConsumer consumer1 = session1.createConsumer(queueName1);

         session1.start();

         final int numMessages = 50;

         final SimpleString propKey = SimpleString.of("testkey");

         final int LARGE_MESSAGE_SIZE = 1024;
         for (int i = 0; i < numMessages; i++) {
            ClientMessage message = session0.createMessage(true);
            message.setBodyInputStream(createFakeLargeStream(LARGE_MESSAGE_SIZE));

            message.putIntProperty(propKey, i);

            producer0.send(message);
         }

         session0.commit();

         for (int i = 0; i < numMessages; i++) {
            ClientMessage message = consumer1.receive(5000);

            assertNotNull(message);

            assertEquals(i, message.getObjectProperty(propKey));

            ActiveMQBuffer buff = message.getBodyBuffer();

            for (int posMsg = 0; posMsg < LARGE_MESSAGE_SIZE; posMsg++) {
               assertEquals(getSamplebyte(posMsg), buff.readByte());
            }

            message.acknowledge();
         }

         session1.commit();

         assertNull(consumer1.receiveImmediate());

         session0.close();

         session1.close();

         sf0.close();

         sf1.close();

      } finally {
         if (locator != null) {
            locator.close();
         }
         try {
            server0.stop();
         } catch (Throwable ignored) {
         }

         try {
            server1.stop();
         } catch (Throwable ignored) {
         }
      }

      assertEquals(0, loadQueues(server0).size());
   }

   @TestTemplate
   public void testBridgeWithVeryLargeMessage() throws Exception {
      ActiveMQServer server0 = null;
      ActiveMQServer server1 = null;

      final int PAGE_MAX = 1024 * 1024;

      final int PAGE_SIZE = 10 * 1024;
      ServerLocator locator = null;

      try {
         Map<String, Object> server0Params = new HashMap<>();
         server0 = createClusteredServerWithParams(isNetty(), 0, true, PAGE_SIZE, PAGE_MAX, server0Params);

         Map<String, Object> server1Params = new HashMap<>();
         addTargetParameters(server1Params);
         server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);

         final String testAddress = "testAddress";
         final String queueName0 = "queue0";
         final String forwardAddress = "forwardAddress";
         final String queueName1 = "queue1";

         Map<String, TransportConfiguration> connectors = new HashMap<>();
         TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);

         TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);
         connectors.put(server1tc.getName(), server1tc);

         server0.getConfiguration().setConnectorConfigurations(connectors);

         List<String> staticConnectors = new ArrayList<>();
         staticConnectors.add(server1tc.getName());

         int minLargeMessageSize = 200 * 1024;

         BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setForwardingAddress(forwardAddress).setRetryInterval(1000).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(false).setConfirmationWindowSize(1024).setStaticConnectors(staticConnectors).setMinLargeMessageSize(minLargeMessageSize).setProducerWindowSize(minLargeMessageSize / 2);

         List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
         bridgeConfigs.add(bridgeConfiguration);
         server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

         QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
         List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
         queueConfigs0.add(queueConfig0);
         server0.getConfiguration().setQueueConfigs(queueConfigs0);

         QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(forwardAddress);
         List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
         queueConfigs1.add(queueConfig1);
         server1.getConfiguration().setQueueConfigs(queueConfigs1);

         server1.start();
         server0.start();

         locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));

         ClientSessionFactory sf0 = locator.createSessionFactory(server0tc);

         ClientSessionFactory sf1 = locator.createSessionFactory(server1tc);

         ClientSession session0 = sf0.createSession(false, true, true);

         ClientSession session1 = sf1.createSession(false, true, true);

         ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

         ClientConsumer consumer1 = session1.createConsumer(queueName1);

         session1.start();

         //create a large message bigger than Integer.MAX_VALUE
         final long largeMessageSize = 200 * 1024;

         ClientMessage largeMessage = createLargeMessage(session0, largeMessageSize);

         producer0.send(largeMessage);

         session0.commit();

         //check target queue for large message arriving
         Wait.waitFor(() -> session1.queueQuery(SimpleString.of(queueName1)).getMessageCount() > 0);

         //receive the message
         ClientMessage message = consumer1.receive(5000);
         message.acknowledge();

         File outputFile = new File(getTemporaryDir(), "huge_message_received.dat");

         logger.debug("-----message save to: {}", outputFile.getAbsolutePath());
         FileOutputStream fileOutputStream = new FileOutputStream(outputFile);

         BufferedOutputStream bufferedOutput = new BufferedOutputStream(fileOutputStream);

         message.setOutputStream(bufferedOutput);

         if (!message.waitOutputStreamCompletion(5 * 60 * 1000)) {
            fail("message didn't get received to disk in 5 min. Is the machine slow?");
         }
         session1.commit();

         assertNull(consumer1.receiveImmediate());

         session0.close();

         session1.close();

         sf0.close();

         sf1.close();

      } finally {
         if (locator != null) {
            locator.close();
         }
         try {
            server0.stop();
         } catch (Throwable ignored) {
         }

         try {
            server1.stop();
         } catch (Throwable ignored) {
         }
      }

      assertEquals(0, loadQueues(server0).size());
   }

   private ClientMessage createLargeMessage(ClientSession session, long largeMessageSize) throws Exception {
      File fileInput = new File(getTemporaryDir(), "huge_message_to_send.dat");

      createFile(fileInput, largeMessageSize);

      logger.debug("File created at: {}", fileInput.getAbsolutePath());

      ClientMessage message = session.createMessage(Message.BYTES_TYPE, true);

      FileInputStream fileInputStream = new FileInputStream(fileInput);
      BufferedInputStream bufferedInput = new BufferedInputStream(fileInputStream);

      message.setBodyInputStream(bufferedInput);

      return message;
   }

   private static void createFile(final File file, final long fileSize) throws IOException {
      if (file.exists()) {
         logger.warn("---file already there {}", file.length());
         return;
      }
      FileOutputStream fileOut = new FileOutputStream(file);
      BufferedOutputStream buffOut = new BufferedOutputStream(fileOut);
      byte[] outBuffer = new byte[1024 * 1024];
      logger.debug(" --- creating file, size: {}", fileSize);
      for (long i = 0; i < fileSize; i += outBuffer.length) {
         buffOut.write(outBuffer);
      }
      buffOut.close();
   }

   @TestTemplate
   public void testNullForwardingAddress() throws Exception {
      Map<String, Object> server0Params = new HashMap<>();
      server0 = createClusteredServerWithParams(isNetty(), 0, false, server0Params);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, false, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String queueName1 = "queue1";

      Map<String, TransportConfiguration> connectors = new HashMap<>();
      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params);

      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);
      connectors.put(server1tc.getName(), server1tc);

      server0.getConfiguration().setConnectorConfigurations(connectors);

      final int messageSize = 1024;

      final int numMessages = 10;

      List<String> staticConnectors = new ArrayList<>();
      staticConnectors.add(server1tc.getName());
      // do not set forwarding address (defaults to null) to use messages' original address
      BridgeConfiguration bridgeConfiguration = new BridgeConfiguration().setName("bridge1").setQueueName(queueName0).setRetryInterval(1000).setReconnectAttemptsOnSameNode(-1).setUseDuplicateDetection(false).setConfirmationWindowSize(numMessages * messageSize / 2).setStaticConnectors(staticConnectors);

      List<BridgeConfiguration> bridgeConfigs = new ArrayList<>();
      bridgeConfigs.add(bridgeConfiguration);
      server0.getConfiguration().setBridgeConfigurations(bridgeConfigs);

      QueueConfiguration queueConfig0 = QueueConfiguration.of(queueName0).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs0 = new ArrayList<>();
      queueConfigs0.add(queueConfig0);
      server0.getConfiguration().setQueueConfigs(queueConfigs0);

      // on server #1, we bind queueName1 to same address testAddress
      QueueConfiguration queueConfig1 = QueueConfiguration.of(queueName1).setAddress(testAddress);
      List<QueueConfiguration> queueConfigs1 = new ArrayList<>();
      queueConfigs1.add(queueConfig1);
      server1.getConfiguration().setQueueConfigs(queueConfigs1);

      server1.start();
      server0.start();

      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = locator.createSessionFactory(server0tc);

      ClientSessionFactory sf1 = locator.createSessionFactory(server1tc);

      ClientSession session0 = sf0.createSession(false, true, true);

      ClientSession session1 = sf1.createSession(false, true, true);

      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));

      ClientConsumer consumer1 = session1.createConsumer(queueName1);

      session1.start();

      final byte[] bytes = new byte[messageSize];

      final SimpleString propKey = SimpleString.of("testkey");

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);

         message.putIntProperty(propKey, i);

         message.getBodyBuffer().writeBytes(bytes);

         producer0.send(message);
      }

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer1.receive(200);

         assertNotNull(message);

         assertEquals(i, message.getObjectProperty(propKey));

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      session0.close();

      session1.close();

      sf0.close();

      sf1.close();
      closeFields();
   }

   @TestTemplate
   public void testInjectedTransformer() throws Exception {
      final SimpleString ADDRESS = SimpleString.of("myAddress");
      final SimpleString QUEUE = SimpleString.of("myQueue");
      final String BRIDGE = "myBridge";

      ServiceRegistryImpl serviceRegistry = new ServiceRegistryImpl();
      Transformer transformer = message -> null;

      serviceRegistry.addBridgeTransformer(BRIDGE, transformer);
      Configuration config = createDefaultInVMConfig().addConnectorConfiguration("in-vm", new TransportConfiguration(INVM_CONNECTOR_FACTORY));
      ActiveMQServer server = addServer(new ActiveMQServerImpl(config, null, null, null, serviceRegistry));
      server.start();
      server.waitForActivation(100, TimeUnit.MILLISECONDS);
      server.createQueue(QueueConfiguration.of(QUEUE).setAddress(ADDRESS).setRoutingType(RoutingType.ANYCAST).setDurable(false));
      List<String> connectors = new ArrayList<>();
      connectors.add("in-vm");
      server.deployBridge(new BridgeConfiguration().setName(BRIDGE).setQueueName(QUEUE.toString()).setForwardingAddress(ADDRESS.toString()).setStaticConnectors(connectors));
      Bridge bridge = server.getClusterManager().getBridges().get(BRIDGE);
      assertNotNull(bridge);
      assertEquals(transformer, ((BridgeImpl) bridge).getTransformer());
   }

   @TestTemplate
   public void testDefaultConfirmationWindowSize() throws Exception {
      final SimpleString ADDRESS = SimpleString.of("myAddress");
      final SimpleString QUEUE = SimpleString.of("myQueue");
      final SimpleString FORWARDING_ADDRESS = SimpleString.of("myForwardingAddress");
      final SimpleString FORWARDING_QUEUE = SimpleString.of("myForwardingQueue");
      final String BRIDGE = "myBridge";

      Configuration config = createDefaultConfig(0, isNetty()).addConnectorConfiguration("myConnector", new TransportConfiguration(getConnector()));
      ActiveMQServer server = addServer(new ActiveMQServerImpl(config));
      server.start();
      server.waitForActivation(100, TimeUnit.MILLISECONDS);
      server.createQueue(QueueConfiguration.of(QUEUE).setAddress(ADDRESS).setRoutingType(RoutingType.ANYCAST).setDurable(false));
      server.createQueue(QueueConfiguration.of(FORWARDING_QUEUE).setAddress(FORWARDING_ADDRESS).setRoutingType(RoutingType.ANYCAST).setDurable(false));
      List<String> connectors = new ArrayList<>();
      connectors.add("myConnector");
      server.deployBridge(new BridgeConfiguration()
                             .setName(BRIDGE)
                             .setQueueName(QUEUE.toString())
                             .setForwardingAddress(FORWARDING_ADDRESS.toString())
                             .setStaticConnectors(connectors));

      // now we actually have to use the bridge to make sure it connected correctly
      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(new TransportConfiguration(getConnector())));
      ClientSessionFactory sf = addSessionFactory(locator.createSessionFactory());
      ClientSession session = addClientSession(sf.createSession(false, true, true));
      ClientProducer producer = addClientProducer(session.createProducer(ADDRESS));
      ClientConsumer consumer = addClientConsumer(session.createConsumer(FORWARDING_QUEUE));
      session.start();
      producer.send(session.createMessage(true));
      assertNotNull(consumer.receive(200));
   }

   @TestTemplate
   public void testManagementLeak() throws Exception {
      final SimpleString ADDRESS = SimpleString.of("myAddress");
      final SimpleString QUEUE = SimpleString.of("myQueue");
      final SimpleString FORWARDING_ADDRESS = SimpleString.of("myForwardingAddress");
      final SimpleString FORWARDING_QUEUE = SimpleString.of("myForwardingQueue");
      final String BRIDGE = "myBridge";

      ActiveMQServer server = addServer(new ActiveMQServerImpl(createDefaultConfig(0, isNetty()).addConnectorConfiguration("myConnector", new TransportConfiguration(getConnector()))));
      server.start();
      server.waitForActivation(100, TimeUnit.MILLISECONDS);
      server.createQueue(QueueConfiguration.of(QUEUE)
                            .setAddress(ADDRESS)
                            .setRoutingType(RoutingType.ANYCAST)
                            .setDurable(false));
      server.createQueue(QueueConfiguration.of(FORWARDING_QUEUE)
                            .setAddress(FORWARDING_ADDRESS)
                            .setRoutingType(RoutingType.ANYCAST)
                            .setDurable(false));
      List<String> connectors = new ArrayList<>();
      connectors.add("myConnector");
      final int concurrency = 20;
      BridgeConfiguration config = new BridgeConfiguration()
         .setName(BRIDGE)
         .setQueueName(QUEUE.toString())
         .setForwardingAddress(FORWARDING_ADDRESS.toString())
         .setStaticConnectors(connectors)
         .setConcurrency(concurrency);
      server.deployBridge(config);
      assertEquals(concurrency, server.getManagementService().getResources(BridgeControl.class).length);
      server.destroyBridge(config.getName());

      assertEquals(0, server.getManagementService().getResources(BridgeControl.class).length);
   }

   @TestTemplate
   public void testPendingAcksNeverArriveOnStop() throws Exception {
      testPendingAcksNeverArrive(true, false);
   }

   @TestTemplate
   public void testPendingAcksNeverArriveOnPause() throws Exception {
      testPendingAcksNeverArrive(false, false);
   }

   @TestTemplate
   public void testPendingAcksNeverArriveOnStopWithLargeMessages() throws Exception {
      testPendingAcksNeverArrive(true, true);
   }

   @TestTemplate
   public void testPendingAcksNeverArriveOnPauseWithLargeMessages() throws Exception {
      testPendingAcksNeverArrive(false, true);
   }

   private void testPendingAcksNeverArrive(boolean stop, boolean large) throws Exception {
      server0 = createClusteredServerWithParams(isNetty(), 0, true, null);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "queue1";
      final long pendingAckTimeout = 2000;
      final int messageSize = 1024;
      final int numMessages = 10;

      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), null);
      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);

      server0.getConfiguration()
             .setConnectorConfigurations(Map.of(server1tc.getName(), server1tc))
             .setBridgeConfigurations(Arrays.asList(new BridgeConfiguration()
                                                       .setName("bridge1")
                                                       .setQueueName(queueName0)
                                                       .setForwardingAddress(forwardAddress)
                                                       .setRetryInterval(1000)
                                                       .setReconnectAttemptsOnSameNode(-1)
                                                       .setUseDuplicateDetection(false)
                                                       .setConfirmationWindowSize(numMessages * messageSize / 2)
                                                       .setMinLargeMessageSize(large ? (messageSize / 2) : (messageSize * 2))
                                                       .setPendingAckTimeout(pendingAckTimeout)
                                                       .setStaticConnectors(Arrays.asList(server1tc.getName()))));
      server0.getConfiguration().setQueueConfigs(Arrays.asList(QueueConfiguration.of(queueName0).setAddress(testAddress)));
      server0.start();

      // this interceptor will prevent the target from returning any send acknowledgements
      Interceptor sendBlockingInterceptor = (packet, connection) -> {
         if (packet.getType() == PacketImpl.SESS_SEND || packet.getType() == PacketImpl.SESS_SEND_LARGE) {
            return false;
         }
         return true;
      };

      server1.getConfiguration().setQueueConfigs(Arrays.asList(QueueConfiguration.of(queueName1).setAddress(forwardAddress)));
      server1.start();
      server1.getRemotingService().addIncomingInterceptor(sendBlockingInterceptor);
      Bridge bridge = server0.getClusterManager().getBridges().get("bridge1");
      Wait.assertTrue(() -> (bridge.isConnected()), 2000, 100);

      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = addSessionFactory(locator.createSessionFactory(server0tc));
      ClientSession session0 = sf0.createSession(false, true, true);
      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));
      final byte[] bytes = new byte[messageSize];

      final SimpleString propKey = SimpleString.of("testkey");

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);
         message.putIntProperty(propKey, i);
         message.getBodyBuffer().writeBytes(bytes);
         producer0.send(message);
      }

      session0.close();
      sf0.close();

      Wait.assertEquals((long) numMessages, () -> bridge.getMetrics().getMessagesPendingAcknowledgement(), 2000, 100);
      long start = System.currentTimeMillis();
      BridgeImpl.State desiredState;
      if (stop) {
         bridge.stop();
         desiredState = BridgeImpl.State.STOPPED;
      } else {
         bridge.pause();
         desiredState = BridgeImpl.State.PAUSED;
      }
      Wait.assertEquals(desiredState, () -> ((BridgeImpl)bridge).getState(), pendingAckTimeout, 25);
      assertTrue(System.currentTimeMillis() - start >= pendingAckTimeout);
      Wait.assertEquals((long) numMessages, () -> server0.locateQueue(queueName0).getMessageCount(), 2000, 100);
      Wait.assertEquals(0L, () -> server0.locateQueue(queueName0).getDeliveringCount(), 2000, 100);
   }

   @TestTemplate
   public void testPendingAcksEventuallyArriveOnStop() throws Exception {
      testPendingAcksEventuallyArrive(true, false);
   }

   @TestTemplate
   public void testPendingAcksEventuallyArriveOnPause() throws Exception {
      testPendingAcksEventuallyArrive(false, false);
   }

   @TestTemplate
   public void testPendingAcksEventuallyArriveOnStopWithLargeMessages() throws Exception {
      testPendingAcksEventuallyArrive(true, true);
   }

   @TestTemplate
   public void testPendingAcksEventuallyArriveOnPauseWithLargeMessages() throws Exception {
      testPendingAcksEventuallyArrive(false, true);
   }

   private void testPendingAcksEventuallyArrive(boolean stop, boolean large) throws Exception {
      server0 = createClusteredServerWithParams(isNetty(), 0, true, null);

      Map<String, Object> server1Params = new HashMap<>();
      addTargetParameters(server1Params);
      server1 = createClusteredServerWithParams(isNetty(), 1, true, server1Params);

      final String testAddress = "testAddress";
      final String queueName0 = "queue0";
      final String forwardAddress = "forwardAddress";
      final String queueName1 = "queue1";
      final long pendingAckTimeout = 2000;
      final int messageSize = 1024;
      final int numMessages = 10;

      TransportConfiguration server0tc = new TransportConfiguration(getConnector(), null);
      TransportConfiguration server1tc = new TransportConfiguration(getConnector(), server1Params);

      server0.getConfiguration()
             .setConnectorConfigurations(Map.of(server1tc.getName(), server1tc))
             .setBridgeConfigurations(Arrays.asList(new BridgeConfiguration()
                                                       .setName("bridge1")
                                                       .setQueueName(queueName0)
                                                       .setForwardingAddress(forwardAddress)
                                                       .setRetryInterval(1000)
                                                       .setReconnectAttemptsOnSameNode(-1)
                                                       .setUseDuplicateDetection(false)
                                                       .setConfirmationWindowSize(numMessages * messageSize / 2)
                                                       .setMinLargeMessageSize(large ? (messageSize / 2) : (messageSize * 2))
                                                       .setPendingAckTimeout(pendingAckTimeout)
                                                       .setStaticConnectors(Arrays.asList(server1tc.getName()))));
      server0.getConfiguration().setQueueConfigs(Arrays.asList(QueueConfiguration.of(queueName0).setAddress(testAddress)));
      server0.start();

      // this interceptor will prevent the target from returning any send acks until a certain amount of time has elapsed
      final CountDownLatch opLatch = new CountDownLatch(1);
      Interceptor sendBlockingInterceptor = (packet, connection) -> {
         if (packet.getType() == PacketImpl.SESS_SEND || packet.getType() == PacketImpl.SESS_SEND_LARGE) {
            try {
               opLatch.await(pendingAckTimeout, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
               throw new RuntimeException(e);
            }
         }
         return true;
      };

      server1.getConfiguration().setQueueConfigs(Arrays.asList(QueueConfiguration.of(queueName1).setAddress(forwardAddress)));
      server1.start();
      server1.getRemotingService().addIncomingInterceptor(sendBlockingInterceptor);
      Bridge bridge = server0.getClusterManager().getBridges().get("bridge1");
      Wait.assertTrue(() -> (bridge.isConnected()), 2000, 100);

      locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(server0tc, server1tc));
      ClientSessionFactory sf0 = addSessionFactory(locator.createSessionFactory(server0tc));
      ClientSession session0 = sf0.createSession(false, true, true);
      ClientProducer producer0 = session0.createProducer(SimpleString.of(testAddress));
      final byte[] bytes = new byte[messageSize];

      final SimpleString propKey = SimpleString.of("testkey");

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session0.createMessage(true);
         message.putIntProperty(propKey, i);
         message.getBodyBuffer().writeBytes(bytes);
         producer0.send(message);
      }

      session0.close();
      sf0.close();

      Wait.assertEquals((long) numMessages, () -> bridge.getMetrics().getMessagesPendingAcknowledgement(), 2000, 100);
      assertEquals((long) numMessages, server0.locateQueue(queueName0).getDeliveringCount());
      BridgeImpl.State desiredState;
      if (stop) {
         bridge.stop();
         desiredState = BridgeImpl.State.STOPPED;
      } else {
         bridge.pause();
         desiredState = BridgeImpl.State.PAUSED;
      }
      Thread.sleep(pendingAckTimeout / 2);
      opLatch.countDown();
      Wait.assertEquals(desiredState, () -> ((BridgeImpl)bridge).getState(), pendingAckTimeout, 25);
      Wait.assertEquals(0L, () -> server0.locateQueue(queueName0).getMessageCount(), 2000, 100);
      Wait.assertEquals(0L, () -> server0.locateQueue(queueName0).getDeliveringCount(), 2000, 100);
      Wait.assertEquals((long) numMessages, () -> server1.locateQueue(queueName1).getMessageCount(), 2000, 100);
   }

   /**
    * It will inspect the journal directly and determine if there are queues on this journal,
    *
    * @return a Map containing the reference counts per queue
    */
   protected Map<Long, AtomicInteger> loadQueues(ActiveMQServer serverToInvestigate) throws Exception {
      SequentialFileFactory messagesFF = new NIOSequentialFileFactory(serverToInvestigate.getConfiguration().getJournalLocation(), 1);

      JournalImpl messagesJournal = new JournalImpl(serverToInvestigate.getConfiguration().getJournalFileSize(), serverToInvestigate.getConfiguration().getJournalMinFiles(), serverToInvestigate.getConfiguration().getJournalPoolFiles(), 0, 0, messagesFF, "activemq-data", "amq", 1);
      List<RecordInfo> records = new LinkedList<>();

      List<PreparedTransactionInfo> preparedTransactions = new LinkedList<>();

      messagesJournal.start();
      messagesJournal.load(records, preparedTransactions, null);

      // These are more immutable integers
      Map<Long, AtomicInteger> messageRefCounts = new HashMap<>();

      for (RecordInfo info : records) {
         Object o = DescribeJournal.newObjectEncoding(info);
         if (info.getUserRecordType() == JournalRecordIds.ADD_REF) {
            DescribeJournal.ReferenceDescribe ref = (DescribeJournal.ReferenceDescribe) o;
            AtomicInteger count = messageRefCounts.get(ref.refEncoding.queueID);
            if (count == null) {
               count = new AtomicInteger(1);
               messageRefCounts.put(ref.refEncoding.queueID, count);
            } else {
               count.incrementAndGet();
            }
         }
      }

      messagesJournal.stop();

      return messageRefCounts;

   }
}
