content: /*
   * 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.MessageProducer;
  import javax.jms.TextMessage;
  import java.io.ByteArrayOutputStream;
  import java.lang.invoke.MethodHandles;
  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.protocol.stomp.StompSession;
  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.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.jupiter.api.AfterEach;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static org.apache.activemq.artemis.utils.collections.IterableStream.iterableOf;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertNotEquals;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.junit.jupiter.api.Assertions.fail;
  
  public class StompTest extends StompTestBase {
  
     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  
     protected StompClientConnection conn;
  
     public StompTest() {
        super("tcp+v10.stomp");
     }
  
     @Override
     protected ActiveMQServer createServer() throws Exception {
        ActiveMQServer server = super.createServer();
        server.getConfiguration().setAddressQueueScanPeriod(100);
        return server;
     }
  
     @Override
     @BeforeEach
     public void setUp() throws Exception {
        super.setUp();
        conn = StompClientConnectionFactory.createClientConnection(uri);
     }
  
     @Override
     @AfterEach
     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(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");
  
        assertEquals(1, array.size(), "number of producers returned from query");
  
        JsonObject producer = array.getJsonObject(0);
        assertEquals(1, producer.getInt(ProducerField.MESSAGE_SENT.getName()), 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");
  
        assertEquals(1, array.size(), "number of producers returned from query");
  
        producer = array.getJsonObject(0);
        assertEquals(2, producer.getInt(ProducerField.MESSAGE_SENT.getName()), 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");
  
        assertEquals(0, array.size(), "number of producers returned from query");
     }
  
     @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(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);
  
        assertEquals(Stomp.Responses.CONNECTED, response.getCommand());
        assertEquals("1", response.getHeader(Stomp.Headers.Connected.RESPONSE_ID));
     }
  
     @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);
  
        assertEquals(Stomp.Responses.CONNECTED, response.getCommand());
        assertEquals("1", response.getHeader(Stomp.Headers.Connected.RESPONSE_ID));
  
        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);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
        // Assert default priority 4 is used when priority header is not set
        assertEquals(4, message.getJMSPriority(), "getJMSPriority");
  
        // Make sure that the timestamp is valid - should
        // be very close to the current time.
        long tnow = System.currentTimeMillis();
        long tmsg = message.getJMSTimestamp();
        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);
        assertNotNull(message);
        assertFalse(((ActiveMQMessage)message).getCoreMessage().getPropertyNames().contains(MessageUtil.CORRELATIONID_HEADER_NAME));
        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);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
        // Assert default priority 4 is used when priority header is not set
        assertEquals(4, message.getJMSPriority(), "getJMSPriority");
  
        // Make sure that the timestamp is valid - should
        // be very close to the current time.
        long tnow = System.currentTimeMillis();
        long tmsg = message.getJMSTimestamp();
        assertTrue(Math.abs(tnow - tmsg) < 1500);
  
        // closing the consumer here should trigger auto-deletion
        assertNotNull(server.getPostOffice().getBinding(SimpleString.of(queue)));
        consumer.close();
        Wait.assertTrue(() -> server.getPostOffice().getBinding(SimpleString.of(queue)) == null);
     }
  
     @Test
     public void testSendMessageToNonExistentQueue() throws Exception {
        sendMessageToNonExistentQueue(getQueuePrefix(), RandomUtil.randomUUIDString(), RoutingType.ANYCAST);
     }
  
     @Test
     public void testSendMessageToNonExistentQueueUsingExplicitDefaultRouting() throws Exception {
        String nonExistentQueue = RandomUtil.randomUUIDString();
        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);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
        // Assert default priority 4 is used when priority header is not set
        assertEquals(4, message.getJMSPriority(), "getJMSPriority");
  
        // Make sure that the timestamp is valid - should
        // be very close to the current time.
        long tnow = System.currentTimeMillis();
        long tmsg = message.getJMSTimestamp();
        assertTrue(Math.abs(tnow - tmsg) < 1500);
  
        assertNotNull(server.getAddressInfo(SimpleString.of(topic)));
  
        // closing the consumer here should trigger auto-deletion of the subscription queue and address
        consumer.close();
        Wait.assertTrue(() -> server.getAddressInfo(SimpleString.of(topic)) == null);
     }
  
     @Test
     public void testSendMessageToNonExistentTopic() throws Exception {
        sendMessageToNonExistentTopic(getTopicPrefix(), RandomUtil.randomUUIDString(), RoutingType.MULTICAST);
     }
  
     @Test
     public void testSendMessageToNonExistentTopicUsingExplicitDefaultRouting() throws Exception {
        String nonExistentTopic = RandomUtil.randomUUIDString();
        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.randomUUIDString(), 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);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
  
        send(conn, getQueuePrefix() + getQueueName(), null, "Hello World");
        message = (TextMessage) consumer.receive(1000);
        assertNotNull(message);
        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();
        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);
        assertNotNull(message);
        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();
        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);
        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);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
        // differ from StompConnect
        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);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
        assertEquals("abc", message.getStringProperty("foo"), "foo");
        assertEquals("123", message.getStringProperty("bar"), "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);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
        assertEquals("abc", message.getStringProperty("foo"), "foo");
        assertEquals("123", message.getStringProperty("b-ar"), "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);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
        assertEquals("c123", message.getJMSCorrelationID(), "JMSCorrelationID");
        assertEquals("t345", message.getJMSType(), "getJMSType");
        assertEquals(3, message.getJMSPriority(), "getJMSPriority");
        assertEquals(javax.jms.DeliveryMode.PERSISTENT, message.getJMSDeliveryMode());
        assertEquals("abc", message.getStringProperty("foo"), "foo");
        assertEquals("123", message.getStringProperty("bar"), "bar");
  
        assertEquals("abc", message.getStringProperty("JMSXGroupID"), "JMSXGroupID");
        // FIXME do we support it?
        // Assert.assertEquals("GroupID", "abc", amqMessage.getGroupID());
     }
  
     @Test
     public void testSendMessageWithLongHeaders() throws Exception {
        final int headerSize = 1024;
        MessageConsumer consumer = session.createConsumer(queue);
  
        conn.connect(defUser, defPass);
  
        String longHeader = "a".repeat(headerSize);
  
        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", longHeader)
           .setBody("Hello World");
        conn.sendFrame(frame);
  
        TextMessage message = (TextMessage) consumer.receive(1000);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
        assertEquals("c123", message.getJMSCorrelationID(), "JMSCorrelationID");
        assertEquals("t345", message.getJMSType(), "getJMSType");
        assertEquals(3, message.getJMSPriority(), "getJMSPriority");
        assertEquals(javax.jms.DeliveryMode.PERSISTENT, message.getJMSDeliveryMode());
        assertEquals("abc", message.getStringProperty("foo"), "foo");
        assertEquals(headerSize, message.getStringProperty("longHeader").length(), "longHeader");
  
        assertEquals("abc", message.getStringProperty("JMSXGroupID"), "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(consumer.receive(100), "Should not receive message yet");
  
        TextMessage message = (TextMessage) consumer.receive(4000);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
        assertEquals("c123", message.getJMSCorrelationID(), "JMSCorrelationID");
        assertEquals("t345", message.getJMSType(), "getJMSType");
        assertEquals(3, message.getJMSPriority(), "getJMSPriority");
        assertEquals(javax.jms.DeliveryMode.PERSISTENT, message.getJMSDeliveryMode());
        assertEquals("abc", message.getStringProperty("foo"), "foo");
        assertEquals("abc", message.getStringProperty("JMSXGroupID"), "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(consumer.receive(100), "Should not receive message yet");
  
        TextMessage message = (TextMessage) consumer.receive(4000);
        assertNotNull(message);
        assertEquals("Hello World", message.getText());
        assertEquals("c123", message.getJMSCorrelationID(), "JMSCorrelationID");
        assertEquals("t345", message.getJMSType(), "getJMSType");
        assertEquals(3, message.getJMSPriority(), "getJMSPriority");
        assertEquals(javax.jms.DeliveryMode.PERSISTENT, message.getJMSDeliveryMode());
        assertEquals("abc", message.getStringProperty("foo"), "foo");
        assertEquals("abc", message.getStringProperty("JMSXGroupID"), "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(consumer.receive(100), "Should not receive message yet");
  
        ClientStompFrame error = conn.receiveFrame();
  
        assertNotNull(error);
        assertEquals("ERROR", error.getCommand());
     }
  
     @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(consumer.receive(100), "Should not receive message yet");
  
        ClientStompFrame error = conn.receiveFrame();
  
        assertNotNull(error);
        assertEquals("ERROR", error.getCommand());
     }
  
     @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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        assertEquals(getName(), frame.getBody());
  
        Pattern cl = Pattern.compile(Stomp.Headers.CONTENT_LENGTH + ":\\s*(\\d+)", Pattern.CASE_INSENSITIVE);
        Matcher cl_matcher = cl.matcher(frame.toString());
        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);
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        assertEquals(getName(), frame.getBody());
        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);
  
        StompProtocolManager stompProtocolManager = getStompProtocolManager();
  
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        String ingressTimestampHeader = frame.getHeader(Stomp.Headers.Message.INGRESS_TIMESTAMP);
        assertNotNull(ingressTimestampHeader);
        long ingressTimestamp = Long.parseLong(ingressTimestampHeader);
        assertTrue(ingressTimestamp >= beforeSend && ingressTimestamp <= afterSend, "Ingress timstamp " + ingressTimestamp + " should be >= " + beforeSend + " and <= " + 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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        assertEquals(RoutingType.ANYCAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
        assertNull(frame.getHeader(org.apache.activemq.artemis.api.core.Message.HDR_ROUTING_TYPE.toString()));
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        assertEquals(RoutingType.MULTICAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
        assertNull(frame.getHeader(org.apache.activemq.artemis.api.core.Message.HDR_ROUTING_TYPE.toString()));
        assertEquals(getName(), frame.getBody());
  
        conn.disconnect();
     }
  
     @Test
     public void testSubscriptionQueueCreatedWhenAutoCreateDisabled() throws Exception {
        SimpleString topic = SimpleString.of(getTopicPrefix() + getTopicName());
        server.getAddressSettingsRepository().getMatch(topic.toString()).setAutoCreateQueues(false);
        conn.connect(defUser, defPass);
  
        assertEquals(0, server.getPostOffice().getBindingsForAddress(topic).size());
        subscribeTopic(conn, null, null, null, true);
        Wait.assertEquals(1, () -> server.getPostOffice().getBindingsForAddress(topic).size(), 2000, 100);
  
        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);
  
        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());
        assertTrue(cl_matcher.find());
        assertEquals("5", cl_matcher.group(1));
  
        assertFalse(Pattern.compile("type:\\s*null", Pattern.CASE_INSENSITIVE).matcher(frame.toString()).find());
        assertTrue(frame.getBody().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);
        assertNotNull(frame);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals("value", frame.getHeader("S"));
        assertEquals("false", frame.getHeader("n"));
        assertEquals("9", frame.getHeader("byte"));
        assertEquals("2.0", frame.getHeader("d"));
        assertEquals("6.0", frame.getHeader("f"));
        assertEquals("10", frame.getHeader("i"));
        assertEquals("121", frame.getHeader("l"));
        assertEquals("12", frame.getHeader("s"));
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Message.DESTINATION));
        assertEquals("mysubid", frame.getHeader(Stomp.Headers.Message.SUBSCRIPTION));
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Message.DESTINATION));
        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);
           assertEquals(data[i], frame.getBody(), "Message not in order");
        }
  
        // 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);
           assertEquals(data[i], frame.getBody(), "Message not in order");
        }
  
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals("Real message", frame.getBody(), "Should have received the real message but got: " + frame);
  
        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.of("Ignored message"));
  
        ClientMessage realMessage = clientSession.createMessage(false);
        realMessage.putStringProperty("foo-bar", "zzz");
        realMessage.getBodyBuffer().writeNullableSimpleString(SimpleString.of("Real message"));
  
        producer.send(ignoredMessage);
        producer.send(realMessage);
  
        ClientStompFrame frame = conn.receiveFrame(10000);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals("Real message", frame.getBody(), "Should have received the real message but got: " + frame);
  
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals("<root><a key='first' num='1'/><b key='second' num='2'>b</b></root>", frame.getBody(), "Should have received the real message but got: " + frame);
  
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        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);
        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);
        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);
        assertNotNull(message);
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
  
        logger.debug("Reconnecting!");
  
        if (sendDisconnect) {
           conn.disconnect();
        }
        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);
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        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);
        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);
        assertNull(frame, "No message should have been received since subscription was removed");
     }
  
     @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);
        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);
        assertNull(frame, "No message should have been received since subscription was removed");
  
     }
  
     @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);
        assertNotNull(message, "Should have received a 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);
        assertNotNull(message, "Should have received a 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);
        assertNotNull(message, "Should have received a message");
     }
  
     @Test
     public void testBeginSameTransactionTwice() throws Exception {
        conn.connect(defUser, defPass);
        beginTransaction(conn, "tx1");
        beginTransaction(conn, "tx1");
  
        ClientStompFrame frame = conn.receiveFrame(1000);
        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);
        assertNotNull(message);
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        assertEquals(getName(), frame.getBody());
  
        unsubscribe(conn, null, getTopicPrefix() + getTopicName(), true, false);
  
        sendJmsMessage(getName(), topic);
  
        frame = conn.receiveFrame(100);
        logger.debug("Received frame: {}", frame);
        assertNull(frame, "No message should have been received since subscription was removed");
  
        assertEquals(0, server.getActiveMQServerControl().getQueueNames().length - baselineQueueCount, "Subscription queue should be deleted");
  
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        assertEquals(getName(), frame.getBody());
  
        unsubscribe(conn, null, getQueuePrefix() + getQueueName(), true, false);
  
        sendJmsMessage(getName(), queue);
  
        frame = conn.receiveFrame(100);
        logger.debug("Received frame: {}", frame);
        assertNull(frame, "No message should have been received since subscription was removed");
  
        assertEquals(baselineQueueCount, server.getActiveMQServerControl().getQueueNames().length, "Subscription queue should not be deleted");
  
        conn.disconnect();
     }
  
     @Test
     public void testSubscribeToNonExistentQueue() throws Exception {
        String nonExistentQueue = RandomUtil.randomUUIDString();
  
        conn.connect(defUser, defPass);
        subscribe(conn, null, null, null, null, getQueuePrefix() + nonExistentQueue, true);
  
        sendJmsMessage(getName(), ActiveMQJMSClient.createQueue(nonExistentQueue));
  
        ClientStompFrame frame = conn.receiveFrame(1000);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getQueuePrefix() + nonExistentQueue, frame.getHeader(Stomp.Headers.Send.DESTINATION));
        assertEquals(getName(), frame.getBody());
  
        assertNotNull(server.getPostOffice().getBinding(SimpleString.of(nonExistentQueue)));
  
        final Queue subscription = ((LocalQueueBinding) server.getPostOffice().getBinding(SimpleString.of(nonExistentQueue))).getQueue();
  
        Wait.assertEquals(0, subscription::getMessageCount);
  
        unsubscribe(conn, null, getQueuePrefix() + nonExistentQueue, true, false);
  
        Wait.assertTrue(() -> server.getPostOffice().getBinding(SimpleString.of(nonExistentQueue)) == null);
  
        sendJmsMessage(getName(), ActiveMQJMSClient.createQueue(nonExistentQueue));
  
        frame = conn.receiveFrame(100);
        logger.debug("Received frame: {}", frame);
        assertNull(frame, "No message should have been received since subscription was removed");
  
        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(frame, "Should have received a message from the durable subscription");
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        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(frame, "Should have received a message from the durable subscription");
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        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
        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
        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.of("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.of("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.of("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.of("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.of("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.of("myclientid." + getName())) == null);
     }
  
     @Test
     public void testSubscribeToTopicWithNoLocalSendWithJMS() 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);
        assertNull(frame, "No message should have been received since subscription was removed");
  
        // send message on another JMS connection => it should be received
        // a JMS message automatically gets a property named __AMQ_CID
        sendJmsMessage(getName(), topic);
        frame = conn.receiveFrame(10000);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        assertEquals(getName(), frame.getBody());
  
        conn.disconnect();
     }
  
     @Test
     public void testSubscribeToTopicWithNoLocalSendWithStomp() throws Exception {
        conn.connect(defUser, defPass);
        StompClientConnection conn2 = StompClientConnectionFactory.createClientConnection(uri);
        conn2.connect(defUser, defPass);
  
        try {
           subscribeTopic(conn, null, null, null, true, true);
  
           // send a message on the same connection => it should not be received as noLocal = true on subscribe
           send(conn, getTopicPrefix() + getTopicName(), null, "Hello World");
  
           ClientStompFrame frame = conn.receiveFrame(100);
           assertNull(frame, "No message should have been received since subscription specified noLocal");
  
           // send message on another STOMP connection => it should be received
           // a STOMP message does *not* automatically get a property named __AMQ_CID if there aren't any noLocal subscriptions on the connection
           send(conn2, getTopicPrefix() + getTopicName(), null, getName());
           frame = conn.receiveFrame(10000);
           assertNotNull(frame);
           assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
           assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
           assertEquals(getName(), frame.getBody());
        } finally {
           conn.disconnect();
           conn2.disconnect();
        }
     }
  
     @Test
     public void testSubscribeToTopicWithNoLocalAndNormal() throws Exception {
        conn.connect(defUser, defPass);
        String noLocalSubscriptionId = RandomUtil.randomUUIDString();
        String normalSubscriptionId = RandomUtil.randomUUIDString();
        subscribeTopic(conn, noLocalSubscriptionId, null, null, true, true);
        subscribeTopic(conn, normalSubscriptionId, null, null, true, false);
  
        StompProtocolManager stompProtocolManager = getStompProtocolManager();
        int totalSubCount = 0;
        int noLocalSubCount = 0;
        for (StompSession session : stompProtocolManager.getSessions()) {
           totalSubCount += session.getSubscriptionCount();
           noLocalSubCount += session.getNoLocalSubscriptionCount();
        }
        assertEquals(1, noLocalSubCount);
        assertEquals(2, totalSubCount);
  
        { // Send a message on the same connection. It should be received by the normal subscription and not by the noLocal one.
           send(conn, getTopicPrefix() + getTopicName(), null, "Hello World");
  
           ClientStompFrame frame = conn.receiveFrame(100);
           assertNotNull(frame);
           assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
           assertEquals(normalSubscriptionId, frame.getHeader(Stomp.Headers.Message.SUBSCRIPTION));
           assertNotNull(frame.getHeader("__AMQ_CID"));
           frame = conn.receiveFrame(100);
           assertNull(frame);
        }
  
        unsubscribe(conn, noLocalSubscriptionId, true);
  
        totalSubCount = 0;
        noLocalSubCount = 0;
        for (StompSession session : stompProtocolManager.getSessions()) {
           totalSubCount += session.getSubscriptionCount();
           noLocalSubCount += session.getNoLocalSubscriptionCount();
        }
        assertEquals(0, noLocalSubCount);
        assertEquals(1, totalSubCount);
  
        { // Send another message on the same connection. It should be received by the normal subscription.
           send(conn, getTopicPrefix() + getTopicName(), null, "Hello World");
  
           ClientStompFrame frame = conn.receiveFrame(100);
           assertNotNull(frame);
           assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
           assertEquals(normalSubscriptionId, frame.getHeader(Stomp.Headers.Message.SUBSCRIPTION));
           assertNull(frame.getHeader("__AMQ_CID"));
        }
  
        unsubscribe(conn, normalSubscriptionId, true);
  
        totalSubCount = 0;
        noLocalSubCount = 0;
        for (StompSession session : stompProtocolManager.getSessions()) {
           totalSubCount += session.getSubscriptionCount();
           noLocalSubCount += session.getNoLocalSubscriptionCount();
        }
        assertEquals(0, noLocalSubCount);
        assertEquals(0, totalSubCount);
  
        conn.disconnect();
     }
  
     @Test
     public void testSubscribeToTopicWithNoLocalAndSelector() throws Exception {
        conn.connect(defUser, defPass);
        subscribeTopic(conn, null, null, null, true, true, "a=foo OR b=bar");
  
        // send a message on the same connection => it should not be received as noLocal = true on subscribe
        ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND).addHeader(Stomp.Headers.Send.DESTINATION, getTopicPrefix() + getTopicName()).addHeader("b", "bar").setBody("Hello World");
        conn.sendFrame(frame);
  
        frame = conn.receiveFrame(100);
        assertNull(frame, "No message should have been received since noLocal=true");
  
        // send message on another JMS connection => it should be received
        sendJmsMessage(getName().getBytes(StandardCharsets.UTF_8), "b", "bar", topic);
        frame = conn.receiveFrame(10000);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        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);
        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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(getQueuePrefix() + getQueueName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
        String messageID = frame.getHeader(Stomp.Headers.Message.MESSAGE_ID);
        assertNotNull(messageID);
        assertEquals(getName(), frame.getBody());
  
        beginTransaction(conn, "tx1");
        ack(conn, null, messageID, "tx1");
        abortTransaction(conn, "tx1");
  
        frame = conn.receiveFrame(100);
        assertNull(frame, "No message should have been received as the message was acked even though the transaction has been aborted");
  
        unsubscribe(conn, null, getQueuePrefix() + getQueueName(), false, false);
  
        conn.disconnect();
     }
  
     @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);
           assertNotNull(consumer1.receive(TIME_OUT));
           assertNotNull(consumer2.receive(TIME_OUT));
           ClientStompFrame frame = conn.receiveFrame(TIME_OUT);
           assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
           assertEquals(getTopicPrefix() + getTopicName(), frame.getHeader(Stomp.Headers.Send.DESTINATION));
           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);
        assertNull(frame, "No message should have been received since subscription was removed");
  
        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.of(ADDRESS));
        assertNotNull(addressInfo, "No address was created with the name " + ADDRESS);
  
        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.
      */
     @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.of(ADDRESS));
        assertNotNull(addressInfo, "No address was created with the name " + ADDRESS);
        assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.ANYCAST));
        assertFalse(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST));
        assertNotNull(server.locateQueue(SimpleString.of(ADDRESS)));
  
        // sending a MULTICAST message should alter the address to support MULTICAST
        frame = send(conn, "/topic/" + ADDRESS, null, "Hello World 1", true);
        assertNotEquals("ERROR", frame.getCommand());
        addressInfo = server.getAddressInfo(SimpleString.of(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);
        assertNull(frame);
  
        // sending a message to the ANYCAST queue, should be received
        frame = send(conn, "/queue/" + ADDRESS, null, "Hello World 2", true);
        assertNotEquals("ERROR", frame.getCommand());
        frame = conn.receiveFrame(1000);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals("Hello World 2", frame.getBody());
        assertEquals(RoutingType.ANYCAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
        assertEquals("/queue/" + ADDRESS, frame.getHeader(Stomp.Headers.Send.DESTINATION));
        frame = conn.receiveFrame(100);
        assertNull(frame);
  
        unsubscribe(conn, null, "/queue/" + ADDRESS, true, false);
  
        Wait.assertTrue(() -> server.getAddressInfo(SimpleString.of(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);
        assertNotEquals("ERROR", frame.getCommand());
  
        // receive that message on the topic subscription
        frame = conn.receiveFrame(1000);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals("Hello World 3", frame.getBody());
        assertEquals(RoutingType.MULTICAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
        assertEquals("/topic/" + ADDRESS, frame.getHeader(Stomp.Headers.Send.DESTINATION));
        frame = conn.receiveFrame(100);
        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.
      */
     @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.of(ADDRESS));
        assertNotNull(addressInfo, "No address was created with the name " + ADDRESS);
        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);
        assertNotEquals("ERROR", frame.getCommand());
        addressInfo = server.getAddressInfo(SimpleString.of(ADDRESS));
        assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.ANYCAST));
        assertTrue(addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST));
        assertNotNull(server.locateQueue(SimpleString.of(ADDRESS)));
  
        // however, no message should be routed to the MULTICAST queue
        frame = conn.receiveFrame(100);
        assertNull(frame);
  
        // sending a message to the MULTICAST queue, should be received
        frame = send(conn, "/topic/" + ADDRESS, null, "Hello World 2", true);
        assertNotEquals("ERROR", frame.getCommand());
        frame = conn.receiveFrame(2000);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals("Hello World 2", frame.getBody());
        assertEquals(RoutingType.MULTICAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
        assertEquals("/topic/" + ADDRESS, frame.getHeader(Stomp.Headers.Send.DESTINATION));
        frame = conn.receiveFrame(100);
        assertNull(frame);
  
        frame = unsubscribe(conn, null, "/topic/" + ADDRESS, true, false);
        assertNotEquals("ERROR", frame.getCommand());
  
        // 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);
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals("Hello World 1", frame.getBody());
        assertEquals(RoutingType.ANYCAST.toString(), frame.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
        assertEquals("/queue/" + ADDRESS, frame.getHeader(Stomp.Headers.Send.DESTINATION));
        frame = conn.receiveFrame(100);
        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);
        assertNotNull(frame, "Should have received a message");
        assertEquals(Stomp.Responses.MESSAGE, frame.getCommand());
        assertEquals(ADDRESS, frame.getHeader(Stomp.Headers.Send.DESTINATION));
        assertEquals("Hello World", frame.getBody());
  
        conn.disconnect();
     }
  
     @Test
     public void testMulticastOperationsOnAnycastAddress() throws Exception {
        server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoCreateAddresses(false).setAutoCreateQueues(false));
        testRoutingSemantics(getQueuePrefix() + getQueueName(), RoutingType.MULTICAST.toString());
     }
  
     @Test
     public void testAnycastOperationsOnMulticastAddress() throws Exception {
        server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoCreateAddresses(false).setAutoCreateQueues(false));
        testRoutingSemantics(getTopicPrefix() + getTopicName(), RoutingType.ANYCAST.toString());
     }
  
     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);
  
        assertEquals(Boolean.TRUE.toString(), frame.getHeader(ManagementHelper.HDR_OPERATION_SUCCEEDED.toString()));
        // the address will be returned in the message body in a JSON array
        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);
  
        assertEquals(Boolean.TRUE.toString(), frame.getHeader(ManagementHelper.HDR_OPERATION_SUCCEEDED.toString()));
        // there is no such messages => 0 returned in a JSON array
        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(QueueConfiguration.of(queueA).setAddress(addressA).setRoutingType(RoutingType.ANYCAST).toJSON());
        serverControl.createQueue(QueueConfiguration.of(queueB).setAddress(addressA).setRoutingType(RoutingType.ANYCAST).toJSON());
        serverControl.createQueue(QueueConfiguration.of(queueC).setAddress(addressA).setRoutingType(RoutingType.MULTICAST).toJSON());
  
        send(conn, addressA, null, "Hello World!", true, RoutingType.ANYCAST);
  
        Wait.assertTrue(() -> server.locateQueue(SimpleString.of(queueA)).getMessageCount() + server.locateQueue(SimpleString.of(queueB)).getMessageCount() == 1);
        Wait.assertTrue(() -> server.locateQueue(SimpleString.of(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(QueueConfiguration.of(queueA).setAddress(addressA).setRoutingType(RoutingType.ANYCAST).toJSON());
        serverControl.createQueue(QueueConfiguration.of(queueB).setAddress(addressA).setRoutingType(RoutingType.MULTICAST).toJSON());
        serverControl.createQueue(QueueConfiguration.of(queueC).setAddress(addressA).setRoutingType(RoutingType.MULTICAST).toJSON());
  
        send(conn, addressA, null, "Hello World!", true, RoutingType.MULTICAST);
  
        Wait.assertTrue(() -> server.locateQueue(SimpleString.of(queueA)).getMessageCount() == 0);
        Wait.assertTrue(() -> server.locateQueue(SimpleString.of(queueC)).getMessageCount() + server.locateQueue(SimpleString.of(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(QueueConfiguration.of(queueA).setAddress(addressA).setRoutingType(RoutingType.ANYCAST).toJSON());
        serverControl.createQueue(QueueConfiguration.of(queueB).setAddress(addressA).setRoutingType(RoutingType.ANYCAST).toJSON());
        serverControl.createQueue(QueueConfiguration.of(queueC).setAddress(addressA).setRoutingType(RoutingType.MULTICAST).toJSON());
        serverControl.createQueue(QueueConfiguration.of(queueD).setAddress(addressA).setRoutingType(RoutingType.MULTICAST).toJSON());
  
        send(conn, addressA, null, "Hello World!", true);
  
        Wait.assertTrue(() -> server.locateQueue(SimpleString.of(queueA)).getMessageCount() + server.locateQueue(SimpleString.of(queueB)).getMessageCount() == 1);
        Wait.assertTrue(() -> server.locateQueue(SimpleString.of(queueC)).getMessageCount() + server.locateQueue(SimpleString.of(queueD)).getMessageCount() == 2);
     }
  
     @Test
     public void testAutoCreatedAnycastAddress() throws Exception {
        conn.connect(defUser, defPass);
  
        String queueName = UUID.randomUUID().toString();
        SimpleString simpleQueueName = SimpleString.of(queueName);
  
        assertNull(server.getAddressInfo(simpleQueueName));
        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.of(queueName);
  
        assertNull(server.getAddressInfo(simpleQueueName));
        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));
        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))) {
           assertFalse(((QueueBinding) b).getQueue().isDirectDeliver(), "Queue " + ((QueueBinding) b).getQueue().getName());
        }
  
        // 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 testSameMessageHasDifferentMessageIdPerConsumer() throws Exception {
        conn.connect(defUser, defPass);
  
        subscribeTopic(conn, "sub1", "client", null);
        subscribeTopic(conn, "sub2", "client", null);
  
        sendJmsMessage(getName(), topic);
  
        ClientStompFrame frame1 = conn.receiveFrame();
        String firstMessageID = frame1.getHeader(Stomp.Headers.Message.MESSAGE_ID);
        assertNotNull(firstMessageID);
  
        ClientStompFrame frame2 = conn.receiveFrame();
        String secondMessageID = frame2.getHeader(Stomp.Headers.Message.MESSAGE_ID);
        assertNotNull(secondMessageID);
        assertNotEquals(firstMessageID, secondMessageID, "MessageIDs must not equal");
  
        ack(conn, "sub1", frame1);
        ack(conn, "sub2", frame2);
  
        unsubscribe(conn, "sub1");
        unsubscribe(conn, "sub2");
  
        conn.disconnect();
     }
  
     @Test
     public void testMultipleSubscriptionsOnMulticastAddressReadSameMessage() throws Exception {
        doTestMultipleSubscriptionsOnMulticastAddressReadSameMessage(10);
     }
  
     @Test
     public void testMultipleSubscriptionsOnMulticastAddressReadSameLargeMessage() throws Exception {
        doTestMultipleSubscriptionsOnMulticastAddressReadSameMessage(120_000);
     }
  
     private void doTestMultipleSubscriptionsOnMulticastAddressReadSameMessage(int size) throws Exception {
        final String body = "A".repeat(size);
  
        final StompClientConnection conn_r1 = StompClientConnectionFactory.createClientConnection(uri);
        final StompClientConnection conn_r2 = StompClientConnectionFactory.createClientConnection(uri);
  
        try {
           conn_r1.connect(defUser, defPass);
           subscribeTopic(conn_r1, null, null, null);
  
           conn_r2.connect(defUser, defPass);
           subscribeTopic(conn_r2, null, null, null);
  
           // Sender
           conn.connect(defUser, defPass);
           send(conn, getTopicPrefix() + getTopicName(), null, body, true, RoutingType.MULTICAST);
  
           ClientStompFrame frame1 = conn_r1.receiveFrame(10000);
           ClientStompFrame frame2 = conn_r2.receiveFrame(10000);
  
           assertEquals(Stomp.Responses.MESSAGE, frame2.getCommand());
           assertEquals(Stomp.Responses.MESSAGE, frame1.getCommand());
  
           assertEquals(getTopicPrefix() + getTopicName(), frame2.getHeader(Stomp.Headers.Send.DESTINATION));
           assertEquals(getTopicPrefix() + getTopicName(), frame1.getHeader(Stomp.Headers.Send.DESTINATION));
  
           assertEquals(RoutingType.MULTICAST.toString(), frame2.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
           assertEquals(RoutingType.MULTICAST.toString(), frame1.getHeader(Stomp.Headers.Send.DESTINATION_TYPE));
  
           assertNull(frame2.getHeader(org.apache.activemq.artemis.api.core.Message.HDR_ROUTING_TYPE.toString()));
           assertNull(frame1.getHeader(org.apache.activemq.artemis.api.core.Message.HDR_ROUTING_TYPE.toString()));
  
           assertEquals(body, frame2.getBody());
           assertEquals(body, frame1.getBody());
        } finally {
           conn.disconnect();
           conn_r1.disconnect();
           conn_r2.disconnect();
        }
     }
  
     @Test
     public void testSubscribeToTopicWithEmptySelector() throws Exception {
        conn.connect(defUser, defPass);
        ClientStompFrame frame = subscribeTopic(conn, null, null, null, true, true, "   ");
        assertNotNull(frame);
        assertEquals(Stomp.Responses.ERROR, frame.getCommand());
        assertEquals("Selector cannot be empty", frame.getHeader(Stomp.Headers.Error.MESSAGE));
     }
  
     @Test
     public void testSubscribeToQueueWithEmptySelector() throws Exception {
        conn.connect(defUser, defPass);
        ClientStompFrame frame = subscribe(conn, null, null, null, "   ");
        assertNotNull(frame);
        assertEquals(Stomp.Responses.ERROR, frame.getCommand());
        assertEquals("Selector cannot be empty", frame.getHeader(Stomp.Headers.Error.MESSAGE));
     }
  
     @Test
     public void testSubscribeToTopicWithInvalidSelector() throws Exception {
        conn.connect(defUser, defPass);
        ClientStompFrame frame = subscribeTopic(conn, null, null, null, true, true, ") foo = 'bar' (");
        assertNotNull(frame);
        assertEquals(Stomp.Responses.ERROR, frame.getCommand());
        assertTrue(frame.getHeader(Stomp.Headers.Error.MESSAGE).contains("Invalid selector"));
     }
  
     @Test
     public void testSubscribeToQueueWithInvalidSelector() throws Exception {
        conn.connect(defUser, defPass);
        ClientStompFrame frame = subscribe(conn, null, null, null, ") foo = 'bar' (");
        assertNotNull(frame);
        assertEquals(Stomp.Responses.ERROR, frame.getCommand());
        assertTrue(frame.getHeader(Stomp.Headers.Error.MESSAGE).contains("Invalid selector"));
     }
  }
  