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

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;
import java.lang.invoke.MethodHandles;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.ActiveMQAddressDoesNotExistException;
import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException;
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.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.core.client.impl.ClientLargeMessageImpl;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.config.TransformerConfiguration;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.impl.DivertBinding;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServers;
import org.apache.activemq.artemis.core.server.ComponentConfigurationRoutingType;
import org.apache.activemq.artemis.core.server.Divert;
import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.QueueManagerImpl;
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.AddressSettings;
import org.apache.activemq.artemis.core.settings.impl.DeletionPolicy;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.CFUtil;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.command.ActiveMQTopic;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

public class DivertTest extends ActiveMQTestBase {

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

   private static final int TIMEOUT = 3000;

   @Test
   public void testDivertedNotificationMessagePropertiesOpenWire() throws Exception {
      final String testAddress = ActiveMQDefaultConfiguration.getDefaultManagementNotificationAddress().toString();

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress).setFilterString("_AMQ_NotifType = 'CONSUMER_CREATED' OR _AMQ_NotifType = 'CONSUMER_CLOSED'");

      Configuration config = createDefaultNettyConfig().addDivertConfiguration(divertConf);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");

      connectionFactory.setClientID("myClientID");

      Topic forwardTopic = new ActiveMQTopic(forwardAddress);
      Connection connection = connectionFactory.createConnection();

      connection.start();

      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      TopicSubscriber subscriber = session.createDurableSubscriber(forwardTopic, "mySubscriptionName");

      javax.jms.Message message = subscriber.receive(DivertTest.TIMEOUT);

      connection.close();

      assertNotNull(message);

      assertEquals("CONSUMER_CREATED", message.getStringProperty("_AMQ_NotifType"));
   }

   @Test
   public void testSingleNonExclusiveDivert() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();

      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      final int numMessages = 1;

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

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

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

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

         assertEquals("forwardAddress", message.getAddress());

         assertEquals("testAddress", message.getStringProperty(Message.HDR_ORIGINAL_ADDRESS));

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer2.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         assertEquals("testAddress", message.getAddress());

         message.acknowledge();
      }

      assertNull(consumer2.receiveImmediate());
   }

   @Test
   public void testDivertAndQueueWithSameName() throws Exception {
      final String name = RandomUtil.randomUUIDString();

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig()
                                                                             .addDivertConfiguration(new DivertConfiguration()
                                                                                                        .setName(name)
                                                                                                        .setRoutingName(RandomUtil.randomUUIDString())
                                                                                                        .setAddress(RandomUtil.randomUUIDString())
                                                                                                        .setForwardingAddress(RandomUtil.randomUUIDString())), false));

      server.start();

      try {
         server.createQueue(QueueConfiguration.of(name));
         fail();
      } catch (ActiveMQIllegalStateException e) {
         // expected
      }
   }

   @Test
   public void testCrossProtocol() throws Exception {
      final String testForConvert = "testConvert";

      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress).
         setRoutingType(ComponentConfigurationRoutingType.ANYCAST);

      Configuration config = createDefaultNettyConfig().addDivertConfiguration(divertConf);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      final SimpleString queueName1 = SimpleString.of(testAddress);

      final SimpleString queueName2 = SimpleString.of(forwardAddress);

      { // this is setting up the queues
         ServerLocator locator = createInVMNonHALocator();

         ClientSessionFactory sf = createSessionFactory(locator);

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

         session.createQueue(QueueConfiguration.of(queueName1).setAddress(testAddress).setRoutingType(RoutingType.ANYCAST));

         session.createQueue(QueueConfiguration.of(SimpleString.of(testForConvert)).setAddress(testForConvert).setRoutingType(RoutingType.ANYCAST));

         session.createQueue(QueueConfiguration.of(queueName2).setAddress(forwardAddress).setRoutingType(RoutingType.ANYCAST));
      }

      ConnectionFactory coreCF = CFUtil.createConnectionFactory("CORE", "tcp://localhost:61616");
      Connection coreConnection = coreCF.createConnection();
      Session coreSession = coreConnection.createSession(Session.AUTO_ACKNOWLEDGE);
      MessageProducer producerCore = coreSession.createProducer(coreSession.createQueue(testForConvert));

      for (int i = 0; i < 10; i++) {
         TextMessage textMessage = coreSession.createTextMessage("text" + i);
         //if (i % 2 == 0) textMessage.setIntProperty("key", i);
         producerCore.send(textMessage);
      }

      producerCore.close();

      ConnectionFactory amqpCF = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:61616");

      Connection amqpConnection = amqpCF.createConnection();
      Session amqpSession = amqpConnection.createSession(Session.AUTO_ACKNOWLEDGE);
      Queue amqpQueue = amqpSession.createQueue(testAddress);
      MessageProducer producer = amqpSession.createProducer(amqpQueue);
      MessageConsumer consumerFromConvert = amqpSession.createConsumer(amqpSession.createQueue(testForConvert));
      amqpConnection.start();

      for (int i = 0; i < 10; i++) {
         javax.jms.Message received =  consumerFromConvert.receive(5000);
         assertNotNull(received);
         producer.send(received);
      }


      Queue outQueue = coreSession.createQueue(queueName2.toString());
      MessageConsumer consumer = coreSession.createConsumer(outQueue);
      coreConnection.start();

      for (int i = 0; i < 10; i++) {
         TextMessage textMessage = (TextMessage)consumer.receive(5000);
         assertNotNull(textMessage);
         assertEquals("text" + i, textMessage.getText());
         //if (i % 2 == 0) Assert.assertEquals(i, textMessage.getIntProperty("key"));
      }

      assertNull(consumer.receiveNoWait());

   }

   @Test
   public void testSingleNonExclusiveDivertWithRoutingType() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();

      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress).setRoutingType(RoutingType.ANYCAST).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      final int numMessages = 1;

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

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

         message.setRoutingType(RoutingType.MULTICAST);

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer2.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer2.receiveImmediate());
   }

   @Test
   public void testSingleExclusiveDivertWithRoutingType() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress).setExclusive(true);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();

      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress).setRoutingType(RoutingType.ANYCAST).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      final int numMessages = 1;

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

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

         message.setRoutingType(RoutingType.MULTICAST);

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());
   }

   @Test
   public void testSingleDivertWithExpiry() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      final String expiryAddress = "expiryAddress";

      AddressSettings expirySettings = new AddressSettings().setExpiryAddress(SimpleString.of(expiryAddress));

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf).clearAddressSettings().addAddressSetting("#", expirySettings);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, true));

      server.start();

      ServerLocator locator = createInVMNonHALocator();

      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(testAddress));

      session.createQueue(QueueConfiguration.of(expiryAddress));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      final int numMessages = 1;

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

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

         message.putIntProperty(propKey, i);

         message.setExpiration(System.currentTimeMillis() + 1000);

         producer.send(message);
      }
      session.commit();

      // this context is validating if these messages are routed correctly
      {
         int count1 = 0;
         ClientMessage message = null;
         while ((message = consumer1.receiveImmediate()) != null) {
            message.acknowledge();
            count1++;
         }

         int count2 = 0;
         while ((message = consumer2.receiveImmediate()) != null) {
            message.acknowledge();
            count2++;
         }

         assertEquals(1, count1);
         assertEquals(1, count2);
         session.rollback();
      }
      Thread.sleep(2000);

      // it must been expired by now
      assertNull(consumer1.receiveImmediate());
      // it must been expired by now
      assertNull(consumer2.receiveImmediate());

      int countOriginal1 = 0;
      int countOriginal2 = 0;
      ClientConsumer consumerExpiry = session.createConsumer(expiryAddress);

      for (int i = 0; i < numMessages * 2; i++) {
         ClientMessage message = consumerExpiry.receive(5000);
         logger.debug("Received message {}", message);
         assertNotNull(message);

         if (message.getStringProperty(Message.HDR_ORIGINAL_QUEUE).equals("queue1")) {
            countOriginal1++;
         } else if (message.getStringProperty(Message.HDR_ORIGINAL_QUEUE).equals("queue2")) {
            countOriginal2++;
         } else {
            logger.debug("message not part of any expired queue {}", message);
         }
      }

      assertEquals(numMessages, countOriginal1);
      assertEquals(numMessages, countOriginal2);
   }

   @Test
   public void testSingleNonExclusiveDivert2() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();

      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      final SimpleString queueName3 = SimpleString.of("queue3");

      final SimpleString queueName4 = SimpleString.of("queue4");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(testAddress).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName3).setAddress(testAddress).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName4).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      ClientConsumer consumer3 = session.createConsumer(queueName3);

      ClientConsumer consumer4 = session.createConsumer(queueName4);

      final int numMessages = 10;

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

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

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer2.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer2.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer3.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer3.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer4.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer4.receiveImmediate());
   }

   @Test
   public void testSingleNonExclusiveDivert3() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();
      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      final int numMessages = 10;

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

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

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());
   }

   @Test
   public void testSingleExclusiveDivert() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress).setExclusive(true);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();
      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      final SimpleString queueName3 = SimpleString.of("queue3");

      final SimpleString queueName4 = SimpleString.of("queue4");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(testAddress).setDurable(false));
      session.createQueue(QueueConfiguration.of(queueName3).setAddress(testAddress).setDurable(false));
      session.createQueue(QueueConfiguration.of(queueName4).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      ClientConsumer consumer3 = session.createConsumer(queueName3);

      ClientConsumer consumer4 = session.createConsumer(queueName4);

      final int numMessages = 10;

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

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

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

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

         assertEquals("forwardAddress", message.getAddress());

         assertEquals("testAddress", message.getStringProperty(Message.HDR_ORIGINAL_ADDRESS));

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      assertNull(consumer2.receiveImmediate());

      assertNull(consumer3.receiveImmediate());

      assertNull(consumer4.receiveImmediate());
   }

   @Test
   public void testCompositeDivert() throws Exception {
      final String testAddress = "testAddress";
      final String forwardAddress1 = "forwardAddress1";
      final String forwardAddress2 = "forwardAddress2";
      final String forwardAddress3 = "forwardAddress3";
      final String forwardAddresses = forwardAddress1 + ", " + forwardAddress2 + ", " + forwardAddress3;

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(new DivertConfiguration()
                                                                                 .setName("divert1")
                                                                                 .setRoutingName("divert1")
                                                                                 .setAddress(testAddress)
                                                                                 .setForwardingAddress(forwardAddresses)
                                                                                 .setExclusive(true));

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();
      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");
      final SimpleString queueName2 = SimpleString.of("queue2");
      final SimpleString queueName3 = SimpleString.of("queue3");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress1).setDurable(false));
      session.createQueue(QueueConfiguration.of(queueName2).setAddress(forwardAddress2).setDurable(false));
      session.createQueue(QueueConfiguration.of(queueName3).setAddress(forwardAddress3).setDurable(false));

      session.start();

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

      final int numMessages = 10;

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

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session.createMessage(false);
         message.putIntProperty(propKey, i);
         producer.send(message);
      }

      ClientConsumer consumer1 = session.createConsumer(queueName1);
      ClientConsumer consumer2 = session.createConsumer(queueName2);
      ClientConsumer consumer3 = session.createConsumer(queueName3);

      ClientConsumer[] consumers = new ClientConsumer[] {consumer1, consumer2, consumer3};

      for (int i = 0; i < numMessages; i++) {
         for (int j = 0; j < consumers.length; j++) {
            ClientMessage message = consumers[j].receive(DivertTest.TIMEOUT);
            assertNotNull(message);
            assertEquals(i, message.getObjectProperty(propKey));
            assertEquals("forwardAddress" + (j + 1), message.getAddress());
            assertEquals("testAddress", message.getStringProperty(Message.HDR_ORIGINAL_ADDRESS));
            message.acknowledge();
         }
      }

      assertNull(consumer1.receiveImmediate());
      assertNull(consumer2.receiveImmediate());
      assertNull(consumer3.receiveImmediate());
   }

   @Test
   public void testSinglePersistedDivert() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress).setExclusive(true);


      QueueConfiguration q1 = QueueConfiguration.of("forwardAddress1").setDurable(true).setRoutingType(RoutingType.ANYCAST);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf).addQueueConfiguration(q1);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, true));

      server.start();

      server.stop();

      divertConf.setRoutingName("divert2");

      server.start();

      Binding divert1 = server.getPostOffice().getBinding(SimpleString.of("divert1"));

      assertNotNull(divert1);

      assertEquals(divert1.getRoutingName(), SimpleString.of("divert2"));
   }

   @Test
   public void testSinglePersistedNewDivert() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress).setExclusive(true);


      QueueConfiguration q1 = QueueConfiguration.of("forwardAddress1").setDurable(true).setRoutingType(RoutingType.ANYCAST);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf).addQueueConfiguration(q1);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, true));

      server.start();

      server.stop();

      divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert2").setAddress(testAddress).setForwardingAddress(forwardAddress).setExclusive(true);

      config.getDivertConfigurations().clear();

      config.getDivertConfigurations().add(divertConf);

      server.start();

      Binding divert1 = server.getPostOffice().getBinding(SimpleString.of("divert1"));

      assertNotNull(divert1);

      assertEquals(divert1.getRoutingName(), SimpleString.of("divert2"));
   }

   @Test
   public void testSinglePersistedNoDeleteDivert() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress).setExclusive(true);


      QueueConfiguration q1 = QueueConfiguration.of("forwardAddress1").setDurable(true).setRoutingType(RoutingType.ANYCAST);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf).addQueueConfiguration(q1);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, true));

      server.start();

      server.stop();

      config.getDivertConfigurations().clear();

      server.start();

      Binding divert1 = server.getPostOffice().getBinding(SimpleString.of("divert1"));

      assertNotNull(divert1);

      assertEquals(divert1.getRoutingName(), SimpleString.of("divert1"));
   }

   @Test
   public void testSinglePersistedDeleteDivert() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress).setExclusive(true);


      QueueConfiguration q1 = QueueConfiguration.of("forwardAddress1").setDurable(true).setRoutingType(RoutingType.ANYCAST);

      AddressSettings addressSettings = new AddressSettings();

      addressSettings.setConfigDeleteDiverts(DeletionPolicy.FORCE);



      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf).addQueueConfiguration(q1);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, true));

      server.getAddressSettingsRepository().addMatch(testAddress, addressSettings);

      server.start();

      server.stop();

      config.getDivertConfigurations().clear();

      server.start();

      Binding divert1 = server.getPostOffice().getBinding(SimpleString.of("divert1"));

      assertNull(divert1);
   }

   @Test
   public void testMixedPersistedDeleteDivert() throws Exception {
      final String testAddress = "testAddress";

      final String testAddress2 = "testAddress2";

      final String forwardAddress = "forwardAddress";

      final String forwardAddress2 = "forwardAddress2";

      DivertConfiguration divertConf = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress).setExclusive(true);

      DivertConfiguration divertConf2 = new DivertConfiguration().setName("divert2").setRoutingName("divert2").setAddress(testAddress2).setForwardingAddress(forwardAddress2).setExclusive(true);


      QueueConfiguration q1 = QueueConfiguration.of("forwardAddress1").setDurable(true).setRoutingType(RoutingType.ANYCAST);

      AddressSettings addressSettings = new AddressSettings();

      addressSettings.setConfigDeleteDiverts(DeletionPolicy.FORCE);

      AddressSettings addressSettings2 = new AddressSettings();

      addressSettings2.setConfigDeleteDiverts(DeletionPolicy.OFF);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf).addDivertConfiguration(divertConf2).addQueueConfiguration(q1);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, true));

      server.getAddressSettingsRepository().addMatch(testAddress, addressSettings);

      server.getAddressSettingsRepository().addMatch(testAddress2, addressSettings2);

      server.start();

      server.stop();

      config.getDivertConfigurations().clear();

      server.start();

      Binding divert1 = server.getPostOffice().getBinding(SimpleString.of("divert1"));

      assertNull(divert1);

      Binding divert2 = server.getPostOffice().getBinding(SimpleString.of("divert2"));

      assertNotNull(divert2);
   }


   @Test
   public void testMultipleNonExclusiveDivert() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress1 = "forwardAddress1";
      final String forwardAddress2 = "forwardAddress2";
      final String forwardAddress3 = "forwardAddress3";

      DivertConfiguration divertConf1 = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress1);

      DivertConfiguration divertConf2 = new DivertConfiguration().setName("divert2").setRoutingName("divert2").setAddress(testAddress).setForwardingAddress(forwardAddress2);

      DivertConfiguration divertConf3 = new DivertConfiguration().setName("divert3").setRoutingName("divert3").setAddress(testAddress).setForwardingAddress(forwardAddress3);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf1).addDivertConfiguration(divertConf2).addDivertConfiguration(divertConf3);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();
      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      final SimpleString queueName3 = SimpleString.of("queue3");

      final SimpleString queueName4 = SimpleString.of("queue4");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress1).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(forwardAddress2).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName3).setAddress(forwardAddress3).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName4).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      ClientConsumer consumer3 = session.createConsumer(queueName3);

      ClientConsumer consumer4 = session.createConsumer(queueName4);

      final int numMessages = 10;

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

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

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer2.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer2.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer3.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer3.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer4.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer4.receiveImmediate());
   }

   @Test
   public void testMultipleExclusiveDivert() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress1 = "forwardAddress1";
      final String forwardAddress2 = "forwardAddress2";
      final String forwardAddress3 = "forwardAddress3";

      DivertConfiguration divertConf1 = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress1).setExclusive(true);

      DivertConfiguration divertConf2 = new DivertConfiguration().setName("divert2").setRoutingName("divert2").setAddress(testAddress).setForwardingAddress(forwardAddress2).setExclusive(true);

      DivertConfiguration divertConf3 = new DivertConfiguration().setName("divert3").setRoutingName("divert3").setAddress(testAddress).setForwardingAddress(forwardAddress3).setExclusive(true);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf1).addDivertConfiguration(divertConf2).addDivertConfiguration(divertConf3);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();

      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      final SimpleString queueName3 = SimpleString.of("queue3");

      final SimpleString queueName4 = SimpleString.of("queue4");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress1).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(forwardAddress2).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName3).setAddress(forwardAddress3).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName4).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      ClientConsumer consumer3 = session.createConsumer(queueName3);

      ClientConsumer consumer4 = session.createConsumer(queueName4);

      final int numMessages = 10;

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

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

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer2.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer2.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer3.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer3.receiveImmediate());

      assertNull(consumer4.receiveImmediate());
   }

   @Test
   public void testMixExclusiveAndNonExclusiveDiverts() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress1 = "forwardAddress1";
      final String forwardAddress2 = "forwardAddress2";
      final String forwardAddress3 = "forwardAddress3";

      DivertConfiguration divertConf1 = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress1).setExclusive(true);

      DivertConfiguration divertConf2 = new DivertConfiguration().setName("divert2").setRoutingName("divert2").setAddress(testAddress).setForwardingAddress(forwardAddress2).setExclusive(true);

      DivertConfiguration divertConf3 = new DivertConfiguration().setName("divert3").setRoutingName("divert3").setAddress(testAddress).setForwardingAddress(forwardAddress3);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf1).addDivertConfiguration(divertConf2).addDivertConfiguration(divertConf3);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();
      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      final SimpleString queueName3 = SimpleString.of("queue3");

      final SimpleString queueName4 = SimpleString.of("queue4");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress1).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(forwardAddress2).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName3).setAddress(forwardAddress3).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName4).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      ClientConsumer consumer3 = session.createConsumer(queueName3);

      ClientConsumer consumer4 = session.createConsumer(queueName4);

      final int numMessages = 10;

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

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

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer2.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer2.receiveImmediate());

      assertNull(consumer3.receiveImmediate());

      assertNull(consumer4.receiveImmediate());
   }

   // If no exclusive diverts match then non exclusive ones should be called
   @Test
   public void testSingleExclusiveNonMatchingAndNonExclusiveDiverts() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress1 = "forwardAddress1";
      final String forwardAddress2 = "forwardAddress2";
      final String forwardAddress3 = "forwardAddress3";

      final String filter = "animal='antelope'";

      DivertConfiguration divertConf1 = new DivertConfiguration().setName("divert1").setRoutingName("divert1").setAddress(testAddress).setForwardingAddress(forwardAddress1).setExclusive(true).setFilterString(filter);

      DivertConfiguration divertConf2 = new DivertConfiguration().setName("divert2").setRoutingName("divert2").setAddress(testAddress).setForwardingAddress(forwardAddress2);

      DivertConfiguration divertConf3 = new DivertConfiguration().setName("divert3").setRoutingName("divert3").setAddress(testAddress).setForwardingAddress(forwardAddress3);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf1).addDivertConfiguration(divertConf2).addDivertConfiguration(divertConf3);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();

      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      final SimpleString queueName3 = SimpleString.of("queue3");

      final SimpleString queueName4 = SimpleString.of("queue4");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress1).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(forwardAddress2).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName3).setAddress(forwardAddress3).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName4).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      ClientConsumer consumer3 = session.createConsumer(queueName3);

      ClientConsumer consumer4 = session.createConsumer(queueName4);

      final int numMessages = 10;

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

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

         message.putStringProperty(SimpleString.of("animal"), SimpleString.of("giraffe"));

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

      // for (int i = 0; i < numMessages; i++)
      // {
      // ClientMessage message = consumer1.receive(200);
      //
      // assertNotNull(message);
      //
      // assertEquals((Integer)i, (Integer)message.getProperty(propKey));
      //
      // message.acknowledge();
      // }

      assertNull(consumer1.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer2.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer2.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer3.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer3.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer4.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer4.receiveImmediate());

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

         message.putStringProperty(SimpleString.of("animal"), SimpleString.of("antelope"));

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      assertNull(consumer2.receiveImmediate());

      assertNull(consumer3.receiveImmediate());

      assertNull(consumer4.receiveImmediate());
   }

   @Test
   public void testRoundRobinDiverts() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress1 = "forwardAddress1";
      final String forwardAddress2 = "forwardAddress2";
      final String forwardAddress3 = "forwardAddress3";

      DivertConfiguration divertConf1 = new DivertConfiguration().setName("divert1").setRoutingName("thename").setAddress(testAddress).setForwardingAddress(forwardAddress1);

      DivertConfiguration divertConf2 = new DivertConfiguration().setName("divert2").setRoutingName("thename").setAddress(testAddress).setForwardingAddress(forwardAddress2);

      DivertConfiguration divertConf3 = new DivertConfiguration().setName("divert3").setRoutingName("thename").setAddress(testAddress).setForwardingAddress(forwardAddress3);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf1).addDivertConfiguration(divertConf2).addDivertConfiguration(divertConf3);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      ServerLocator locator = createInVMNonHALocator();

      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      final SimpleString queueName3 = SimpleString.of("queue3");

      final SimpleString queueName4 = SimpleString.of("queue4");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress1).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(forwardAddress2).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName3).setAddress(forwardAddress3).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName4).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      ClientConsumer consumer3 = session.createConsumer(queueName3);

      ClientConsumer consumer4 = session.createConsumer(queueName4);

      final int numMessages = 10;

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

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

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

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

         message.acknowledge();

         i++;

         if (i == numMessages) {
            break;
         }

         message = consumer2.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         message.acknowledge();

         i++;

         if (i == numMessages) {
            break;
         }

         message = consumer3.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         message.acknowledge();

         i++;
      }

      assertNull(consumer1.receiveImmediate());
      assertNull(consumer2.receiveImmediate());
      assertNull(consumer3.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer4.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

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

         message.acknowledge();
      }

      assertNull(consumer4.receiveImmediate());
   }

   @Test
   public void testDeployDivertsSameUniqueName() throws Exception {
      final String testAddress = "testAddress";

      final String forwardAddress1 = "forwardAddress1";
      final String forwardAddress2 = "forwardAddress2";
      final String forwardAddress3 = "forwardAddress3";

      DivertConfiguration divertConf1 = new DivertConfiguration().setName("divert1").setRoutingName("thename1").setAddress(testAddress).setForwardingAddress(forwardAddress1);

      DivertConfiguration divertConf2 = new DivertConfiguration().setName("divert1").setRoutingName("thename2").setAddress(testAddress).setForwardingAddress(forwardAddress2);

      DivertConfiguration divertConf3 = new DivertConfiguration().setName("divert2").setRoutingName("thename3").setAddress(testAddress).setForwardingAddress(forwardAddress3);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf1).addDivertConfiguration(divertConf2).addDivertConfiguration(divertConf3);

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));

      server.start();

      // Only the first and third should be deployed

      ServerLocator locator = createInVMNonHALocator();

      ClientSessionFactory sf = createSessionFactory(locator);

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

      final SimpleString queueName1 = SimpleString.of("queue1");

      final SimpleString queueName2 = SimpleString.of("queue2");

      final SimpleString queueName3 = SimpleString.of("queue3");

      final SimpleString queueName4 = SimpleString.of("queue4");

      session.createQueue(QueueConfiguration.of(queueName1).setAddress(forwardAddress1).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName2).setAddress(forwardAddress2).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName3).setAddress(forwardAddress3).setDurable(false));

      session.createQueue(QueueConfiguration.of(queueName4).setAddress(testAddress).setDurable(false));

      session.start();

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

      ClientConsumer consumer1 = session.createConsumer(queueName1);

      ClientConsumer consumer2 = session.createConsumer(queueName2);

      ClientConsumer consumer3 = session.createConsumer(queueName3);

      ClientConsumer consumer4 = session.createConsumer(queueName4);

      final int numMessages = 10;

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

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

         message.putIntProperty(propKey, i);

         producer.send(message);
      }

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

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer1.receiveImmediate());

      assertNull(consumer2.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer3.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer3.receiveImmediate());

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer4.receive(DivertTest.TIMEOUT);

         assertNotNull(message);

         assertEquals(i, message.getIntProperty(propKey).intValue());

         message.acknowledge();
      }

      assertNull(consumer4.receiveImmediate());
   }

   @Test
   public void testInjectedTransformer() throws Exception {
      final SimpleString ADDRESS = SimpleString.of("myAddress");
      final String DIVERT = "myDivert";

      ServiceRegistryImpl serviceRegistry = new ServiceRegistryImpl();
      Transformer transformer = message -> null;
      serviceRegistry.addDivertTransformer(DIVERT, transformer);

      ActiveMQServer server = addServer(new ActiveMQServerImpl(createBasicConfig(), null, null, null, serviceRegistry));
      server.start();
      server.waitForActivation(100, TimeUnit.MILLISECONDS);
      server.createQueue(QueueConfiguration.of("myQueue").setAddress(ADDRESS).setDurable(false));
      server.deployDivert(new DivertConfiguration().setName(DIVERT).setAddress(ADDRESS.toString()).setForwardingAddress(ADDRESS.toString()));
      Collection<Binding> bindings = server.getPostOffice().getBindingsForAddress(ADDRESS).getBindings();
      Divert divert = null;
      for (Binding binding : bindings) {
         if (binding instanceof DivertBinding divertBinding) {
            divert = divertBinding.getDivert();
         }
      }
      assertNotNull(divert);
      assertEquals(transformer, divert.getTransformer());

      server.destroyDivert(SimpleString.of(DIVERT));
      assertNull(serviceRegistry.getDivertTransformer(DIVERT, null));
   }

   @Test
   public void testProperties() throws Exception {
      final String testAddress = "testAddress";
      final SimpleString queue = SimpleString.of("queue");
      final int COUNT = 25;

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig(), false));
      server.start();

      server.createQueue(QueueConfiguration.of(queue).setAddress(testAddress + (COUNT)).setRoutingType(RoutingType.ANYCAST));
      for (int i = 0; i < COUNT; i++) {
         server.deployDivert(new DivertConfiguration()
                                .setName("divert" + i)
                                .setAddress(testAddress + i)
                                .setForwardingAddress(testAddress + (i + 1)));
      }

      ServerLocator locator = createInVMNonHALocator();
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession session = sf.createSession(false, true, true);
      session.start();

      ClientProducer producer = session.createProducer(SimpleString.of(testAddress + "0"));
      ClientConsumer consumer1 = session.createConsumer(queue);
      ClientMessage message = session.createMessage(false);
      producer.send(message);

      message = consumer1.receive(DivertTest.TIMEOUT);
      assertNotNull(message);
      message.acknowledge();
      assertEquals("testAddress" + COUNT, message.getAddress());
      assertEquals("testAddress" + (COUNT - 1), message.getStringProperty(Message.HDR_ORIGINAL_ADDRESS));
   }

   @Test
   public void testTransformedToLarge() throws Exception {
      final String address = "address";
      final SimpleString queue = SimpleString.of("queue");
      final String forwardingAddress = "forwardingAddress";
      final SimpleString forwardingQueue = SimpleString.of("forwardingQueue");
      final int journalBufferSize = 1024;
      final int headerCount = 5;

      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig().setJournalBufferSize_NIO(journalBufferSize).setJournalType(JournalType.NIO), true));
      server.start();

      // configure the transformer to add headers to increase the size of the message past the journal buffer size
      Map<String, String> headers = new HashMap<>();
      for (int i = 0; i < headerCount; i++) {
         headers.put(RandomUtil.randomAlphaNumericString(32), RandomUtil.randomAlphaNumericString(32));
      }
      TransformerConfiguration transformerConfiguration = new TransformerConfiguration();
      transformerConfiguration.setClassName(AddHeadersTransformer.class.getName());
      transformerConfiguration.setProperties(headers);

      server.createQueue(QueueConfiguration.of(queue).setAddress(address).setRoutingType(RoutingType.ANYCAST));
      server.createQueue(QueueConfiguration.of(forwardingQueue).setAddress(forwardingAddress).setRoutingType(RoutingType.ANYCAST));
      server.deployDivert(new DivertConfiguration()
                             .setName("divert")
                             .setAddress(address)
                             .setForwardingAddress(forwardingAddress)
                             .setTransformerConfiguration(transformerConfiguration));

      ServerLocator locator = createInVMNonHALocator();
      locator.setMinLargeMessageSize(Integer.MAX_VALUE);
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession session = sf.createSession(false, true, true);
      session.start();

      ClientProducer producer = session.createProducer(SimpleString.of(address));
      ClientConsumer consumer = session.createConsumer(queue);
      ClientConsumer divertedConsumer = session.createConsumer(forwardingQueue);
      ClientMessage message = session.createMessage(true);
      message.getBodyBuffer().writeBytes(RandomUtil.randomBytes(journalBufferSize / 4));
      producer.send(message);

      // ensure the non-diverted message is not turned into a large message
      message = consumer.receive(DivertTest.TIMEOUT);
      assertNotNull(message);
      message.acknowledge();
      assertFalse(message instanceof ClientLargeMessageImpl);

      // ensure the diverted message is turned into a large message
      message = divertedConsumer.receive(DivertTest.TIMEOUT);
      assertNotNull(message);
      message.acknowledge();
      assertTrue(message instanceof ClientLargeMessageImpl);
   }

   @Test
   public void testDivertToNewAddress() throws Exception {
      final String queueName = "queue";
      final String dummyQueueName = "dummy";
      final String noDivertAutoCreateQName = "notAllowed";
      final String propKey = "newQueue";
      final String DIVERT = "myDivert";
      final int numMessages = 10;

      Transformer transformer = message -> message.setAddress(message.getStringProperty(propKey));

      ServiceRegistryImpl serviceRegistry = new ServiceRegistryImpl();
      serviceRegistry.addDivertTransformer(DIVERT, transformer);

      AddressSettings autoCreateDestinationsAS = new AddressSettings().setAutoCreateAddresses(true).setAutoCreateQueues(true);
      AddressSettings noAutoCreateDestinationsAS = new AddressSettings().setAutoCreateAddresses(false).setAutoCreateQueues(false);

      ActiveMQServer server = addServer(new ActiveMQServerImpl(createDefaultInVMConfig(), null, null, null, serviceRegistry));

      server.getConfiguration().addAddressSetting("#", autoCreateDestinationsAS);
      server.getConfiguration().addAddressSetting(noDivertAutoCreateQName, noAutoCreateDestinationsAS);

      server.start();

      server.createQueue(QueueConfiguration.of(queueName));
      server.deployDivert(new DivertConfiguration()
                             .setName(DIVERT)
                             .setAddress(queueName)
                             .setRoutingType(ComponentConfigurationRoutingType.ANYCAST)
                             .setForwardingAddress(dummyQueueName)
                             .setExclusive(true));

      ServerLocator locator = createInVMNonHALocator();
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession session = sf.createSession(false, true, true);
      session.start();

      ClientMessage message;
      ClientProducer producer = session.createProducer(queueName);

      for (int i = 0; i < numMessages; i++) {
         message = session.createMessage(true);
         message.putStringProperty(propKey, queueName + "." + i);
         producer.send(message);
      }

      for (int i = 0; i < numMessages; i++) {
         ClientConsumer consumer = session.createConsumer(queueName + "." + i);
         message = consumer.receive(DivertTest.TIMEOUT);
         assertNotNull(message);
         message.acknowledge();
         consumer.close();
      }

      ClientMessage failMessage = session.createMessage(true);

      assertThrows(ActiveMQAddressDoesNotExistException.class, () -> {
         failMessage.putStringProperty(propKey, noDivertAutoCreateQName);
         producer.send(failMessage);
      });

      producer.close();

      assertNull(server.locateQueue(noDivertAutoCreateQName));
      assertNull(server.locateQueue(dummyQueueName));

   }

   @Test
   public void testHandleAutoDeleteDestination() throws Exception {
      final String testAddress = "testAddress";
      final String forwardAddress = "forwardAddress";

      DivertConfiguration divertConf = new DivertConfiguration()
         .setName("divert")
         .setRoutingType(ComponentConfigurationRoutingType.ANYCAST)
         .setExclusive(true)
         .setAddress(testAddress)
         .setForwardingAddress(forwardAddress);

      AddressSettings addressSettings = new AddressSettings()
         .setAutoCreateAddresses(true)
         .setAutoCreateQueues(true)
         .setAutoDeleteAddresses(true)
         .setAutoDeleteQueues(true);

      Configuration config = createDefaultInVMConfig().addDivertConfiguration(divertConf).addAddressSetting("#", addressSettings);
      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(config, false));
      server.start();

      ServerLocator locator = createInVMNonHALocator();
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession session = sf.createSession(false, true, true);

      session.createQueue(QueueConfiguration.of(testAddress).setAddress(testAddress).setRoutingType(RoutingType.ANYCAST).setDurable(true));
      session.createQueue(QueueConfiguration.of(forwardAddress).setAddress(forwardAddress).setRoutingType(RoutingType.ANYCAST).setDurable(true));
      session.start();

      ClientProducer producer = session.createProducer(testAddress);
      ClientConsumer consumer = session.createConsumer(forwardAddress);

      final int numMessages = 5;

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session.createMessage(true);
         message.setRoutingType(RoutingType.ANYCAST);
         producer.send(message);
      }

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer.receive(DivertTest.TIMEOUT);
         assertNotNull(message);
         message.acknowledge();
      }

      assertNull(consumer.receiveImmediate());
      consumer.close();

      //Trigger autoDelete instead of waiting
      QueueManagerImpl.performAutoDeleteQueue(server, server.locateQueue(forwardAddress));
      Wait.assertTrue(() -> server.getPostOffice().getAddressInfo(SimpleString.of(forwardAddress))
         .getBindingRemovedTimestamp() != -1, DivertTest.TIMEOUT, 100);

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = session.createMessage(true);
         producer.send(message);
      }

      consumer = session.createConsumer(forwardAddress);
      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = consumer.receive(DivertTest.TIMEOUT);
         assertNotNull(message);
         message.acknowledge();
      }

      assertNull(consumer.receiveImmediate());
   }

}
