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

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

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
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 java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.util.EnumSet;
import java.util.UUID;

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.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.cli.commands.ActionContext;
import org.apache.activemq.artemis.cli.commands.tools.xml.XmlDataExporter;
import org.apache.activemq.artemis.cli.commands.tools.xml.XmlDataImporter;
import org.apache.activemq.artemis.core.persistence.impl.journal.BatchingIDGenerator;
import org.apache.activemq.artemis.core.persistence.impl.journal.JournalStorageManager;
import org.apache.activemq.artemis.core.persistence.impl.journal.LargeServerMessageImpl;
import org.apache.activemq.artemis.core.registry.JndiBindingRegistry;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.jms.server.JMSServerManager;
import org.apache.activemq.artemis.jms.server.impl.JMSServerManagerImpl;
import org.apache.activemq.artemis.tests.unit.util.InVMContext;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.CFUtil;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.UUIDGenerator;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;

/**
 * A test of the XML export/import functionality
 */
public class XmlImportExportTest extends ActiveMQTestBase {

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

   public XmlImportExportTest() {

   }

   public static final int CONSUMER_TIMEOUT = 5000;
   private static final String QUEUE_NAME = "A1";
   private ServerLocator locator;
   private ActiveMQServer server;
   private JMSServerManager jmsServer;
   private ClientSessionFactory factory;
   private InVMContext namingContext;

   // this is to force nextID > Integer.MAX_VALUE.
   // just to make it more challenging if there's any encoding using integer internally by mistake.
   protected void forceLong() {
      JournalStorageManager manager = (JournalStorageManager) server.getStorageManager();
      BatchingIDGenerator idGenerator = (BatchingIDGenerator) manager.getIDGenerator();
      idGenerator.forceNextID((Integer.MAX_VALUE) + 1L);
   }

   @Test
   public void testMessageProperties() throws Exception {
      ClientSession session = basicSetUp();

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

      ClientProducer producer = session.createProducer(QUEUE_NAME);

      StringBuilder international = new StringBuilder();
      for (char x = 800; x < 1200; x++) {
         international.append(x);
      }

      String special = "\"<>'&";

      for (int i = 0; i < 5; i++) {
         ClientMessage msg = session.createMessage(true);
         msg.getBodyBuffer().writeString("Bob the giant pig " + i);
         msg.putBooleanProperty("myBooleanProperty", Boolean.TRUE);
         msg.putByteProperty("myByteProperty", Byte.parseByte("0"));
         msg.putBytesProperty("myBytesProperty", new byte[]{0, 1, 2, 3, 4});
         msg.putDoubleProperty("myDoubleProperty", i * 1.6);
         msg.putFloatProperty("myFloatProperty", i * 2.5F);
         msg.putIntProperty("myIntProperty", i);
         msg.putLongProperty("myLongProperty", Long.MAX_VALUE - i);
         msg.putObjectProperty("myObjectProperty", i);
         msg.putObjectProperty("myNullObjectProperty", null);
         msg.putShortProperty("myShortProperty", Integer.valueOf(i).shortValue());
         msg.putStringProperty("myStringProperty", "myStringPropertyValue_" + i);
         msg.putStringProperty("myNullStringProperty", null);
         msg.putStringProperty("myNonAsciiStringProperty", international.toString());
         msg.putStringProperty("mySpecialCharacters", special);
         msg.putStringProperty(SimpleString.of("mySimpleStringProperty"), SimpleString.of("mySimpleStringPropertyValue_" + i));
         msg.putStringProperty(SimpleString.of("myNullSimpleStringProperty"), (SimpleString) null);
         producer.send(msg);
      }

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);
      ClientConsumer consumer = session.createConsumer(QUEUE_NAME);
      session.start();

      for (int i = 0; i < 5; i++) {
         ClientMessage msg = consumer.receive(CONSUMER_TIMEOUT);
         byte[] body = new byte[msg.getBodySize()];
         msg.getBodyBuffer().readBytes(body);
         assertTrue(new String(body).contains("Bob the giant pig " + i));
         assertEquals(Boolean.TRUE, msg.getBooleanProperty("myBooleanProperty"));
         assertEquals(msg.getByteProperty("myByteProperty"), Byte.valueOf("0"));
         byte[] bytes = msg.getBytesProperty("myBytesProperty");
         for (int j = 0; j < 5; j++) {
            assertEquals(j, bytes[j]);
         }
         assertEquals(i * 1.6, msg.getDoubleProperty("myDoubleProperty"), 0.000001);
         assertEquals(i * 2.5F, msg.getFloatProperty("myFloatProperty"), 0.000001);
         assertEquals(i, msg.getIntProperty("myIntProperty").intValue());
         assertEquals(Long.MAX_VALUE - i, msg.getLongProperty("myLongProperty").longValue());
         assertEquals(i, msg.getObjectProperty("myObjectProperty"));
         assertTrue(msg.getPropertyNames().contains(SimpleString.of("myNullObjectProperty")));
         assertNull(msg.getObjectProperty("myNullObjectProperty"));
         assertEquals(Integer.valueOf(i).shortValue(), msg.getShortProperty("myShortProperty").shortValue());
         assertEquals("myStringPropertyValue_" + i, msg.getStringProperty("myStringProperty"));
         assertTrue(msg.getPropertyNames().contains(SimpleString.of("myNullStringProperty")));
         assertNull(msg.getStringProperty("myNullStringProperty"));
         assertEquals(international.toString(), msg.getStringProperty("myNonAsciiStringProperty"));
         assertEquals(special, msg.getStringProperty("mySpecialCharacters"));
         assertEquals(SimpleString.of("mySimpleStringPropertyValue_" + i), msg.getSimpleStringProperty(SimpleString.of("mySimpleStringProperty")));
         assertTrue(msg.getPropertyNames().contains(SimpleString.of("myNullSimpleStringProperty")));
         assertNull(msg.getSimpleStringProperty("myNullSimpleStringProperty"));
      }
   }

   /**
    * @return ClientSession
    */
   private ClientSession basicSetUp() throws Exception {
      server = createServer(true);
      server.getConfiguration().getConnectorConfigurations().put("in-vm1", new TransportConfiguration(INVM_CONNECTOR_FACTORY));
      server.getConfiguration().getConnectorConfigurations().put("in-vm2", new TransportConfiguration(INVM_CONNECTOR_FACTORY));
      jmsServer = new JMSServerManagerImpl(server);
      addActiveMQComponent(jmsServer);
      namingContext = new InVMContext();
      jmsServer.setRegistry(new JndiBindingRegistry(namingContext));
      jmsServer.start();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      forceLong();
      return addClientSession(factory.createSession(false, true, true));
   }

   @Test
   public void testMessageTypes() throws Exception {

      ClientSession session = basicSetUp();

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

      ClientProducer producer = session.createProducer(QUEUE_NAME);

      ClientMessage msg = session.createMessage(Message.BYTES_TYPE, true);
      producer.send(msg);
      msg = session.createMessage(Message.DEFAULT_TYPE, true);
      producer.send(msg);
      msg = session.createMessage(Message.MAP_TYPE, true);
      producer.send(msg);
      msg = session.createMessage(Message.OBJECT_TYPE, true);
      producer.send(msg);
      msg = session.createMessage(Message.STREAM_TYPE, true);
      producer.send(msg);
      msg = session.createMessage(Message.TEXT_TYPE, true);
      producer.send(msg);
      msg = session.createMessage(true);
      producer.send(msg);

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);
      ClientConsumer consumer = session.createConsumer(QUEUE_NAME);
      session.start();

      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertEquals(Message.BYTES_TYPE, msg.getType());
      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertEquals(Message.DEFAULT_TYPE, msg.getType());
      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertEquals(Message.MAP_TYPE, msg.getType());
      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertEquals(Message.OBJECT_TYPE, msg.getType());
      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertEquals(Message.STREAM_TYPE, msg.getType());
      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertEquals(Message.TEXT_TYPE, msg.getType());
      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertEquals(Message.DEFAULT_TYPE, msg.getType());
   }

   @Test
   public void testTextMessage() throws Exception {
      String data = "X".repeat(2608);

      ClientSession session = basicSetUp();

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

      ClientProducer producer = session.createProducer(QUEUE_NAME);
      ClientMessage msg = session.createMessage(Message.TEXT_TYPE, true);
      msg.getBodyBuffer().writeString(data);
      producer.send(msg);

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);
      ClientConsumer consumer = session.createConsumer(QUEUE_NAME);
      session.start();

      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertEquals(Message.TEXT_TYPE, msg.getType());
      assertEquals(data, msg.getBodyBuffer().readString());
   }

   @Test
   public void testBytesMessage() throws Exception {
      String data = "X".repeat(2610);

      ClientSession session = basicSetUp();

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

      ClientProducer producer = session.createProducer(QUEUE_NAME);
      ClientMessage msg = session.createMessage(Message.BYTES_TYPE, true);
      msg.getBodyBuffer().writeBytes(data.getBytes());
      producer.send(msg);

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);
      ClientConsumer consumer = session.createConsumer(QUEUE_NAME);
      session.start();

      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertEquals(Message.BYTES_TYPE, msg.getType());
      byte[] result = new byte[msg.getBodySize()];
      msg.getBodyBuffer().readBytes(result);
      assertEquals(data.getBytes().length, result.length);
   }

   @Test
   public void testMessageAttributes() throws Exception {

      ClientSession session = basicSetUp();

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

      ClientProducer producer = session.createProducer(QUEUE_NAME);

      ClientMessage msg = session.createMessage(Message.BYTES_TYPE, true);
      msg.setExpiration(Long.MAX_VALUE);
      msg.setPriority((byte) 0);
      msg.setTimestamp(Long.MAX_VALUE - 1);
      msg.setUserID(UUIDGenerator.getInstance().generateUUID());
      producer.send(msg);

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);
      ClientConsumer consumer = session.createConsumer(QUEUE_NAME);
      session.start();

      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertEquals(Long.MAX_VALUE, msg.getExpiration());
      assertEquals((byte) 0, msg.getPriority());
      assertEquals(Long.MAX_VALUE - 1, msg.getTimestamp());
      assertNotNull(msg.getUserID());
   }

   @Test
   public void testBindingAttributes() throws Exception {
      ClientSession session = basicSetUp();

      session.createQueue(QueueConfiguration.of("queueName1").setAddress("addressName1"));
      session.createQueue(QueueConfiguration.of("queueName2").setAddress("addressName1").setFilterString("bob"));

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);

      ClientSession.QueueQuery queueQuery = session.queueQuery(SimpleString.of("queueName1"));

      assertEquals("addressName1", queueQuery.getAddress().toString());
      assertNull(queueQuery.getFilterString());

      queueQuery = session.queueQuery(SimpleString.of("queueName2"));

      assertEquals("addressName1", queueQuery.getAddress().toString());
      assertEquals("bob", queueQuery.getFilterString().toString());
      assertTrue(queueQuery.isDurable());
   }

   @Test
   public void testLargeMessage() throws Exception {
      server = createServer(true);
      server.start();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      ClientSession session = factory.createSession(false, false);

      LargeServerMessageImpl fileMessage = new LargeServerMessageImpl(server.getStorageManager());

      fileMessage.setMessageID(1005);
      fileMessage.setDurable(true);

      for (int i = 0; i < 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; i++) {
         fileMessage.addBytes(new byte[]{getSamplebyte(i)});
      }

      fileMessage.putLongProperty(Message.HDR_LARGE_BODY_SIZE, 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE);

      fileMessage.releaseResources(false, true);

      session.createQueue(QueueConfiguration.of("A"));

      ClientProducer prod = session.createProducer("A");

      prod.send(fileMessage);

      fileMessage.deleteFile();

      session.commit();

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);
      session.close();
      session = factory.createSession(false, false);
      session.start();

      ClientConsumer cons = session.createConsumer("A");

      ClientMessage msg = cons.receive(CONSUMER_TIMEOUT);

      assertNotNull(msg);

      assertEquals(2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, msg.getBodySize());

      for (int i = 0; i < 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; i++) {
         assertEquals(getSamplebyte(i), msg.getBodyBuffer().readByte());
      }

      msg.acknowledge();
      session.commit();
   }

   @Test
   public void testLargeMessagesNoTmpFiles() throws Exception {
      server = createServer(true);
      server.start();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      ClientSession session = factory.createSession(false, false);

      LargeServerMessageImpl fileMessage = new LargeServerMessageImpl(server.getStorageManager());

      fileMessage.setMessageID(1005);
      fileMessage.setDurable(true);

      for (int i = 0; i < 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; i++) {
         fileMessage.addBytes(new byte[]{getSamplebyte(i)});
      }

      fileMessage.putLongProperty(Message.HDR_LARGE_BODY_SIZE, 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE);

      fileMessage.releaseResources(false, true);

      session.createQueue(QueueConfiguration.of("A"));

      ClientProducer prod = session.createProducer("A");

      prod.send(fileMessage);
      prod.send(fileMessage);

      fileMessage.deleteFile();

      session.commit();

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.sort = true;
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);
      session.close();
      session = factory.createSession(false, false);
      session.start();

      ClientConsumer cons = session.createConsumer("A");

      ClientMessage msg = cons.receive(CONSUMER_TIMEOUT);
      assertNotNull(msg);
      assertEquals(2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, msg.getBodySize());

      for (int i = 0; i < 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; i++) {
         assertEquals(getSamplebyte(i), msg.getBodyBuffer().readByte());
      }
      msg = cons.receive(CONSUMER_TIMEOUT);
      assertNotNull(msg);
      assertEquals(2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, msg.getBodySize());

      for (int i = 0; i < 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; i++) {
         assertEquals(getSamplebyte(i), msg.getBodyBuffer().readByte());
      }

      msg.acknowledge();
      session.commit();

      //make sure there is not tmp file left
      File workingDir = new File(System.getProperty("user.dir"));
      String[] flist = workingDir.list();
      for (String fn : flist) {
         assertFalse(fn.endsWith(".tmp"), "leftover: " + fn);
      }
   }

   @Test
   public void testLargeJmsTextMessage() throws Exception {
      basicSetUp();
      ConnectionFactory cf = ActiveMQJMSClient.createConnectionFactory("vm://0", "test");
      Connection c = cf.createConnection();
      Session s = c.createSession(false, Session.AUTO_ACKNOWLEDGE);
      server.createQueue(QueueConfiguration.of("A").setRoutingType(RoutingType.ANYCAST));
      MessageProducer p = s.createProducer(ActiveMQJMSClient.createQueue("A"));
      p.setDeliveryMode(DeliveryMode.PERSISTENT);
      TextMessage textMessage = s.createTextMessage(RandomUtil.randomAlphaNumericString(200 * 1024));
      textMessage.setStringProperty("_AMQ_DUPL_ID", String.valueOf(UUID.randomUUID()));
      p.send(textMessage);
      c.close();

      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      ClientSession session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);
      session.close();

      c = cf.createConnection();
      s = c.createSession();
      MessageConsumer mc = s.createConsumer(ActiveMQJMSClient.createQueue("A"));
      c.start();
      javax.jms.Message msg = mc.receive(CONSUMER_TIMEOUT);

      assertNotNull(msg);

      c.close();
   }

   @Test
   public void testPartialQueue() throws Exception {
      ClientSession session = basicSetUp();

      session.createQueue(QueueConfiguration.of("myQueue1").setAddress("myAddress"));
      session.createQueue(QueueConfiguration.of("myQueue2").setAddress("myAddress"));

      ClientProducer producer = session.createProducer("myAddress");

      ClientMessage msg = session.createMessage(true);
      producer.send(msg);

      ClientConsumer consumer = session.createConsumer("myQueue1");
      session.start();
      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertNotNull(msg);
      msg.acknowledge();
      consumer.close();

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);
      consumer = session.createConsumer("myQueue1");
      session.start();
      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertNull(msg);
      consumer.close();

      consumer = session.createConsumer("myQueue2");
      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertNotNull(msg);
   }

   @Test
   public void testPagedMessageWithMissingBinding() throws Exception {
      final String MY_ADDRESS = "myAddress";
      final String MY_QUEUE = "myQueue";
      final String MY_QUEUE2 = "myQueue2";

      server = createServer(true);

      AddressSettings defaultSetting = new AddressSettings().setPageSizeBytes(10 * 1024).setMaxSizeBytes(20 * 1024).setMaxReadPageBytes(-1).setMaxReadPageMessages(-1);
      server.getAddressSettingsRepository().addMatch("#", defaultSetting);
      server.start();

      ServerLocator locator = createInVMNonHALocator()
         // Making it synchronous, just because we want to stop sending messages as soon as the page-store becomes in
         // page mode and we could only guarantee that by setting it to synchronous
         .setBlockOnNonDurableSend(true).setBlockOnDurableSend(true).setBlockOnAcknowledge(true);

      ClientSessionFactory factory = locator.createSessionFactory();
      ClientSession session = factory.createSession(false, true, true);

      session.createQueue(QueueConfiguration.of(MY_QUEUE).setAddress(MY_ADDRESS));
      session.createQueue(QueueConfiguration.of(MY_QUEUE2).setAddress(MY_ADDRESS));

      ClientProducer producer = session.createProducer(MY_ADDRESS);

      ClientMessage message = session.createMessage(true);
      message.getBodyBuffer().writeBytes(new byte[1024]);

      for (int i = 0; i < 200; i++) {
         producer.send(message);
      }

      session.deleteQueue(MY_QUEUE2);

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = locator.createSessionFactory();
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);

      ClientConsumer consumer = session.createConsumer(MY_QUEUE);

      session.start();

      for (int i = 0; i < 200; i++) {
         message = consumer.receive(CONSUMER_TIMEOUT);

         assertNotNull(message);
      }

      session.close();
      locator.close();
      server.stop();
   }

   @Test
   public void testPaging() throws Exception {
      final String MY_ADDRESS = "myAddress";
      final String MY_QUEUE = "myQueue";

      server = createServer(true);

      AddressSettings defaultSetting = new AddressSettings().setPageSizeBytes(10 * 1024).setMaxSizeBytes(20 * 1024).setMaxReadPageBytes(-1).setMaxReadPageMessages(-1);
      server.getAddressSettingsRepository().addMatch("#", defaultSetting);
      server.start();

      locator = createInVMNonHALocator()
         // Making it synchronous, just because we want to stop sending messages as soon as the page-store becomes in
         // page mode and we could only guarantee that by setting it to synchronous
         .setBlockOnNonDurableSend(true).setBlockOnDurableSend(true).setBlockOnAcknowledge(true);

      factory = createSessionFactory(locator);
      ClientSession session = factory.createSession(false, true, true);

      session.createQueue(QueueConfiguration.of(MY_QUEUE).setAddress(MY_ADDRESS));

      ClientProducer producer = session.createProducer(MY_ADDRESS);

      ClientMessage message = session.createMessage(true);
      message.getBodyBuffer().writeBytes(new byte[1024]);

      for (int i = 0; i < 200; i++) {
         producer.send(message);
      }

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);

      ClientConsumer consumer = session.createConsumer(MY_QUEUE);

      session.start();

      for (int i = 0; i < 200; i++) {
         message = consumer.receive(CONSUMER_TIMEOUT);

         assertNotNull(message);
      }
   }

   @Test
   public void testPagedLargeMessage() throws Exception {
      final String MY_ADDRESS = "myAddress";
      final String MY_QUEUE = "myQueue";

      server = createServer(true);

      AddressSettings defaultSetting = new AddressSettings().setPageSizeBytes(10 * 1024).setMaxSizeBytes(20 * 1024).setMaxReadPageBytes(-1).setMaxReadPageMessages(-1);
      server.getAddressSettingsRepository().addMatch("#", defaultSetting);
      server.start();

      ServerLocator locator = createInVMNonHALocator()
         // Making it synchronous, just because we want to stop sending messages as soon as the page-store becomes in
         // page mode and we could only guarantee that by setting it to synchronous
         .setBlockOnNonDurableSend(true).setBlockOnDurableSend(true).setBlockOnAcknowledge(true);

      ClientSessionFactory factory = locator.createSessionFactory();
      ClientSession session = factory.createSession(false, true, true);

      session.createQueue(QueueConfiguration.of(MY_QUEUE).setAddress(MY_ADDRESS));

      ClientProducer producer = session.createProducer(MY_ADDRESS);

      ClientMessage message = session.createMessage(true);
      message.getBodyBuffer().writeBytes(new byte[1024]);

      for (int i = 0; i < 200; i++) {
         producer.send(message);
      }

      LargeServerMessageImpl fileMessage = new LargeServerMessageImpl(server.getStorageManager());

      fileMessage.setMessageID(1005);
      fileMessage.setDurable(true);

      for (int i = 0; i < 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; i++) {
         fileMessage.addBytes(new byte[]{getSamplebyte(i)});
      }

      fileMessage.putLongProperty(Message.HDR_LARGE_BODY_SIZE, 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE);

      fileMessage.releaseResources(false, true);

      producer.send(fileMessage);

      fileMessage.deleteFile();

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      //System.out.print(new String(xmlOutputStream.toByteArray()));

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = locator.createSessionFactory();
      session = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session);

      ClientConsumer consumer = session.createConsumer(MY_QUEUE);

      session.start();

      for (int i = 0; i < 200; i++) {
         message = consumer.receive(CONSUMER_TIMEOUT);

         assertNotNull(message);
      }

      ClientMessage msg = consumer.receive(CONSUMER_TIMEOUT);

      assertNotNull(msg);

      assertEquals(2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, msg.getBodySize());

      for (int i = 0; i < 2 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; i++) {
         assertEquals(getSamplebyte(i), msg.getBodyBuffer().readByte());
      }

      session.close();
      locator.close();
      server.stop();
   }

   @Test
   public void testTransactional() throws Exception {
      ClientSession session = basicSetUp();

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

      ClientProducer producer = session.createProducer(QUEUE_NAME);

      ClientMessage msg = session.createMessage(true);
      producer.send(msg);

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);
      session = factory.createSession(false, false, true);
      ClientSession managementSession = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session, managementSession);
      ClientConsumer consumer = session.createConsumer(QUEUE_NAME);
      session.start();

      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertNotNull(msg);
   }

   @Test
   public void testBody() throws Exception {
      final String QUEUE_NAME = "A1";
      server = createServer(true);
      server.start();
      ServerLocator locator = createInVMNonHALocator();
      ClientSessionFactory factory = locator.createSessionFactory();
      ClientSession session = factory.createSession(false, true, true);

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

      ClientProducer producer = session.createProducer(QUEUE_NAME);

      ClientMessage msg = session.createMessage(Message.TEXT_TYPE, true);
      msg.getBodyBuffer().writeString("bob123");
      producer.send(msg);

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = locator.createSessionFactory();
      session = factory.createSession(false, false, true);
      ClientSession managementSession = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session, managementSession);
      ClientConsumer consumer = session.createConsumer(QUEUE_NAME);
      session.start();

      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertNotNull(msg);
      assertEquals("bob123", msg.getBodyBuffer().readString());

      session.close();
      locator.close();
      server.stop();
   }

   @Test
   public void testBody2() throws Exception {
      final String QUEUE_NAME = "A1";
      server = createServer(true);
      server.start();
      ServerLocator locator = createInVMNonHALocator();
      ClientSessionFactory factory = locator.createSessionFactory();
      ClientSession session = factory.createSession(false, true, true);

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

      ClientProducer producer = session.createProducer(QUEUE_NAME);

      ClientMessage msg = session.createMessage(true);
      byte[] bodyTst = new byte[10];
      for (int i = 0; i < 10; i++) {
         bodyTst[i] = (byte) (i + 1);
      }
      msg.getBodyBuffer().writeBytes(bodyTst);
      assertEquals(bodyTst.length, msg.getBodySize());
      producer.send(msg);

      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = locator.createSessionFactory();
      session = factory.createSession(false, false, true);
      ClientSession managementSession = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session, managementSession);
      ClientConsumer consumer = session.createConsumer(QUEUE_NAME);
      session.start();

      msg = consumer.receive(CONSUMER_TIMEOUT);
      assertNotNull(msg);
      assertEquals(msg.getBodySize(), bodyTst.length);
      byte[] bodyRead = new byte[bodyTst.length];
      msg.getBodyBuffer().readBytes(bodyRead);
      assertEqualsByteArrays(bodyTst, bodyRead);

      session.close();
      locator.close();
      server.stop();
   }

   @Test
   public void testRoutingTypes() throws Exception {
      SimpleString myAddress = SimpleString.of("myAddress");
      ClientSession session = basicSetUp();

      EnumSet<RoutingType> routingTypes = EnumSet.of(RoutingType.ANYCAST, RoutingType.MULTICAST);

      session.createAddress(myAddress, routingTypes, false);

      session.createQueue(QueueConfiguration.of("myQueue1").setAddress(myAddress));
      session.createQueue(QueueConfiguration.of("myQueue2").setAddress(myAddress));

      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = locator.createSessionFactory();
      session = factory.createSession(false, false, true);
      ClientSession managementSession = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session, managementSession);

      assertTrue(server.getAddressInfo(myAddress).getRoutingTypes().contains(RoutingType.ANYCAST));
      assertTrue(server.getAddressInfo(myAddress).getRoutingTypes().contains(RoutingType.MULTICAST));
   }

   @Test
   public void testEmptyRoutingTypes() throws Exception {
      SimpleString myAddress = SimpleString.of("myAddress");
      ClientSession session = basicSetUp();

      EnumSet<RoutingType> routingTypes = EnumSet.noneOf(RoutingType.class);

      session.createAddress(myAddress, routingTypes, false);

      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = locator.createSessionFactory();
      session = factory.createSession(false, false, true);
      ClientSession managementSession = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session, managementSession);

      assertEquals(0, server.getAddressInfo(myAddress).getRoutingTypes().size());
   }

   @Test
   public void testImportWrongRoutingType() throws Exception {
      SimpleString myAddress = SimpleString.of("myAddress");
      SimpleString myQueue = SimpleString.of("myQueue");
      SimpleString dla = SimpleString.of("DLA");
      SimpleString dlaPrefix = SimpleString.of("DLA.");
      String payload = "myMessagePayload";

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

      //Create ANYCAST queue and set "AutoCreateDeadLetterResources"
      //Send message with ANYCAST RoutingType
      server.getAddressSettingsRepository().addMatch(myAddress.toString(), new AddressSettings().setMaxDeliveryAttempts(1).setDeadLetterAddress(dla).setAutoCreateDeadLetterResources(true).setDeadLetterQueuePrefix(dlaPrefix).setMaxReadPageBytes(-1).setMaxReadPageMessages(-1));
      session.createQueue(QueueConfiguration.of(myQueue).setAddress(myAddress).setDurable(true).setRoutingType(RoutingType.ANYCAST));

      ClientProducer producer = session.createProducer(myAddress);
      producer.send(createTextMessage(session, payload).putByteProperty(Message.HDR_ROUTING_TYPE, (byte) 1));
      session.start();
      ClientConsumer consumer = session.createConsumer(myQueue);
      ClientMessage m = consumer.receive(5000);
      m.acknowledge();

      assertNotNull(m);
      assertEquals(payload, m.getBodyBuffer().readString());
      assertEquals(RoutingType.ANYCAST, m.getRoutingType());

      // Rollback to place ANYCAST message on DLA (MULTICAST)
      session.rollback();
      m = consumer.receiveImmediate();
      assertNull(m);

      consumer.close();
      session.close();
      locator.close();
      server.stop();

      ByteArrayOutputStream xmlOutputStream = new ByteArrayOutputStream();
      XmlDataExporter xmlDataExporter = new XmlDataExporter();
      xmlDataExporter.process(xmlOutputStream, server.getConfiguration().getBindingsDirectory(), server.getConfiguration().getJournalDirectory(), server.getConfiguration().getPagingDirectory(), server.getConfiguration().getLargeMessagesDirectory());
      if (logger.isDebugEnabled()) {
         logger.debug(new String(xmlOutputStream.toByteArray()));
      }

      clearDataRecreateServerDirs();
      server.start();
      forceLong();
      locator = createInVMNonHALocator();
      factory = locator.createSessionFactory();
      session = factory.createSession(false, false, true);
      ClientSession managementSession = factory.createSession(false, true, true);

      ByteArrayInputStream xmlInputStream = new ByteArrayInputStream(xmlOutputStream.toByteArray());
      XmlDataImporter xmlDataImporter = new XmlDataImporter();
      xmlDataImporter.validate(xmlInputStream);
      xmlInputStream.reset();
      xmlDataImporter.process(xmlInputStream, session, managementSession);

      //Check that message is imported with no "routingType" and is intact
      Wait.assertTrue(() -> server.getTotalMessageCount() == 1);
      session.start();
      consumer = session.createConsumer(dlaPrefix.concat(myAddress));
      m = consumer.receive(5000);

      assertNotNull(m);
      assertEquals(payload, m.getBodyBuffer().readString());
      assertNull(m.getRoutingType());

      consumer.close();
      session.close();
      locator.close();
      server.stop();
   }


   @Test
   public void testRemovedQueue() throws Exception {

      String undefinedPrefix = "undef_" + RandomUtil.randomUUIDString() + "_";
      final int numberOfMessages = 100;

      server = createServer(true, true);
      server.start();
      forceLong();

      String anycastQueueName = getTestClassName() + RandomUtil.randomUUIDString();
      String multicastQueueName = getTestClassName() + RandomUtil.randomUUIDString();
      createAnycastPair(server, anycastQueueName);
      server.addAddressInfo(new AddressInfo(multicastQueueName).addRoutingType(RoutingType.MULTICAST).setAutoCreated(false));
      server.createQueue(QueueConfiguration.of(multicastQueueName).setRoutingType(RoutingType.MULTICAST).setAddress(multicastQueueName));

      org.apache.activemq.artemis.core.server.Queue anycastServerQueue = server.locateQueue(anycastQueueName);
      assertNotNull(anycastServerQueue);
      assertEquals(RoutingType.ANYCAST, anycastServerQueue.getRoutingType());
      org.apache.activemq.artemis.core.server.Queue multiCastServerQueue = server.locateQueue(multicastQueueName);
      assertNotNull(multiCastServerQueue);
      assertEquals(RoutingType.MULTICAST, multiCastServerQueue.getRoutingType());

      {
         ConnectionFactory factory = CFUtil.createConnectionFactory("CORE", "tcp://localhost:61616");
         try (Connection connection = factory.createConnection()) {
            Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
            Topic topic = session.createTopic(multicastQueueName);
            Queue queue = session.createQueue(anycastQueueName);

            try (MessageProducer producer = session.createProducer(queue)) {
               for (int i = 0; i < numberOfMessages; i++) {
                  producer.send(session.createTextMessage("hello " + i));
               }
            }

            try (MessageProducer producer = session.createProducer(topic)) {
               for (int i = 0; i < numberOfMessages; i++) {
                  producer.send(session.createTextMessage("hello " + i));
               }
            }

            session.commit();
         }
      }

      // this is forcing a situation where the queue was removed and the messages are still in the journal
      removeAddressAndQueue(anycastServerQueue);
      removeAddressAndQueue(multiCastServerQueue);

      server.stop();

      final String fileName = "test.out";

      FileOutputStream fileOutputStream = new FileOutputStream(new File(getTestDir(), fileName));
      BufferedOutputStream bufferOut = new BufferedOutputStream(fileOutputStream);
      XmlDataExporter xmlDataExporter = new XmlDataExporter();

      xmlDataExporter.setUndefinedPrefix(undefinedPrefix);

      // the journal should still export even though the bindings don't exist any more
      // this is to "facilitate" users recovering or undoing mistakes
      xmlDataExporter.process(bufferOut, getBindingsDir(), getJournalDir(), getPageDir(), getLargeMessagesDir());
      bufferOut.close();
      assertNull(xmlDataExporter.getLastError());

      server.start();

      XmlDataImporter importer = new XmlDataImporter();
      importer.input = new File(getTestDir(), fileName).getAbsolutePath();
      importer.execute(new ActionContext());

      {
         ConnectionFactory factory = CFUtil.createConnectionFactory("CORE", "tcp://localhost:61616");
         try (Connection connection = factory.createConnection()) {
            Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
            connection.start();

            Queue anycastJMSQueue = session.createQueue(undefinedPrefix + anycastServerQueue.getID());
            Queue multicastJMSQueue = session.createQueue(undefinedPrefix + multiCastServerQueue.getID() + "::" + undefinedPrefix + multiCastServerQueue.getID());

            try (MessageConsumer consumer = session.createConsumer(anycastJMSQueue)) {
               for (int i = 0; i < numberOfMessages; i++) {
                  TextMessage message = (TextMessage) consumer.receive(5000);
                  assertNotNull(message);
                  assertEquals("hello " + i, message.getText());
               }
            }

            try (MessageConsumer consumer = session.createConsumer(multicastJMSQueue)) {
               for (int i = 0; i < numberOfMessages; i++) {
                  TextMessage message = (TextMessage) consumer.receive(5000);
                  assertNotNull(message);
                  assertEquals("hello " + i, message.getText());
               }
            }

            session.commit();
         }
      }
   }

   private void removeAddressAndQueue(org.apache.activemq.artemis.core.server.Queue serverQueue) throws Exception {
      AddressInfo addressInfo = server.getAddressInfo(serverQueue.getAddress());
      long tx = server.getStorageManager().generateID();
      server.getStorageManager().deleteAddressBinding(tx, addressInfo.getId());
      server.getStorageManager().deleteQueueBinding(tx, serverQueue.getID());
      server.getStorageManager().commitBindings(tx);
   }

}
