StompTest.java

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

import javax.jms.BytesMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.TextMessage;
import java.io.ByteArrayOutputStream;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.JsonUtil;
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.ActiveMQClient;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.ClientProducer;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
import org.apache.activemq.artemis.api.core.client.ServerLocator;
import org.apache.activemq.artemis.api.core.management.ActiveMQServerControl;
import org.apache.activemq.artemis.api.core.management.ManagementHelper;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.core.management.impl.view.ProducerField;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.QueueBinding;
import org.apache.activemq.artemis.core.postoffice.impl.LocalQueueBinding;
import org.apache.activemq.artemis.core.protocol.stomp.Stomp;
import org.apache.activemq.artemis.core.protocol.stomp.StompProtocolManager;
import org.apache.activemq.artemis.core.protocol.stomp.StompProtocolManagerFactory;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.jms.client.ActiveMQMessage;
import org.apache.activemq.artemis.json.JsonArray;
import org.apache.activemq.artemis.json.JsonObject;
import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
import org.apache.activemq.artemis.reader.MessageUtil;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.apache.activemq.artemis.tests.integration.mqtt.FuseMQTTClientProvider;
import org.apache.activemq.artemis.tests.integration.mqtt.MQTTClientProvider;
import org.apache.activemq.artemis.tests.integration.stomp.util.ClientStompFrame;
import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnection;
import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnectionFactory;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;

import static org.apache.activemq.artemis.utils.collections.IterableStream.iterableOf;

@RunWith(Parameterized.class)
public class StompTest extends StompTestBase {

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

   protected StompClientConnection conn;

   @Override
   protected ActiveMQServer createServer() throws Exception {
      ActiveMQServer server = super.createServer();
      server.getConfiguration().setAddressQueueScanPeriod(100);
      return server;
   }

   @Override
   @Before
   public void setUp() throws Exception {
      super.setUp();
      conn = StompClientConnectionFactory.createClientConnection(uri);
   }

   @Override
   @After
   public void tearDown() throws Exception {
      try {
         boolean connected = conn != null && conn.isConnected();
         logger.debug("Connection 1.0 connected: {}", connected);
         if (connected) {
            try {
               conn.disconnect();
            } catch (Exception e) {
               // ignore
            }
         }
      } finally {
         super.tearDown();
         conn.closeTransport();
      }
   }

   @Test
   public void testConnectionTTL() throws Exception {
      int port = 61614;

      URI uri = createStompClientUri(scheme, hostname, port);

      server.getRemotingService().createAcceptor("test", "tcp://127.0.0.1:" + port + "?connectionTtl=1000").start();
      StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri);
      conn.connect("brianm", "wombats");

      Thread.sleep(5000);

      ClientStompFrame frame = conn.receiveFrame();

      assertEquals(Stomp.Responses.ERROR, frame.getCommand());

      assertFalse(conn.isConnected());
   }

   @Test
   public void testSendManyMessages() throws Exception {
      conn.connect(defUser, defPass);

      MessageConsumer consumer = session.createConsumer(queue);

      int count = 1000;
      final CountDownLatch latch = new CountDownLatch(count);
      consumer.setMessageListener(new MessageListener() {

         @Override
         public void onMessage(Message arg0) {
            latch.countDown();
         }
      });

      for (int i = 1; i <= count; i++) {
         send(conn, getQueuePrefix() + getQueueName(), null, "Hello World!");
      }

      assertTrue(latch.await(60, TimeUnit.SECONDS));
   }

   @Test
   public void testProducerMetrics() throws Exception {
      conn.connect(defUser, defPass);

      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World!", true);

      String filterString = createJsonFilter("", "", "");
      String producersAsJsonString = server.getActiveMQServerControl().listProducers(filterString, 1, 50);
      JsonObject producersAsJsonObject = JsonUtil.readJsonObject(producersAsJsonString);
      JsonArray array = (JsonArray) producersAsJsonObject.get("data");

      Assert.assertEquals("number of producers returned from query", 1, array.size());

      JsonObject producer = array.getJsonObject(0);
      Assert.assertEquals(ProducerField.MESSAGE_SENT.getName(), 1, producer.getInt(ProducerField.MESSAGE_SENT.getName()));

      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World!", true);

      producersAsJsonString = server.getActiveMQServerControl().listProducers(filterString, 1, 50);
      producersAsJsonObject = JsonUtil.readJsonObject(producersAsJsonString);
      array = (JsonArray) producersAsJsonObject.get("data");

      Assert.assertEquals("number of producers returned from query", 1, array.size());

      producer = array.getJsonObject(0);
      Assert.assertEquals(ProducerField.MESSAGE_SENT.getName(), 2, producer.getInt(ProducerField.MESSAGE_SENT.getName()));

      conn.closeTransport();

      Wait.assertEquals(0, () -> ((JsonArray) JsonUtil.readJsonObject(server.getActiveMQServerControl().listProducers(filterString, 1, 50)).get("data")).size());

      producersAsJsonString = server.getActiveMQServerControl().listProducers(filterString, 1, 50);
      producersAsJsonObject = JsonUtil.readJsonObject(producersAsJsonString);
      array = (JsonArray) producersAsJsonObject.get("data");

      Assert.assertEquals("number of producers returned from query", 0, array.size());
   }

   @Test
   public void testSendOverDiskFull() throws Exception {
      try (AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler()) {
         MessageConsumer consumer = session.createConsumer(queue);

         conn.connect(defUser, defPass);
         int count = 1000;
         final CountDownLatch latch = new CountDownLatch(count);
         consumer.setMessageListener(new MessageListener() {

            @Override
            public void onMessage(Message arg0) {
               latch.countDown();
            }
         });

         ((ActiveMQServerImpl) server).getMonitor().setMaxUsage(0).tick();

         // Connection should be closed by broker when disk is full and attempt to send
         Exception e = null;
         try {
            for (int i = 1; i <= count; i++) {
               send(conn, getQueuePrefix() + getQueueName(), null, "Hello World!");
            }
         } catch (Exception se) {
            e = se;
         }
         assertNotNull(e);
         // It should encounter the exception on logs
         loggerHandler.findText("AMQ119119");
      }
   }

   @Test
   public void testConnect() throws Exception {
      ClientStompFrame frame = conn.createFrame(Stomp.Commands.CONNECT).addHeader(Stomp.Headers.Connect.LOGIN, defUser).addHeader(Stomp.Headers.Connect.PASSCODE, defPass).addHeader(Stomp.Headers.Connect.REQUEST_ID, "1");
      ClientStompFrame response = conn.sendFrame(frame);

      Assert.assertTrue(response.getCommand().equals(Stomp.Responses.CONNECTED));
      Assert.assertTrue(response.getHeader(Stomp.Headers.Connected.RESPONSE_ID).equals("1"));
   }

   @Test
   public void testDisconnectAndError() throws Exception {
      ClientStompFrame frame = conn.createFrame(Stomp.Commands.CONNECT).addHeader(Stomp.Headers.Connect.LOGIN, defUser).addHeader(Stomp.Headers.Connect.PASSCODE, defPass).addHeader(Stomp.Headers.Connect.REQUEST_ID, "1");
      ClientStompFrame response = conn.sendFrame(frame);

      Assert.assertTrue(response.getCommand().equals(Stomp.Responses.CONNECTED));
      Assert.assertTrue(response.getHeader(Stomp.Headers.Connected.RESPONSE_ID).equals("1"));

      conn.disconnect();

      // sending a message will result in an error
      try {
         send(conn, getQueuePrefix() + getQueueName(), null, "Hello World!");
         fail("Should have thrown an exception since connection is disconnected");
      } catch (Exception e) {
         // ignore
      }
   }

   @Test
   public void testSendMessage() throws Exception {

      MessageConsumer consumer = session.createConsumer(queue);

      conn.connect(defUser, defPass);

      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World");

      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());
      // Assert default priority 4 is used when priority header is not set
      Assert.assertEquals("getJMSPriority", 4, message.getJMSPriority());

      // Make sure that the timestamp is valid - should
      // be very close to the current time.
      long tnow = System.currentTimeMillis();
      long tmsg = message.getJMSTimestamp();
      Assert.assertTrue(Math.abs(tnow - tmsg) < 1000);
   }

   @Test
   public void testNullCorrelationIDandTypeProperties() throws Exception {

      MessageConsumer consumer = session.createConsumer(queue);

      conn.connect(defUser, defPass);

      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World");

      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertFalse(((ActiveMQMessage)message).getCoreMessage().getPropertyNames().contains(MessageUtil.CORRELATIONID_HEADER_NAME));
      Assert.assertFalse(((ActiveMQMessage)message).getCoreMessage().getPropertyNames().contains(MessageUtil.TYPE_HEADER_NAME));
   }

   @Test
   public void sendSTOMPReceiveMQTT() throws Exception {
      // Set up MQTT Subscription
      MQTTClientProvider clientProvider = new FuseMQTTClientProvider();
      clientProvider.connect("tcp://" + hostname + ":" + port);
      clientProvider.subscribe(getTopicPrefix() + getTopicName(), 0);

      String stompPayload = "This is a test message";

      // Set up STOMP connection and send STOMP Message
      conn.connect(defUser, defPass);
      send(conn, getTopicPrefix() + getTopicName(), null, stompPayload);

      // Receive MQTT Message
      byte[] mqttPayload = clientProvider.receive(10000);
      clientProvider.disconnect();

      assertEquals(stompPayload, new String(mqttPayload, StandardCharsets.UTF_8));
      clientProvider.disconnect();
   }

   @Test
   public void sendMQTTReceiveSTOMP() throws Exception {
      String payload = "This is a test message";

      // Set up STOMP subscription
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      // Send MQTT Message
      MQTTClientProvider clientProvider = new FuseMQTTClientProvider();
      clientProvider.connect("tcp://" + hostname + ":" + port);
      clientProvider.publish(getQueuePrefix() + getQueueName(), payload.getBytes(), 0);
      clientProvider.disconnect();

      // Receive STOMP Message
      ClientStompFrame frame = conn.receiveFrame();
      assertTrue(frame.getBody().contains(payload));

   }

   @Test
   public void sendEmptyCoreMessage() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      // send core JMS message
      MessageProducer mp = session.createProducer(session.createQueue(getQueuePrefix() + getQueueName()));
      Message m = session.createMessage();
      mp.send(m);

      // Receive STOMP Message
      ClientStompFrame frame = conn.receiveFrame();
      assertNotNull(frame);
      assertNull(frame.getBody());
   }

   public void sendMessageToNonExistentQueue(String queuePrefix,
                                             String queue,
                                             RoutingType routingType) throws Exception {
      conn.connect(defUser, defPass);
      send(conn, queuePrefix + queue, null, "Hello World", true, routingType);

      MessageConsumer consumer = session.createConsumer(ActiveMQJMSClient.createQueue(queue));
      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());
      // Assert default priority 4 is used when priority header is not set
      Assert.assertEquals("getJMSPriority", 4, message.getJMSPriority());

      // Make sure that the timestamp is valid - should
      // be very close to the current time.
      long tnow = System.currentTimeMillis();
      long tmsg = message.getJMSTimestamp();
      Assert.assertTrue(Math.abs(tnow - tmsg) < 1500);

      // closing the consumer here should trigger auto-deletion
      assertNotNull(server.getPostOffice().getBinding(new SimpleString(queue)));
      consumer.close();
      Wait.assertTrue(() -> server.getPostOffice().getBinding(new SimpleString(queue)) == null);
   }

   @Test
   public void testSendMessageToNonExistentQueue() throws Exception {
      sendMessageToNonExistentQueue(getQueuePrefix(), RandomUtil.randomString(), RoutingType.ANYCAST);
   }

   @Test
   public void testSendMessageToNonExistentQueueUsingExplicitDefaultRouting() throws Exception {
      String nonExistentQueue = RandomUtil.randomString();
      server.getAddressSettingsRepository().addMatch(nonExistentQueue, new AddressSettings().setDefaultAddressRoutingType(RoutingType.ANYCAST).setDefaultQueueRoutingType(RoutingType.ANYCAST));
      sendMessageToNonExistentQueue(getQueuePrefix(), nonExistentQueue, null);
   }

   public void sendMessageToNonExistentTopic(String topicPrefix,
                                             String topic,
                                             RoutingType routingType) throws Exception {
      conn.connect(defUser, defPass);

      // first send a message to ensure that sending to a non-existent topic won't throw an error
      send(conn, topicPrefix + topic, null, "Hello World", true, routingType);

      // create a subscription on the topic and send/receive another message
      MessageConsumer consumer = session.createConsumer(ActiveMQJMSClient.createTopic(topic));
      send(conn, topicPrefix + topic, null, "Hello World", true, routingType);
      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());
      // Assert default priority 4 is used when priority header is not set
      Assert.assertEquals("getJMSPriority", 4, message.getJMSPriority());

      // Make sure that the timestamp is valid - should
      // be very close to the current time.
      long tnow = System.currentTimeMillis();
      long tmsg = message.getJMSTimestamp();
      Assert.assertTrue(Math.abs(tnow - tmsg) < 1500);

      assertNotNull(server.getAddressInfo(new SimpleString(topic)));

      // closing the consumer here should trigger auto-deletion of the subscription queue and address
      consumer.close();
      Wait.assertTrue(() -> server.getAddressInfo(new SimpleString(topic)) == null);
   }

   @Test
   public void testSendMessageToNonExistentTopic() throws Exception {
      sendMessageToNonExistentTopic(getTopicPrefix(), RandomUtil.randomString(), RoutingType.MULTICAST);
   }

   @Test
   public void testSendMessageToNonExistentTopicUsingExplicitDefaultRouting() throws Exception {
      String nonExistentTopic = RandomUtil.randomString();
      server.getAddressSettingsRepository().addMatch(nonExistentTopic, new AddressSettings().setDefaultAddressRoutingType(RoutingType.MULTICAST).setDefaultQueueRoutingType(RoutingType.MULTICAST));
      sendMessageToNonExistentTopic(getTopicPrefix(), nonExistentTopic, null);
   }

   @Test
   public void testSendMessageToNonExistentTopicUsingImplicitDefaultRouting() throws Exception {
      sendMessageToNonExistentTopic(getTopicPrefix(), RandomUtil.randomString(), null);
   }

   /*
    * Some STOMP clients erroneously put a new line \n *after* the terminating NUL char at the end of the frame
    * This means next frame read might have a \n a the beginning.
    * This is contrary to STOMP spec but we deal with it so we can work nicely with crappy STOMP clients
    */
   @Test
   public void testSendMessageWithLeadingNewLine() throws Exception {
      conn.connect(defUser, defPass);
      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).setBody("Hello World");
      conn.sendWickedFrame(frame);

      MessageConsumer consumer = session.createConsumer(queue);
      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());

      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World");
      message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());

      // Make sure that the timestamp is valid - should
      // be very close to the current time.
      long tnow = System.currentTimeMillis();
      long tmsg = message.getJMSTimestamp();
      Assert.assertTrue(Math.abs(tnow - tmsg) < 1000);
   }

   @Test
   public void testSendMessageWithReceipt() throws Exception {
      MessageConsumer consumer = session.createConsumer(queue);

      conn.connect(defUser, defPass);

      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World", true);

      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());

      // Make sure that the timestamp is valid - should
      // be very close to the current time.
      long tnow = System.currentTimeMillis();
      long tmsg = message.getJMSTimestamp();
      Assert.assertTrue(Math.abs(tnow - tmsg) < 1000);
   }

   @Test
   public void testSendMessageWithContentLength() throws Exception {
      MessageConsumer consumer = session.createConsumer(queue);
      conn.connect(defUser, defPass);

      byte[] data = new byte[]{1, 0, 0, 4};
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      baos.write(data);
      baos.flush();

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader(Stomp.Headers.CONTENT_LENGTH, Integer.toString(data.length)).setBody(new String(baos.toByteArray()));
      conn.sendFrame(frame);

      BytesMessage message = (BytesMessage) consumer.receive(10000);
      Assert.assertNotNull(message);
      assertEquals(data.length, message.getBodyLength());
      assertEquals(data[0], message.readByte());
      assertEquals(data[1], message.readByte());
      assertEquals(data[2], message.readByte());
      assertEquals(data[3], message.readByte());
   }

   @Test
   public void testJMSXGroupIdCanBeSet() throws Exception {
      final String jmsxGroupID = "JMSXGroupID";
      MessageConsumer consumer = session.createConsumer(queue);

      conn.connect(defUser, defPass);

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader("JMSXGroupID", jmsxGroupID).setBody("Hello World");
      conn.sendFrame(frame);

      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());
      // differ from StompConnect
      Assert.assertEquals(jmsxGroupID, message.getStringProperty("JMSXGroupID"));
   }

   @Test
   public void testSendMessageWithCustomHeadersAndSelector() throws Exception {

      MessageConsumer consumer = session.createConsumer(queue, "foo = 'abc'");

      conn.connect(defUser, defPass);

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader("foo", "abc").addHeader("bar", "123").setBody("Hello World");
      conn.sendFrame(frame);

      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());
      Assert.assertEquals("foo", "abc", message.getStringProperty("foo"));
      Assert.assertEquals("bar", "123", message.getStringProperty("bar"));
   }

   @Test
   public void testSendMessageWithCustomHeadersAndHyphenatedSelector() throws Exception {

      MessageConsumer consumer = session.createConsumer(queue, "hyphenated_props:b-ar = '123'");

      conn.connect(defUser, defPass);

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader("foo", "abc").addHeader("b-ar", "123").setBody("Hello World");
      conn.sendFrame(frame);

      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());
      Assert.assertEquals("foo", "abc", message.getStringProperty("foo"));
      Assert.assertEquals("b-ar", "123", message.getStringProperty("b-ar"));
   }

   @Test
   public void testSendMessageWithStandardHeaders() throws Exception {

      MessageConsumer consumer = session.createConsumer(queue);

      conn.connect(defUser, defPass);

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader("foo", "abc").addHeader("bar", "123").addHeader("correlation-id", "c123").addHeader("persistent", "true").addHeader("type", "t345").addHeader("JMSXGroupID", "abc").addHeader("priority", "3").setBody("Hello World");
      conn.sendFrame(frame);

      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());
      Assert.assertEquals("JMSCorrelationID", "c123", message.getJMSCorrelationID());
      Assert.assertEquals("getJMSType", "t345", message.getJMSType());
      Assert.assertEquals("getJMSPriority", 3, message.getJMSPriority());
      Assert.assertEquals(javax.jms.DeliveryMode.PERSISTENT, message.getJMSDeliveryMode());
      Assert.assertEquals("foo", "abc", message.getStringProperty("foo"));
      Assert.assertEquals("bar", "123", message.getStringProperty("bar"));

      Assert.assertEquals("JMSXGroupID", "abc", message.getStringProperty("JMSXGroupID"));
      // FIXME do we support it?
      // Assert.assertEquals("GroupID", "abc", amqMessage.getGroupID());
   }

   @Test
   public void testSendMessageWithLongHeaders() throws Exception {
      MessageConsumer consumer = session.createConsumer(queue);

      conn.connect(defUser, defPass);

      StringBuffer buffer = new StringBuffer();
      for (int i = 0; i < 1024; i++) {
         buffer.append("a");
      }

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader("foo", "abc").addHeader("bar", "123").addHeader("correlation-id", "c123").addHeader("persistent", "true").addHeader("type", "t345").addHeader("JMSXGroupID", "abc").addHeader("priority", "3").addHeader("longHeader", buffer.toString()).setBody("Hello World");
      conn.sendFrame(frame);

      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());
      Assert.assertEquals("JMSCorrelationID", "c123", message.getJMSCorrelationID());
      Assert.assertEquals("getJMSType", "t345", message.getJMSType());
      Assert.assertEquals("getJMSPriority", 3, message.getJMSPriority());
      Assert.assertEquals(javax.jms.DeliveryMode.PERSISTENT, message.getJMSDeliveryMode());
      Assert.assertEquals("foo", "abc", message.getStringProperty("foo"));
      Assert.assertEquals("longHeader", 1024, message.getStringProperty("longHeader").length());

      Assert.assertEquals("JMSXGroupID", "abc", message.getStringProperty("JMSXGroupID"));
   }

   @Test
   public void testSendMessageWithDelay() throws Exception {
      MessageConsumer consumer = session.createConsumer(queue);

      conn.connect(defUser, defPass);

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader("foo", "abc").addHeader("bar", "123").addHeader("correlation-id", "c123").addHeader("persistent", "true").addHeader("type", "t345").addHeader("JMSXGroupID", "abc").addHeader("priority", "3").addHeader("AMQ_SCHEDULED_DELAY", "2000").setBody("Hello World");
      conn.sendFrame(frame);

      assertNull("Should not receive message yet", consumer.receive(100));

      TextMessage message = (TextMessage) consumer.receive(4000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());
      Assert.assertEquals("JMSCorrelationID", "c123", message.getJMSCorrelationID());
      Assert.assertEquals("getJMSType", "t345", message.getJMSType());
      Assert.assertEquals("getJMSPriority", 3, message.getJMSPriority());
      Assert.assertEquals(javax.jms.DeliveryMode.PERSISTENT, message.getJMSDeliveryMode());
      Assert.assertEquals("foo", "abc", message.getStringProperty("foo"));
      Assert.assertEquals("JMSXGroupID", "abc", message.getStringProperty("JMSXGroupID"));
   }

   @Test
   public void testSendMessageWithDeliveryTime() throws Exception {
      MessageConsumer consumer = session.createConsumer(queue);

      conn.connect(defUser, defPass);

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader("foo", "abc").addHeader("bar", "123").addHeader("correlation-id", "c123").addHeader("persistent", "true").addHeader("type", "t345").addHeader("JMSXGroupID", "abc").addHeader("priority", "3").addHeader("AMQ_SCHEDULED_TIME", Long.toString(System.currentTimeMillis() + 2000)).setBody("Hello World");
      conn.sendFrame(frame);

      assertNull("Should not receive message yet", consumer.receive(100));

      TextMessage message = (TextMessage) consumer.receive(4000);
      Assert.assertNotNull(message);
      Assert.assertEquals("Hello World", message.getText());
      Assert.assertEquals("JMSCorrelationID", "c123", message.getJMSCorrelationID());
      Assert.assertEquals("getJMSType", "t345", message.getJMSType());
      Assert.assertEquals("getJMSPriority", 3, message.getJMSPriority());
      Assert.assertEquals(javax.jms.DeliveryMode.PERSISTENT, message.getJMSDeliveryMode());
      Assert.assertEquals("foo", "abc", message.getStringProperty("foo"));
      Assert.assertEquals("JMSXGroupID", "abc", message.getStringProperty("JMSXGroupID"));
   }

   @Test
   public void testSendMessageWithDelayWithBadValue() throws Exception {
      MessageConsumer consumer = session.createConsumer(queue);

      conn.connect(defUser, defPass);

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader("AMQ_SCHEDULED_DELAY", "foo").setBody("Hello World");
      conn.sendFrame(frame);

      assertNull("Should not receive message yet", consumer.receive(100));

      ClientStompFrame error = conn.receiveFrame();

      Assert.assertNotNull(error);
      Assert.assertTrue(error.getCommand().equals("ERROR"));
   }

   @Test
   public void testSendMessageWithDeliveryTimeWithBadValue() throws Exception {
      MessageConsumer consumer = session.createConsumer(queue);

      conn.connect(defUser, defPass);

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader("AMQ_SCHEDULED_TIME", "foo").setBody("Hello World");
      conn.sendFrame(frame);

      assertNull("Should not receive message yet", consumer.receive(100));

      ClientStompFrame error = conn.receiveFrame();

      Assert.assertNotNull(error);
      Assert.assertTrue(error.getCommand().equals("ERROR"));
   }

   @Test
   public void testSubscribeWithAutoAck() throws Exception {
      conn.connect(defUser, defPass);

      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      sendJmsMessage(getName());

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals(getName(), frame.getBody());

      Pattern cl = Pattern.compile(Stomp.Headers.CONTENT_LENGTH + ":\\s*(\\d+)", Pattern.CASE_INSENSITIVE);
      Matcher cl_matcher = cl.matcher(frame.toString());
      Assert.assertFalse(cl_matcher.find());

      conn.disconnect();

      // message should not be received as it was auto-acked
      MessageConsumer consumer = session.createConsumer(queue);
      Message message = consumer.receive(100);
      Assert.assertNull(message);

   }

   @Test
   public void testNullPropertyValue() throws Exception {
      conn.connect(defUser, defPass);

      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      sendJmsMessage(getName(), "foo", null);

      ClientStompFrame frame = conn.receiveFrame(2000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals(getName(), frame.getBody());
      Assert.assertEquals("", frame.getHeader("foo"));

      conn.disconnect();
   }

   @Test
   public void testTransactedSessionLeak() throws Exception {
      for (int i = 0; i < 10; i++) {
         conn = StompClientConnectionFactory.createClientConnection(uri);
         conn.connect(defUser, defPass);


         for (int s = 0; s < 10; s++) {
            String txId = "tx" + i + "_" + s;
            beginTransaction(conn, txId);
            send(conn, getQueuePrefix() + getQueueName(), null, "Hello World", true, null, txId);
            commitTransaction(conn, txId, true);
         }

         Wait.assertEquals(13, () -> server.getSessions().size(), 1000, 100);
         conn.disconnect();
      }

      if (connection != null) {
         connection.close();
      }

      Wait.assertEquals(0, () -> server.getSessions().size(), 1000, 100);

      Acceptor stompAcceptor = server.getRemotingService().getAcceptors().get("stomp");
      StompProtocolManager stompProtocolManager = (StompProtocolManager) stompAcceptor.getProtocolHandler().getProtocolMap().get("STOMP");
      Assert.assertNotNull(stompProtocolManager);

      Assert.assertEquals(0, stompProtocolManager.getTransactedSessions().size());
   }

   @Test
   public void testIngressTimestamp() throws Exception {
      server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setEnableIngressTimestamp(true));
      conn.connect(defUser, defPass);

      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);
      long beforeSend = System.currentTimeMillis();
      sendJmsMessage(getName());
      long afterSend = System.currentTimeMillis();

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      String ingressTimestampHeader = frame.getHeader(Stomp.Headers.Message.INGRESS_TIMESTAMP);
      Assert.assertNotNull(ingressTimestampHeader);
      long ingressTimestamp = Long.parseLong(ingressTimestampHeader);
      assertTrue("Ingress timstamp " + ingressTimestamp + " should be >= " + beforeSend + " and <= " + afterSend,ingressTimestamp >= beforeSend && ingressTimestamp <= afterSend);

      conn.disconnect();
   }

   @Test
   public void testAnycastDestinationTypeMessageProperty() throws Exception {
      conn.connect(defUser, defPass);

      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      send(conn, getQueuePrefix() + getQueueName(), null, getName(), true, RoutingType.ANYCAST);

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals(RoutingType.ANYCAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
      Assert.assertTrue(frame.getHeader(org.apache.activemq.artemis.api.core.Message.HDR_ROUTING_TYPE.toString()) == null);
      Assert.assertEquals(getName(), frame.getBody());

      conn.disconnect();
   }

   @Test
   public void testMulticastDestinationTypeMessageProperty() throws Exception {
      conn.connect(defUser, defPass);

      subscribeTopic(conn, null, null, null);

      send(conn, getTopicPrefix() + getTopicName(), null, getName(), true, RoutingType.MULTICAST);

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals(RoutingType.MULTICAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
      Assert.assertTrue(frame.getHeader(org.apache.activemq.artemis.api.core.Message.HDR_ROUTING_TYPE.toString()) == null);
      Assert.assertEquals(getName(), frame.getBody());

      conn.disconnect();
   }

   @Test
   public void testSubscribeWithAutoAckAndBytesMessage() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      byte[] payload = new byte[]{1, 2, 3, 4, 5};

      sendJmsMessage(payload, queue);

      ClientStompFrame frame = conn.receiveFrame(10000);

      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());

      Pattern cl = Pattern.compile(Stomp.Headers.CONTENT_LENGTH + ":\\s*(\\d+)", Pattern.CASE_INSENSITIVE);
      Matcher cl_matcher = cl.matcher(frame.toString());
      Assert.assertTrue(cl_matcher.find());
      Assert.assertEquals("5", cl_matcher.group(1));

      Assert.assertFalse(Pattern.compile("type:\\s*null", Pattern.CASE_INSENSITIVE).matcher(frame.toString()).find());
      Assert.assertTrue(frame.getBody().toString().indexOf(new String(payload)) > -1);

      conn.disconnect();
   }

   @Test
   public void testSubscribeWithMessageSentWithProperties() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      MessageProducer producer = session.createProducer(queue);
      BytesMessage message = session.createBytesMessage();
      message.setStringProperty("S", "value");
      message.setBooleanProperty("n", false);
      message.setByteProperty("byte", (byte) 9);
      message.setDoubleProperty("d", 2.0);
      message.setFloatProperty("f", (float) 6.0);
      message.setIntProperty("i", 10);
      message.setLongProperty("l", 121);
      message.setShortProperty("s", (short) 12);
      message.writeBytes("Hello World".getBytes(StandardCharsets.UTF_8));
      producer.send(message);

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertNotNull(frame);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals("value", frame.getHeader("S"));
      Assert.assertEquals("false", frame.getHeader("n"));
      Assert.assertEquals("9", frame.getHeader("byte"));
      Assert.assertEquals("2.0", frame.getHeader("d"));
      Assert.assertEquals("6.0", frame.getHeader("f"));
      Assert.assertEquals("10", frame.getHeader("i"));
      Assert.assertEquals("121", frame.getHeader("l"));
      Assert.assertEquals("12", frame.getHeader("s"));
      Assert.assertEquals("Hello World", frame.getBody());

      conn.disconnect();
   }

   @Test
   public void testSubscribeWithID() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, "mysubid", Stomp.Headers.Subscribe.AckModeValues.AUTO);

      sendJmsMessage(getName());

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Message.DESTINATION));
      Assert.assertEquals("mysubid", frame.getHeader(Stomp.Headers.Message.SUBSCRIPTION));
      Assert.assertEquals(getName(), frame.getBody());

      conn.disconnect();
   }

   //
   @Test
   public void testBodyWithUTF8() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      String text = "A" + "\u00ea" + "\u00f1" + "\u00fc" + "C";
      logger.debug(text);
      sendJmsMessage(text);

      ClientStompFrame frame = conn.receiveFrame(10000);
      logger.debug("{}", frame);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Message.DESTINATION));
      Assert.assertEquals(text, frame.getBody());

      conn.disconnect();
   }

   @Test
   public void testMessagesAreInOrder() throws Exception {
      int ctr = 10;
      String[] data = new String[ctr];

      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      for (int i = 0; i < ctr; ++i) {
         data[i] = getName() + i;
         sendJmsMessage(data[i]);
      }

      for (int i = 0; i < ctr; ++i) {
         ClientStompFrame frame = conn.receiveFrame(1000);
         Assert.assertTrue("Message not in order", frame.getBody().equals(data[i]));
      }

      // sleep a while before publishing another set of messages
      Thread.sleep(200);

      for (int i = 0; i < ctr; ++i) {
         data[i] = getName() + Stomp.Headers.SEPARATOR + "second:" + i;
         sendJmsMessage(data[i]);
      }

      for (int i = 0; i < ctr; ++i) {
         ClientStompFrame frame = conn.receiveFrame(1000);
         Assert.assertTrue("Message not in order", frame.getBody().equals(data[i]));
      }

      conn.disconnect();
   }

   @Test
   public void testSubscribeWithAutoAckAndSelector() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO, null, "foo = 'zzz'");

      sendJmsMessage("Ignored message", "foo", "1234");
      sendJmsMessage("Real message", "foo", "zzz");

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertTrue("Should have received the real message but got: " + frame, frame.getBody().equals("Real message"));

      conn.disconnect();
   }

   @Test
   public void testSubscribeWithAutoAckAndHyphenatedSelector() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO, null, "hyphenated_props:foo-bar = 'zzz'");

      ServerLocator serverLocator = addServerLocator(ActiveMQClient.createServerLocator("vm://0"));
      ClientSessionFactory clientSessionFactory = serverLocator.createSessionFactory();
      ClientSession clientSession = clientSessionFactory.createSession(true, true);
      ClientProducer producer = clientSession.createProducer(getQueuePrefix() + getQueueName());

      ClientMessage ignoredMessage = clientSession.createMessage(false);
      ignoredMessage.putStringProperty("foo-bar", "1234");
      ignoredMessage.getBodyBuffer().writeNullableSimpleString(SimpleString.toSimpleString("Ignored message"));

      ClientMessage realMessage = clientSession.createMessage(false);
      realMessage.putStringProperty("foo-bar", "zzz");
      realMessage.getBodyBuffer().writeNullableSimpleString(SimpleString.toSimpleString("Real message"));

      producer.send(ignoredMessage);
      producer.send(realMessage);

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertTrue("Should have received the real message but got: " + frame, frame.getBody().equals("Real message"));

      conn.disconnect();
   }

   @Test
   public void testSubscribeWithAutoAckAndXpathSelector() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO, null, "XPATH 'root/a'");

      sendJmsMessage("<root><b key='first' num='1'/><c key='second' num='2'>c</c></root>");
      sendJmsMessage("<root><a key='first' num='1'/><b key='second' num='2'>b</b></root>");

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertTrue("Should have received the real message but got: " + frame, frame.getBody().equals("<root><a key='first' num='1'/><b key='second' num='2'>b</b></root>"));

      conn.disconnect();
   }

   @Test
   public void testSubscribeWithClientAck() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.CLIENT);

      sendJmsMessage(getName());
      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertNotNull(frame.getHeader(Stomp.Headers.Message.MESSAGE_ID));
      ack(conn, null, frame);

      conn.disconnect();

      // message should not be received since message was acknowledged by the client
      MessageConsumer consumer = session.createConsumer(queue);
      Message message = consumer.receive(100);
      Assert.assertNull(message);
   }

   @Test
   public void testRedeliveryWithClientAck() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.CLIENT);

      sendJmsMessage(getName());
      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());

      conn.disconnect();

      // message should be received since message was not acknowledged
      MessageConsumer consumer = session.createConsumer(queue);
      Message message = consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertTrue(message.getJMSRedelivered());
   }

   @Test
   public void testSubscribeWithClientAckThenConsumingAgainWithAutoAckWithNoDisconnectFrame() throws Exception {
      assertSubscribeWithClientAckThenConsumeWithAutoAck(false);
   }

   @Test
   public void testSubscribeWithClientAckThenConsumingAgainWithAutoAckWithExplicitDisconnect() throws Exception {
      assertSubscribeWithClientAckThenConsumeWithAutoAck(true);
   }

   protected void assertSubscribeWithClientAckThenConsumeWithAutoAck(boolean sendDisconnect) throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.CLIENT);
      sendJmsMessage(getName());

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());

      logger.debug("Reconnecting!");

      if (sendDisconnect) {
         conn.disconnect();
         conn.destroy();
         conn = StompClientConnectionFactory.createClientConnection(uri);
      } else {
         conn.destroy();
         conn = StompClientConnectionFactory.createClientConnection(uri);
      }

      // message should be received since message was not acknowledged
      conn.connect(defUser, defPass);

      subscribe(conn, null);

      frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());

      conn.disconnect();
      conn.destroy();

      conn = StompClientConnectionFactory.createClientConnection(uri);

      // now let's make sure we don't see the message again

      conn.connect(defUser, defPass);

      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO, null, true);

      sendJmsMessage("shouldBeNextMessage");

      frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals("shouldBeNextMessage", frame.getBody());
   }

   @Test
   public void testUnsubscribe() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      // send a message to our queue
      sendJmsMessage("first message");

      // receive message
      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());

      // remove suscription
      unsubscribe(conn, null, getQueuePrefix() + getQueueName(), true, false);

      // send a message to our queue
      sendJmsMessage("second message");

      frame = conn.receiveFrame(100);
      logger.debug("Received frame: {}", frame);
      Assert.assertNull("No message should have been received since subscription was removed", frame);
   }

   @Test
   public void testUnsubscribeWithID() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, "mysubid", Stomp.Headers.Subscribe.AckModeValues.AUTO);

      // send a message to our queue
      sendJmsMessage("first message");

      // receive message from socket
      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());

      // remove suscription
      unsubscribe(conn, "mysubid", null, true, false);

      // send a message to our queue
      sendJmsMessage("second message");

      frame = conn.receiveFrame(100);
      logger.debug("Received frame: {}", frame);
      Assert.assertNull("No message should have been received since subscription was removed", frame);

   }

   @Test
   public void testTransactionCommit() throws Exception {
      MessageConsumer consumer = session.createConsumer(queue);
      conn.connect(defUser, defPass);

      beginTransaction(conn, "tx1");
      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World", true, null, "tx1");

      // check the message is not committed
      assertNull(consumer.receive(100));

      commitTransaction(conn, "tx1", true);

      Message message = consumer.receive(1000);
      Assert.assertNotNull("Should have received a message", message);
   }

   @Test
   public void testSuccessiveTransactionsWithSameID() throws Exception {
      MessageConsumer consumer = session.createConsumer(queue);
      conn.connect(defUser, defPass);

      // first tx
      beginTransaction(conn, "tx1");
      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World", true, null, "tx1");
      commitTransaction(conn, "tx1");

      Message message = consumer.receive(1000);
      Assert.assertNotNull("Should have received a message", message);

      // 2nd tx with same tx ID
      beginTransaction(conn, "tx1");
      send(conn, getQueuePrefix() + getQueueName(), null, "Hello World", true, null, "tx1");
      commitTransaction(conn, "tx1");

      message = consumer.receive(1000);
      Assert.assertNotNull("Should have received a message", message);
   }

   @Test
   public void testBeginSameTransactionTwice() throws Exception {
      conn.connect(defUser, defPass);
      beginTransaction(conn, "tx1");
      beginTransaction(conn, "tx1");

      ClientStompFrame frame = conn.receiveFrame(1000);
      Assert.assertEquals(Stomp.Responses.ERROR, frame.getCommand());
   }

   @Test
   public void testTransactionRollback() throws Exception {
      MessageConsumer consumer = session.createConsumer(queue);
      String txId = "tx1";

      conn.connect(defUser, defPass);
      beginTransaction(conn, txId);
      send(conn, getQueuePrefix() + getQueueName(), null, "first message", true, null, txId);
      abortTransaction(conn, txId);

      beginTransaction(conn, txId);
      send(conn, getQueuePrefix() + getQueueName(), null, "second message", true, null, txId);
      commitTransaction(conn, txId);

      // only second msg should be received since first msg was rolled back
      TextMessage message = (TextMessage) consumer.receive(1000);
      Assert.assertNotNull(message);
      Assert.assertEquals("second message", message.getText());
   }

   @Test
   public void testSubscribeToTopic() throws Exception {
      final int baselineQueueCount = server.getActiveMQServerControl().getQueueNames().length;

      conn.connect(defUser, defPass);

      subscribeTopic(conn, null, null, null, true);

      Wait.assertTrue("Subscription queue should be created here", () -> {
         int length = server.getActiveMQServerControl().getQueueNames().length;
         if (length - baselineQueueCount == 1) {
            return true;
         } else {
            logger.debug("Queue count: {}", (length - baselineQueueCount));
            return false;
         }
      });

      sendJmsMessage(getName(), topic);

      ClientStompFrame frame = conn.receiveFrame(1000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals(getName(), frame.getBody());

      unsubscribe(conn, null, getTopicPrefix() + getTopicName(), true, false);

      sendJmsMessage(getName(), topic);

      frame = conn.receiveFrame(100);
      logger.debug("Received frame: {}", frame);
      Assert.assertNull("No message should have been received since subscription was removed", frame);

      assertEquals("Subscription queue should be deleted", 0, server.getActiveMQServerControl().getQueueNames().length - baselineQueueCount);

      conn.disconnect();
   }

   @Test
   public void testSubscribeToQueue() throws Exception {
      final int baselineQueueCount = server.getActiveMQServerControl().getQueueNames().length;

      conn.connect(defUser, defPass);
      subscribe(conn, null, null, null, true);

      Wait.assertFalse("Queue should not be created here", () -> {
         if (server.getActiveMQServerControl().getQueueNames().length - baselineQueueCount == 1) {
            return true;
         } else {
            return false;
         }
      });

      sendJmsMessage(getName(), queue);

      ClientStompFrame frame = conn.receiveFrame(1000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals(getName(), frame.getBody());

      unsubscribe(conn, null, getQueuePrefix() + getQueueName(), true, false);

      sendJmsMessage(getName(), queue);

      frame = conn.receiveFrame(100);
      logger.debug("Received frame: {}", frame);
      Assert.assertNull("No message should have been received since subscription was removed", frame);

      assertEquals("Subscription queue should not be deleted", baselineQueueCount, server.getActiveMQServerControl().getQueueNames().length);

      conn.disconnect();
   }

   @Test
   public void testSubscribeToNonExistentQueue() throws Exception {
      String nonExistentQueue = RandomUtil.randomString();

      conn.connect(defUser, defPass);
      subscribe(conn, null, null, null, null, getQueuePrefix() + nonExistentQueue, true);

      sendJmsMessage(getName(), ActiveMQJMSClient.createQueue(nonExistentQueue));

      ClientStompFrame frame = conn.receiveFrame(1000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getQueuePrefix() + nonExistentQueue, frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals(getName(), frame.getBody());

      assertNotNull(server.getPostOffice().getBinding(new SimpleString(nonExistentQueue)));

      final Queue subscription = ((LocalQueueBinding) server.getPostOffice().getBinding(new SimpleString(nonExistentQueue))).getQueue();

      Wait.assertEquals(0, subscription::getMessageCount);

      unsubscribe(conn, null, getQueuePrefix() + nonExistentQueue, true, false);

      Wait.assertTrue(() -> server.getPostOffice().getBinding(new SimpleString(nonExistentQueue)) == null);

      sendJmsMessage(getName(), ActiveMQJMSClient.createQueue(nonExistentQueue));

      frame = conn.receiveFrame(100);
      logger.debug("Received frame: {}", frame);
      Assert.assertNull("No message should have been received since subscription was removed", frame);

      conn.disconnect();
   }

   @Test
   public void testDurableSubscriberWithReconnection() throws Exception {
      conn.connect(defUser, defPass, "myclientid");
      subscribeTopic(conn, null, null, getName());

      conn.disconnect();

      Thread.sleep(500);

      // send the message when the durable subscriber is disconnected
      sendJmsMessage(getName(), topic);

      conn.destroy();
      conn = StompClientConnectionFactory.createClientConnection(uri);
      conn.connect(defUser, defPass, "myclientid");

      subscribeTopic(conn, null, null, getName());

      ClientStompFrame frame = conn.receiveFrame(3000);
      assertNotNull("Should have received a message from the durable subscription", frame);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals(getName(), frame.getBody());

      unsubscribe(conn, null, getTopicPrefix() + getTopicName(), true, true);

      conn.disconnect();
   }

   @Test
   public void testDurableSubscriberWithReconnectionLegacy() throws Exception {
      conn.connect(defUser, defPass, "myclientid");
      subscribeTopicLegacyActiveMQ(conn, null, null, getName(), true, false);

      conn.disconnect();

      Thread.sleep(500);

      // send the message when the durable subscriber is disconnected
      sendJmsMessage(getName(), topic);

      conn.destroy();
      conn = StompClientConnectionFactory.createClientConnection(uri);
      conn.connect(defUser, defPass, "myclientid");

      subscribeTopicLegacyActiveMQ(conn, null, null, getName(), true, false);

      ClientStompFrame frame = conn.receiveFrame(3000);
      assertNotNull("Should have received a message from the durable subscription", frame);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals(getName(), frame.getBody());

      unsubscribeLegacyActiveMQ(conn, null, getTopicPrefix() + getTopicName(), true, true);

      conn.disconnect();
   }

   @Test
   public void testDurableSubscriber() throws Exception {
      conn.connect(defUser, defPass, "myclientid");
      subscribeTopic(conn, null, null, getName(), true);
      ClientStompFrame response = subscribeTopic(conn, null, null, getName(), true);

      // creating a subscriber with the same durable-subscriber-name must fail
      Assert.assertEquals(Stomp.Responses.ERROR, response.getCommand());

      conn.disconnect();
   }

   @Test
   public void testDurableSubscriberLegacySubscriptionHeader() throws Exception {
      conn.connect(defUser, defPass, "myclientid");
      subscribeTopicLegacyActiveMQ(conn, null, null, getName(), true, false);
      ClientStompFrame response = subscribeTopicLegacyActiveMQ(conn, null, null, getName(), true, false);

      // creating a subscriber with the same durable-subscriber-name must fail
      Assert.assertEquals(Stomp.Responses.ERROR, response.getCommand());

      conn.disconnect();
   }

   @Test
   public void testDurableUnSubscribe() throws Exception {
      conn.connect(defUser, defPass, "myclientid");
      subscribeTopic(conn, null, null, getName(), true);
      conn.disconnect();
      Thread.sleep(500);

      assertNotNull(server.locateQueue(SimpleString.toSimpleString("myclientid." + getName())));

      conn.destroy();
      conn = StompClientConnectionFactory.createClientConnection(uri);

      conn.connect(defUser, defPass, "myclientid");
      unsubscribe(conn, getName(), getTopicPrefix() + getTopicName(), false, true);
      conn.disconnect();
      Thread.sleep(500);

      Wait.assertTrue(() -> server.locateQueue(SimpleString.toSimpleString("myclientid." + getName())) == null);
   }

   @Test
   public void testDurableUnSubscribeWithoutDurableSubName() throws Exception {
      server.getConfiguration().getWildcardConfiguration().setDelimiter('/');
      server.getAddressSettingsRepository().addMatch("/topic/#", new AddressSettings().setDefaultAddressRoutingType(RoutingType.MULTICAST).setDefaultQueueRoutingType(RoutingType.MULTICAST));
      conn.connect(defUser, defPass, "myclientid");
      String subId = UUID.randomUUID().toString();
      String durableSubName = UUID.randomUUID().toString();
      String receipt = UUID.randomUUID().toString();
      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE).addHeader(Stomp.Headers.Subscribe.DESTINATION, "/topic/test.foo").addHeader(Stomp.Headers.Unsubscribe.ID, subId).addHeader(Stomp.Headers.Subscribe.ACK_MODE, Stomp.Headers.Subscribe.AckModeValues.CLIENT_INDIVIDUAL).addHeader(Stomp.Headers.Subscribe.DURABLE_SUBSCRIPTION_NAME, durableSubName).addHeader(Stomp.Headers.RECEIPT_REQUESTED, receipt);

      frame = conn.sendFrame(frame);
      assertEquals(receipt, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID));

      Wait.assertTrue(() -> server.locateQueue(SimpleString.toSimpleString("myclientid." + durableSubName)) != null);

      receipt = UUID.randomUUID().toString();
      frame = conn.createFrame(Stomp.Commands.UNSUBSCRIBE).addHeader(Stomp.Headers.Unsubscribe.ID, subId).addHeader(Stomp.Headers.RECEIPT_REQUESTED, receipt);

      frame = conn.sendFrame(frame);
      assertEquals(receipt, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID));

      conn.disconnect();

      // make sure the durable subscription queue is still there
      Wait.assertTrue(() -> server.locateQueue(SimpleString.toSimpleString("myclientid." + durableSubName)) != null);
   }

   @Test
   public void testDurableUnSubscribeLegacySubscriptionHeader() throws Exception {
      conn.connect(defUser, defPass, "myclientid");
      subscribeTopicLegacyActiveMQ(conn, null, null, getName(), true, false);
      conn.disconnect();
      Thread.sleep(500);

      assertNotNull(server.locateQueue(SimpleString.toSimpleString("myclientid." + getName())));

      conn.destroy();
      conn = StompClientConnectionFactory.createClientConnection(uri);

      conn.connect(defUser, defPass, "myclientid");
      unsubscribeLegacyActiveMQ(conn, getName(), getTopicPrefix() + getTopicName(), false, true);
      conn.disconnect();
      Thread.sleep(500);

      Wait.assertTrue(() -> server.locateQueue(SimpleString.toSimpleString("myclientid." + getName())) == null);
   }

   @Test
   public void testSubscribeToTopicWithNoLocal() throws Exception {
      conn.connect(defUser, defPass);
      subscribeTopic(conn, null, null, null, true, true);

      // send a message on the same connection => it should not be received is noLocal = true on subscribe
      send(conn, getTopicPrefix() + getTopicName(), null, "Hello World");

      ClientStompFrame frame = conn.receiveFrame(100);
      logger.debug("Received frame: {}", frame);
      Assert.assertNull("No message should have been received since subscription was removed", frame);

      // send message on another JMS connection => it should be received
      sendJmsMessage(getName(), topic);
      frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals(getName(), frame.getBody());

      conn.disconnect();
   }

   @Test
   public void testTopicExistsAfterNoUnsubscribeDisconnect() throws Exception {
      conn.connect(defUser, defPass);
      subscribeTopic(conn, null, null, null, true);

      // disconnect, _without unsubscribing_
      conn.disconnect();

      Thread.sleep(500);

      conn.destroy();

      // connect again
      conn = StompClientConnectionFactory.createClientConnection(uri);
      conn.connect(defUser, defPass);

      // send a receipted message to the topic
      ClientStompFrame response = send(conn, getTopicPrefix() + getTopicName(), null, "Hello World", true);
      assertEquals(Stomp.Responses.RECEIPT, response.getCommand());

      // ...and nothing else
      ClientStompFrame frame = conn.receiveFrame(100);
      logger.debug("Received frame: {}", frame);
      Assert.assertNull(frame);

      conn.disconnect();
   }

   @Test
   public void testClientAckNotPartOfTransaction() throws Exception {
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.CLIENT);

      sendJmsMessage(getName());

      ClientStompFrame frame = conn.receiveFrame(10000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
      String messageID = frame.getHeader(Stomp.Headers.Message.MESSAGE_ID);
      Assert.assertNotNull(messageID);
      Assert.assertEquals(getName(), frame.getBody());

      beginTransaction(conn, "tx1");
      ack(conn, null, messageID, "tx1");
      abortTransaction(conn, "tx1");

      frame = conn.receiveFrame(100);
      Assert.assertNull("No message should have been received as the message was acked even though the transaction has been aborted", frame);

      unsubscribe(conn, null, getQueuePrefix() + getQueueName(), false, false);

      conn.disconnect();
   }

   // HORNETQ-1007
   @Test
   public void testMultiProtocolConsumers() throws Exception {
      final int TIME_OUT = 2000;
      // a timeout for when we expect negative results (like receive==null)
      final int NEGATIVE_TIME_OUT = 100;

      int count = 1000;

      // Create 2 core consumers
      MessageConsumer consumer1 = session.createConsumer(topic);
      MessageConsumer consumer2 = session.createConsumer(topic);

      // connect and subscribe STOMP consumer
      conn.connect(defUser, defPass);
      subscribeTopic(conn, null, null, null, true);

      MessageProducer producer = session.createProducer(topic);
      TextMessage message = session.createTextMessage(getName());

      for (int i = 1; i <= count; i++) {
         producer.send(message);
         Assert.assertNotNull(consumer1.receive(TIME_OUT));
         Assert.assertNotNull(consumer2.receive(TIME_OUT));
         ClientStompFrame frame = conn.receiveFrame(TIME_OUT);
         Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
         Assert.assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
         Assert.assertEquals(getName(), frame.getBody());
      }

      consumer1.close();
      consumer2.close();
      unsubscribe(conn, null, getTopicPrefix() + getTopicName(), true, false);

      sendJmsMessage(getName(), topic);

      ClientStompFrame frame = conn.receiveFrame(NEGATIVE_TIME_OUT);
      logger.debug("Received frame: {}", frame);
      Assert.assertNull("No message should have been received since subscription was removed", frame);

      conn.disconnect();
   }

   @Test
   //stomp should return an ERROR when acking a non-existent message
   public void testUnexpectedAck() throws Exception {
      String messageID = "888888";

      conn.connect(defUser, defPass);
      ack(conn, null, messageID, null);

      ClientStompFrame frame = conn.receiveFrame(1000);
      assertNotNull(frame);
      assertEquals(Stomp.Responses.ERROR, frame.getCommand());

      conn.disconnect();
   }

   @Test
   public void testDotAnycastPrefixOnSend() throws Exception {
      testPrefix("jms.queue.", RoutingType.ANYCAST, true);
   }

   @Test
   public void testDotMulticastPrefixOnSend() throws Exception {
      testPrefix("jms.topic.", RoutingType.MULTICAST, true);
   }

   @Test
   public void testDotAnycastPrefixOnSubscribe() throws Exception {
      testPrefix("jms.queue.", RoutingType.ANYCAST, false);
   }

   @Test
   public void testDotMulticastPrefixOnSubscribe() throws Exception {
      testPrefix("jms.topic.", RoutingType.MULTICAST, false);
   }

   @Test
   public void testSlashAnycastPrefixOnSend() throws Exception {
      testPrefix("/queue/", RoutingType.ANYCAST, true);
   }

   @Test
   public void testSlashMulticastPrefixOnSend() throws Exception {
      testPrefix("/topic/", RoutingType.MULTICAST, true);
   }

   @Test
   public void testSlashAnycastPrefixOnSubscribe() throws Exception {
      testPrefix("/queue/", RoutingType.ANYCAST, false);
   }

   @Test
   public void testSlashMulticastPrefixOnSubscribe() throws Exception {
      testPrefix("/topic/", RoutingType.MULTICAST, false);
   }

   public void testPrefix(final String prefix, final RoutingType routingType, final boolean send) throws Exception {
      int port = 61614;

      URI uri = createStompClientUri(scheme, hostname, port);

      final String ADDRESS = UUID.randomUUID().toString();
      final String PREFIXED_ADDRESS = prefix + ADDRESS;
      String param = routingType.toString();
      String urlParam = param.toLowerCase() + "Prefix";
      server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + port + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME + "&" + urlParam + "=" + prefix).start();
      StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri);
      conn.connect(defUser, defPass);

      // since this queue doesn't exist the broker should create a new address using the routing type matching the prefix
      if (send) {
         send(conn, PREFIXED_ADDRESS, null, "Hello World", true);
      } else {
         String uuid = UUID.randomUUID().toString();

         ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE).addHeader(Stomp.Headers.Subscribe.DESTINATION, PREFIXED_ADDRESS).addHeader(Stomp.Headers.RECEIPT_REQUESTED, uuid);

         frame = conn.sendFrame(frame);

         assertEquals(uuid, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID));
      }

      AddressInfo addressInfo = server.getAddressInfo(SimpleString.toSimpleString(ADDRESS));
      assertNotNull("No address was created with the name " + ADDRESS, addressInfo);

      Set<RoutingType> routingTypes = new HashSet<>();
      routingTypes.add(RoutingType.valueOf(param));
      assertEquals(routingTypes, addressInfo.getRoutingTypes());

      conn.disconnect();
   }

   /**
    * This test and testPrefixedAutoCreatedMulticastAndAnycastWithSameName are basically the same but doing the
    * operations in opposite order. In this test the anycast subscription is created first.
    *
    * @throws Exception
    */
   @Test
   public void testPrefixedAutoCreatedAnycastAndMulticastWithSameName() throws Exception {
      int port = 61614;

      URI uri = createStompClientUri(scheme, hostname, port);

      final String ADDRESS = UUID.randomUUID().toString();
      server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + port + "?protocols=STOMP&anycastPrefix=/queue/&multicastPrefix=/topic/").start();
      StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri);
      conn.connect(defUser, defPass);

      // since this queue doesn't exist the broker should create a new ANYCAST address & queue
      String uuid = UUID.randomUUID().toString();
      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE).addHeader(Stomp.Headers.Subscribe.DESTINATION, "/queue/" + ADDRESS).addHeader(Stomp.Headers.RECEIPT_REQUESTED, uuid);
      frame = conn.sendFrame(frame);
      assertEquals(uuid, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID));

      AddressInfo addressInfo = server.getAddressInfo(SimpleString.toSimpleString(ADDRESS));
      assertNotNull("No address was created with the name " + ADDRESS, addressInfo);
      assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.ANYCAST));
      assertFalse(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST));
      assertNotNull(server.locateQueue(SimpleString.toSimpleString(ADDRESS)));

      // sending a MULTICAST message should alter the address to support MULTICAST
      frame = send(conn, "/topic/" + ADDRESS, null, "Hello World 1", true);
      assertFalse(frame.getCommand().equals("ERROR"));
      addressInfo = server.getAddressInfo(SimpleString.toSimpleString(ADDRESS));
      assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.ANYCAST));
      assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST));

      // however, no message should be routed to the ANYCAST queue
      frame = conn.receiveFrame(100);
      Assert.assertNull(frame);

      // sending a message to the ANYCAST queue, should be received
      frame = send(conn, "/queue/" + ADDRESS, null, "Hello World 2", true);
      assertFalse(frame.getCommand().equals("ERROR"));
      frame = conn.receiveFrame(1000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals("Hello World 2", frame.getBody());
      Assert.assertEquals(RoutingType.ANYCAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
      Assert.assertEquals("/queue/" + ADDRESS, frame.getHeader(Stomp.Headers.Send.DESTINATION));
      frame = conn.receiveFrame(100);
      Assert.assertNull(frame);

      unsubscribe(conn, null, "/queue/" + ADDRESS, true, false);

      Wait.assertTrue(() -> server.getAddressInfo(SimpleString.toSimpleString(ADDRESS)) == null);

      // now subscribe to the address in a MULTICAST way which will create a MULTICAST queue for the subscription
      uuid = UUID.randomUUID().toString();
      frame = conn.createFrame(Stomp.Commands.SUBSCRIBE).addHeader(Stomp.Headers.Subscribe.DESTINATION, "/topic/" + ADDRESS).addHeader(Stomp.Headers.RECEIPT_REQUESTED, uuid);
      frame = conn.sendFrame(frame);
      assertEquals(uuid, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID));

      // send a message which will be routed to the MULTICAST queue
      frame = send(conn, "/topic/" + ADDRESS, null, "Hello World 3", true);
      assertFalse(frame.getCommand().equals("ERROR"));

      // receive that message on the topic subscription
      frame = conn.receiveFrame(1000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals("Hello World 3", frame.getBody());
      Assert.assertEquals(RoutingType.MULTICAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
      Assert.assertEquals("/topic/" + ADDRESS, frame.getHeader(Stomp.Headers.Send.DESTINATION));
      frame = conn.receiveFrame(100);
      Assert.assertNull(frame);

      unsubscribe(conn, null, "/topic/" + ADDRESS, true, false);

      conn.disconnect();
   }

   /**
    * This test and testPrefixedAutoCreatedMulticastAndAnycastWithSameName are basically the same but doing the
    * operations in opposite order. In this test the multicast subscription is created first.
    *
    * @throws Exception
    */
   @Test
   public void testPrefixedAutoCreatedMulticastAndAnycastWithSameName() throws Exception {
      int port = 61614;

      URI uri = createStompClientUri(scheme, hostname, port);

      final String ADDRESS = UUID.randomUUID().toString();
      server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + port + "?protocols=STOMP&anycastPrefix=/queue/&multicastPrefix=/topic/").start();
      StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri);
      conn.connect(defUser, defPass);

      // since this queue doesn't exist the broker should create a new MULTICAST address
      String uuid = UUID.randomUUID().toString();
      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE).addHeader(Stomp.Headers.Subscribe.DESTINATION, "/topic/" + ADDRESS).addHeader(Stomp.Headers.RECEIPT_REQUESTED, uuid);
      frame = conn.sendFrame(frame);
      assertEquals(uuid, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID));

      AddressInfo addressInfo = server.getAddressInfo(SimpleString.toSimpleString(ADDRESS));
      assertNotNull("No address was created with the name " + ADDRESS, addressInfo);
      assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST));
      assertFalse(addressInfo.getRoutingTypes().contains(RoutingType.ANYCAST));

      // sending an ANYCAST message should alter the address to support ANYCAST and create an ANYCAST queue
      frame = send(conn, "/queue/" + ADDRESS, null, "Hello World 1", true);
      assertFalse(frame.getCommand().equals("ERROR"));
      addressInfo = server.getAddressInfo(SimpleString.toSimpleString(ADDRESS));
      assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.ANYCAST));
      assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST));
      assertNotNull(server.locateQueue(SimpleString.toSimpleString(ADDRESS)));

      // however, no message should be routed to the MULTICAST queue
      frame = conn.receiveFrame(100);
      Assert.assertNull(frame);

      // sending a message to the MULTICAST queue, should be received
      frame = send(conn, "/topic/" + ADDRESS, null, "Hello World 2", true);
      assertFalse(frame.getCommand().equals("ERROR"));
      frame = conn.receiveFrame(2000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals("Hello World 2", frame.getBody());
      Assert.assertEquals(RoutingType.MULTICAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
      Assert.assertEquals("/topic/" + ADDRESS, frame.getHeader(Stomp.Headers.Send.DESTINATION));
      frame = conn.receiveFrame(100);
      Assert.assertNull(frame);

      frame = unsubscribe(conn, null, "/topic/" + ADDRESS, true, false);
      assertFalse(frame.getCommand().equals("ERROR"));

      // now subscribe to the address in an ANYCAST way
      uuid = UUID.randomUUID().toString();
      frame = conn.createFrame(Stomp.Commands.SUBSCRIBE).addHeader(Stomp.Headers.Subscribe.DESTINATION, "/queue/" + ADDRESS).addHeader(Stomp.Headers.RECEIPT_REQUESTED, uuid);
      frame = conn.sendFrame(frame);
      assertEquals(uuid, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID));

      // receive that message on the ANYCAST queue
      frame = conn.receiveFrame(1000);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals("Hello World 1", frame.getBody());
      Assert.assertEquals(RoutingType.ANYCAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
      Assert.assertEquals("/queue/" + ADDRESS, frame.getHeader(Stomp.Headers.Send.DESTINATION));
      frame = conn.receiveFrame(100);
      Assert.assertNull(frame);

      unsubscribe(conn, null, "/queue/" + ADDRESS, true, false);

      conn.disconnect();
   }

   @Test
   public void testDotPrefixedSendAndRecieveAnycast() throws Exception {
      testPrefixedSendAndRecieve("jms.queue.", RoutingType.ANYCAST);
   }

   @Test
   public void testDotPrefixedSendAndRecieveMulticast() throws Exception {
      testPrefixedSendAndRecieve("jms.topic.", RoutingType.MULTICAST);
   }

   @Test
   public void testSlashPrefixedSendAndRecieveAnycast() throws Exception {
      testPrefixedSendAndRecieve("/queue/", RoutingType.ANYCAST);
   }

   @Test
   public void testSlashPrefixedSendAndRecieveMulticast() throws Exception {
      testPrefixedSendAndRecieve("/topic/", RoutingType.MULTICAST);
   }

   public void testPrefixedSendAndRecieve(final String prefix, RoutingType routingType) throws Exception {
      int port = 61614;

      URI uri = createStompClientUri(scheme, hostname, port);

      final String ADDRESS = UUID.randomUUID().toString();
      final String PREFIXED_ADDRESS = prefix + ADDRESS;
      String urlParam = routingType.toString().toLowerCase() + "Prefix";
      server.getRemotingService().createAcceptor("test", "tcp://" + hostname + ":" + port + "?protocols=" + StompProtocolManagerFactory.STOMP_PROTOCOL_NAME + "&" + urlParam + "=" + prefix).start();
      StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri);
      conn.connect(defUser, defPass);
      String uuid = UUID.randomUUID().toString();

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE).addHeader(Stomp.Headers.Subscribe.DESTINATION, PREFIXED_ADDRESS).addHeader(Stomp.Headers.RECEIPT_REQUESTED, uuid);

      frame = conn.sendFrame(frame);
      assertEquals(uuid, frame.getHeader(Stomp.Headers.Response.RECEIPT_ID));

      send(conn, ADDRESS, null, "Hello World", true);

      frame = conn.receiveFrame(10000);
      Assert.assertNotNull("Should have received a message", frame);
      Assert.assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
      Assert.assertEquals(ADDRESS, frame.getHeader(Stomp.Headers.Send.DESTINATION));
      Assert.assertEquals("Hello World", frame.getBody());

      conn.disconnect();
   }

   @Test
   public void testMulticastOperationsOnAnycastAddress() throws Exception {
      server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoCreateAddresses(false).setAutoCreateQueues(false));
      testRoutingSemantics(RoutingType.MULTICAST.toString(), getQueuePrefix() + getQueueName());
   }

   @Test
   public void testAnycastOperationsOnMulticastAddress() throws Exception {
      server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoCreateAddresses(false).setAutoCreateQueues(false));
      testRoutingSemantics(RoutingType.ANYCAST.toString(), getTopicPrefix() + getTopicName());
   }

   public void testRoutingSemantics(String routingType, String destination) throws Exception {
      conn.connect(defUser, defPass);

      String uuid = UUID.randomUUID().toString();

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SUBSCRIBE).addHeader(Stomp.Headers.Subscribe.SUBSCRIPTION_TYPE, routingType).addHeader(Stomp.Headers.Subscribe.DESTINATION, destination).addHeader(Stomp.Headers.RECEIPT_REQUESTED, uuid);

      frame = conn.sendFrame(frame);
      assertEquals(Stomp.Responses.ERROR, frame.getCommand());

      uuid = UUID.randomUUID().toString();

      frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION_TYPE, RoutingType.MULTICAST.toString()).addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()).addHeader(Stomp.Headers.RECEIPT_REQUESTED, uuid);

      frame = conn.sendFrame(frame);
      assertEquals(Stomp.Responses.ERROR, frame.getCommand());
   }

   @Test
   public void testGetManagementAttributeFromStomp() throws Exception {
      server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoCreateAddresses(false).setAutoCreateQueues(false));
      conn.connect(defUser, defPass);

      subscribe(conn, null);

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, ActiveMQDefaultConfiguration.getDefaultManagementAddress().toString()).addHeader(Stomp.Headers.Send.REPLY_TO, getQueuePrefix() + getQueueName()).addHeader(ManagementHelper.HDR_RESOURCE_NAME.toString(), ResourceNames.QUEUE + getQueuePrefix() + getQueueName()).addHeader(ManagementHelper.HDR_ATTRIBUTE.toString(), "Address");

      conn.sendFrame(frame);

      frame = conn.receiveFrame(10000);

      logger.debug("Received: {}", frame);

      Assert.assertEquals(Boolean.TRUE.toString(), frame.getHeader(ManagementHelper.HDR_OPERATION_SUCCEEDED.toString()));
      // the address will be returned in the message body in a JSON array
      Assert.assertEquals("[\"" + getQueuePrefix() + getQueueName() + "\"]", frame.getBody());

      unsubscribe(conn, null);

      conn.disconnect();
   }

   @Test
   public void testInvokeOperationFromStomp() throws Exception {
      conn.connect(defUser, defPass);

      subscribe(conn, null);

      ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, ActiveMQDefaultConfiguration.getDefaultManagementAddress().toString()).addHeader(Stomp.Headers.Send.REPLY_TO, getQueuePrefix() + getQueueName()).addHeader(ManagementHelper.HDR_RESOURCE_NAME.toString(), ResourceNames.QUEUE + getQueuePrefix() + getQueueName()).addHeader(ManagementHelper.HDR_OPERATION_NAME.toString(), "countMessages").setBody("[\"color = 'blue'\"]");

      conn.sendFrame(frame);

      frame = conn.receiveFrame(10000);

      logger.debug("Received: {}", frame);

      Assert.assertEquals(Boolean.TRUE.toString(), frame.getHeader(ManagementHelper.HDR_OPERATION_SUCCEEDED.toString()));
      // there is no such messages => 0 returned in a JSON array
      Assert.assertEquals("[0]", frame.getBody());

      unsubscribe(conn, null);

      conn.disconnect();
   }

   @Test
   public void testAnycastMessageRoutingExclusivity() throws Exception {
      conn.connect(defUser, defPass);

      final String addressA = "addressA";
      final String queueA = "queueA";
      final String queueB = "queueB";
      final String queueC = "queueC";

      ActiveMQServerControl serverControl = server.getActiveMQServerControl();
      serverControl.createAddress(addressA, RoutingType.ANYCAST.toString() + "," + RoutingType.MULTICAST.toString());
      serverControl.createQueue(new QueueConfiguration(queueA).setAddress(addressA).setRoutingType(RoutingType.ANYCAST).toJSON());
      serverControl.createQueue(new QueueConfiguration(queueB).setAddress(addressA).setRoutingType(RoutingType.ANYCAST).toJSON());
      serverControl.createQueue(new QueueConfiguration(queueC).setAddress(addressA).setRoutingType(RoutingType.MULTICAST).toJSON());

      send(conn, addressA, null, "Hello World!", true, RoutingType.ANYCAST);

      Wait.assertTrue(() -> server.locateQueue(SimpleString.toSimpleString(queueA)).getMessageCount() + server.locateQueue(SimpleString.toSimpleString(queueB)).getMessageCount() == 1);
      Wait.assertTrue(() -> server.locateQueue(SimpleString.toSimpleString(queueC)).getMessageCount() == 0);
   }

   @Test
   public void testMulticastMessageRoutingExclusivity() throws Exception {
      conn.connect(defUser, defPass);

      final String addressA = "addressA";
      final String queueA = "queueA";
      final String queueB = "queueB";
      final String queueC = "queueC";

      ActiveMQServerControl serverControl = server.getActiveMQServerControl();
      serverControl.createAddress(addressA, RoutingType.ANYCAST.toString() + "," + RoutingType.MULTICAST.toString());
      serverControl.createQueue(new QueueConfiguration(queueA).setAddress(addressA).setRoutingType(RoutingType.ANYCAST).toJSON());
      serverControl.createQueue(new QueueConfiguration(queueB).setAddress(addressA).setRoutingType(RoutingType.MULTICAST).toJSON());
      serverControl.createQueue(new QueueConfiguration(queueC).setAddress(addressA).setRoutingType(RoutingType.MULTICAST).toJSON());

      send(conn, addressA, null, "Hello World!", true, RoutingType.MULTICAST);

      Wait.assertTrue(() -> server.locateQueue(SimpleString.toSimpleString(queueA)).getMessageCount() == 0);
      Wait.assertTrue(() -> server.locateQueue(SimpleString.toSimpleString(queueC)).getMessageCount() + server.locateQueue(SimpleString.toSimpleString(queueB)).getMessageCount() == 2);
   }

   @Test
   public void testAmbiguousMessageRouting() throws Exception {
      conn.connect(defUser, defPass);

      final String addressA = "addressA";
      final String queueA = "queueA";
      final String queueB = "queueB";
      final String queueC = "queueC";
      final String queueD = "queueD";

      ActiveMQServerControl serverControl = server.getActiveMQServerControl();
      serverControl.createAddress(addressA, RoutingType.ANYCAST.toString() + "," + RoutingType.MULTICAST.toString());
      serverControl.createQueue(new QueueConfiguration(queueA).setAddress(addressA).setRoutingType(RoutingType.ANYCAST).toJSON());
      serverControl.createQueue(new QueueConfiguration(queueB).setAddress(addressA).setRoutingType(RoutingType.ANYCAST).toJSON());
      serverControl.createQueue(new QueueConfiguration(queueC).setAddress(addressA).setRoutingType(RoutingType.MULTICAST).toJSON());
      serverControl.createQueue(new QueueConfiguration(queueD).setAddress(addressA).setRoutingType(RoutingType.MULTICAST).toJSON());

      send(conn, addressA, null, "Hello World!", true);

      Wait.assertTrue(() -> server.locateQueue(SimpleString.toSimpleString(queueA)).getMessageCount() + server.locateQueue(SimpleString.toSimpleString(queueB)).getMessageCount() == 1);
      Wait.assertTrue(() -> server.locateQueue(SimpleString.toSimpleString(queueC)).getMessageCount() + server.locateQueue(SimpleString.toSimpleString(queueD)).getMessageCount() == 2);
   }

   @Test
   public void testAutoCreatedAnycastAddress() throws Exception {
      conn.connect(defUser, defPass);

      String queueName = UUID.randomUUID().toString();
      SimpleString simpleQueueName = SimpleString.toSimpleString(queueName);

      Assert.assertNull(server.getAddressInfo(simpleQueueName));
      Assert.assertNull(server.locateQueue(simpleQueueName));

      server.getAddressSettingsRepository().addMatch(queueName, new AddressSettings().setDefaultAddressRoutingType(RoutingType.ANYCAST).setDefaultQueueRoutingType(RoutingType.ANYCAST));

      send(conn, queueName, null, "Hello ANYCAST");

      Wait.assertTrue("Address and queue should be created now", () -> (server.getAddressInfo(simpleQueueName) != null) && (server.locateQueue(simpleQueueName) != null));
      assertTrue(server.getAddressInfo(simpleQueueName).getRoutingTypes().contains(RoutingType.ANYCAST));
      assertEquals(RoutingType.ANYCAST, server.locateQueue(simpleQueueName).getRoutingType());
   }

   @Test
   public void testAutoCreatedMulticastAddress() throws Exception {
      conn.connect(defUser, defPass);

      String queueName = UUID.randomUUID().toString();
      SimpleString simpleQueueName = SimpleString.toSimpleString(queueName);

      Assert.assertNull(server.getAddressInfo(simpleQueueName));
      Assert.assertNull(server.locateQueue(simpleQueueName));

      send(conn, queueName, null, "Hello MULTICAST");

      Wait.assertTrue("Address should be created now", () -> (server.getAddressInfo(simpleQueueName) != null));
      assertTrue(server.getAddressInfo(simpleQueueName).getRoutingTypes().contains(RoutingType.MULTICAST));
      Assert.assertNull(server.locateQueue(simpleQueueName));
   }

   @Test
   public void directDeliverDisabledOnStomp() throws Exception {
      String payload = "This is a test message";

      // Set up STOMP subscription
      conn.connect(defUser, defPass);
      subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO);

      for (Binding b : iterableOf(server.getPostOffice().getAllBindings().filter(QueueBinding.class::isInstance))) {
         Assert.assertFalse("Queue " + ((QueueBinding) b).getQueue().getName(), ((QueueBinding) b).getQueue().isDirectDeliver());
      }

      // Send MQTT Message
      MQTTClientProvider clientProvider = new FuseMQTTClientProvider();
      clientProvider.connect("tcp://" + hostname + ":" + port);
      clientProvider.publish(getQueuePrefix() + getQueueName(), payload.getBytes(), 0);
      clientProvider.disconnect();

      // Receive STOMP Message
      ClientStompFrame frame = conn.receiveFrame();
      assertTrue(frame.getBody().contains(payload));

   }

}