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.mqtt;
  
  import static org.apache.activemq.artemis.utils.collections.IterableStream.iterableOf;
  import static org.junit.jupiter.api.Assertions.assertArrayEquals;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertInstanceOf;
  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;
  
  import javax.jms.BytesMessage;
  import javax.jms.Connection;
  import javax.jms.Destination;
  import javax.jms.MessageConsumer;
  import javax.jms.MessageProducer;
  import javax.jms.Session;
  import java.io.EOFException;
  import java.lang.invoke.MethodHandles;
  import java.net.ProtocolException;
  import java.net.URI;
  import java.nio.charset.StandardCharsets;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.EnumSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Random;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicReference;
  import java.util.regex.Pattern;
  
  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.core.config.Configuration;
  import org.apache.activemq.artemis.core.config.CoreAddressConfiguration;
  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.protocol.mqtt.MQTTUtil;
  import org.apache.activemq.artemis.core.server.ActiveMQServer;
  import org.apache.activemq.artemis.core.server.Queue;
  import org.apache.activemq.artemis.core.server.impl.AddressInfo;
  import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
  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.tests.util.Wait;
  import org.apache.activemq.artemis.utils.RandomUtil;
  import org.apache.activemq.transport.amqp.client.AmqpClient;
  import org.apache.activemq.transport.amqp.client.AmqpConnection;
  import org.apache.activemq.transport.amqp.client.AmqpMessage;
  import org.apache.activemq.transport.amqp.client.AmqpSender;
  import org.apache.activemq.transport.amqp.client.AmqpSession;
  import org.eclipse.paho.client.mqttv3.MqttClient;
  import org.eclipse.paho.client.mqttv3.MqttMessage;
  import org.fusesource.mqtt.client.BlockingConnection;
  import org.fusesource.mqtt.client.MQTT;
  import org.fusesource.mqtt.client.Message;
  import org.fusesource.mqtt.client.QoS;
  import org.fusesource.mqtt.client.Topic;
  import org.fusesource.mqtt.client.Tracer;
  import org.fusesource.mqtt.codec.MQTTFrame;
  import org.fusesource.mqtt.codec.PUBLISH;
  import org.junit.jupiter.api.Disabled;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.Timeout;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * MQTT Test imported from ActiveMQ MQTT component.
   */
  public class MQTTTest extends MQTTTestSupport {
  
     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  
     private static final String AMQP_URI = "tcp://localhost:61616";
  
  
     @Override
     public void configureBroker() throws Exception {
        super.configureBroker();
        server.getConfiguration().setAddressQueueScanPeriod(100);
        server.getConfiguration().setMessageExpiryScanPeriod(100);
     }
  
     @Test
     public void testConnectWithLargePassword() throws Exception {
        for (String version : Arrays.asList("3.1", "3.1.1")) {
           String longString = new String(new char[65535]);
  
           BlockingConnection connection = null;
           try {
              MQTT mqtt = createMQTTConnection("test-" + version, true);
              mqtt.setUserName(longString);
              mqtt.setPassword(longString);
              mqtt.setConnectAttemptsMax(1);
              mqtt.setVersion(version);
              connection = mqtt.blockingConnection();
              connection.connect();
              BlockingConnection finalConnection = connection;
              assertTrue(Wait.waitFor(() -> finalConnection.isConnected(), 5000, 100), "Should be connected");
           } finally {
              if (connection != null && connection.isConnected()) connection.disconnect();
           }
        }
     }
  
     @Test
     @Timeout(60)
     public void testSendAndReceiveMQTT() throws Exception {
        testSendAndReceiveMQTT("");
     }
  
     @Test
     @Timeout(60)
     public void testSendAndReceiveMQTTHugePayload() throws Exception {
        testSendAndReceiveMQTT(RandomUtil.randomAlphaNumericString(110 * 1024));
     }
  
     public void testSendAndReceiveMQTT(String extraPayload) throws Exception {
        final MQTTClientProvider subscriptionProvider = getMQTTClientProvider();
        initializeConnection(subscriptionProvider);
  
        subscriptionProvider.subscribe("foo/bah", AT_MOST_ONCE);
  
        final CountDownLatch latch = new CountDownLatch(NUM_MESSAGES);
  
        Thread thread = new Thread(() -> {
           for (int i = 0; i < NUM_MESSAGES; i++) {
              try {
                 byte[] payload = subscriptionProvider.receive(10000);
                 assertNotNull(payload, "Should get a message");
                 latch.countDown();
              } catch (Exception e) {
                 e.printStackTrace();
                 break;
              }
  
           }
        });
        thread.start();
  
        final MQTTClientProvider publishProvider = getMQTTClientProvider();
        initializeConnection(publishProvider);
  
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Message " + i + extraPayload;
           publishProvider.publish("foo/bah", payload.getBytes(), AT_LEAST_ONCE);
        }
  
        latch.await(10, TimeUnit.SECONDS);
        assertEquals(0, latch.getCount());
        subscriptionProvider.disconnect();
        publishProvider.disconnect();
     }
  
     @Test
     public void testProducerMetrics() throws Exception {
  
        final MQTTClientProvider publishProvider = getMQTTClientProvider();
        initializeConnection(publishProvider);
  
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Message " + i;
           publishProvider.publish("foo/bah", payload.getBytes(), AT_LEAST_ONCE);
        }
  
        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);
  
        assertNotEquals("", producer.getString(ProducerField.ID.getName()), ProducerField.ID.getName());
        assertNotEquals("", producer.getString(ProducerField.SESSION.getName()), ProducerField.SESSION.getName());
        assertNotEquals("", producer.getString(ProducerField.CLIENT_ID.getName()), ProducerField.CLIENT_ID.getName());
        assertEquals("", producer.getString(ProducerField.USER.getName()), ProducerField.USER.getName());
        assertEquals("MQTT", producer.getString(ProducerField.PROTOCOL.getName()), ProducerField.PROTOCOL.getAlternativeName());
        assertEquals("ANONYMOUS", producer.getString(ProducerField.ADDRESS.getName()), ProducerField.ADDRESS.getName());
        assertNotEquals("", producer.getString(ProducerField.LOCAL_ADDRESS.getName()), ProducerField.LOCAL_ADDRESS.getName());
        assertNotEquals("", producer.getString(ProducerField.REMOTE_ADDRESS.getName()), ProducerField.REMOTE_ADDRESS.getName());
        assertNotEquals("", producer.getString(ProducerField.CREATION_TIME.getName()), ProducerField.CREATION_TIME.getName());
        assertEquals(NUM_MESSAGES, producer.getInt(ProducerField.MESSAGE_SENT.getName()), ProducerField.MESSAGE_SENT.getName());
        assertEquals("", producer.getString(ProducerField.LAST_PRODUCED_MESSAGE_ID.getName()), ProducerField.LAST_PRODUCED_MESSAGE_ID.getName());
  
        final MQTTClientProvider publishProvider2 = getMQTTClientProvider();
        initializeConnection(publishProvider2);
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Message " + i;
           publishProvider2.publish("foo/bah", payload.getBytes(), AT_LEAST_ONCE);
        }
        filterString = createJsonFilter("", "", "");
        producersAsJsonString = server.getActiveMQServerControl().listProducers(filterString, 1, 50);
        producersAsJsonObject = JsonUtil.readJsonObject(producersAsJsonString);
        array = (JsonArray) producersAsJsonObject.get("data");
  
        assertEquals(2, array.size(), "number of producers returned from query");
        publishProvider.disconnect();
  
  
        publishProvider2.disconnect();
  
        filterString = createJsonFilter("", "", "");
        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
     @Timeout(60)
     public void testDirectDeliverFalse() throws Exception {
        final MQTTClientProvider subscriptionProvider = getMQTTClientProvider();
        initializeConnection(subscriptionProvider);
  
        subscriptionProvider.subscribe("foo/bah", AT_MOST_ONCE);
  
  
        for (Binding b : iterableOf(server.getPostOffice().getAllBindings().filter(QueueBinding.class::isInstance))) {
           assertFalse(((QueueBinding)b).getQueue().isDirectDeliver(), "Queue " + ((QueueBinding) b).getQueue().getName());
        }
  
        subscriptionProvider.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testUnsubscribeMQTT() throws Exception {
        final MQTTClientProvider subscriptionProvider = getMQTTClientProvider();
        initializeConnection(subscriptionProvider);
  
        String topic = "foo/bah";
  
        subscriptionProvider.subscribe(topic, AT_MOST_ONCE);
  
        final CountDownLatch latch = new CountDownLatch(NUM_MESSAGES / 2);
  
        Thread thread = new Thread(() -> {
           for (int i = 0; i < NUM_MESSAGES; i++) {
              try {
                 byte[] payload = subscriptionProvider.receive(10000);
                 assertNotNull(payload, "Should get a message");
                 latch.countDown();
              } catch (Exception e) {
                 e.printStackTrace();
                 break;
              }
  
           }
        });
        thread.start();
  
        final MQTTClientProvider publishProvider = getMQTTClientProvider();
        initializeConnection(publishProvider);
  
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Message " + i;
           if (i == NUM_MESSAGES / 2) {
              subscriptionProvider.unsubscribe(topic);
           }
           publishProvider.publish(topic, payload.getBytes(), AT_LEAST_ONCE);
        }
  
        latch.await(20, TimeUnit.SECONDS);
        assertEquals(0, latch.getCount());
        subscriptionProvider.disconnect();
        publishProvider.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testSendAtMostOnceReceiveExactlyOnce() throws Exception {
        /*
         * Although subscribing with EXACTLY ONCE, the message gets published with AT_MOST_ONCE - in MQTT the QoS is
         * always determined by the message as published - not the wish of the subscriber
         */
        final MQTTClientProvider provider = getMQTTClientProvider();
        initializeConnection(provider);
        provider.subscribe("foo", EXACTLY_ONCE);
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Test Message: " + i;
           provider.publish("foo", payload.getBytes(), AT_MOST_ONCE);
           byte[] message = provider.receive(5000);
           assertNotNull(message, "Should get a message");
           assertEquals(payload, new String(message));
        }
        provider.disconnect();
     }
  
     @Test
     @Timeout(120)
     public void testManagementQueueMessagesAreAckd() throws Exception {
        String clientId = "test.client.id";
        final MQTTClientProvider provider = getMQTTClientProvider();
        provider.setClientId(clientId);
        initializeConnection(provider);
        provider.subscribe("foo", EXACTLY_ONCE);
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Test Message: " + i;
           provider.publish("foo", payload.getBytes(), EXACTLY_ONCE);
           byte[] message = provider.receive(5000);
           assertNotNull(message, "Should get a message");
           assertEquals(payload, new String(message));
        }
  
        final Queue queue = server.locateQueue(SimpleString.of(MQTTUtil.QOS2_MANAGEMENT_QUEUE_PREFIX + clientId));
  
        Wait.waitFor(() -> queue.getMessageCount() == 0, 1000, 100);
  
        assertEquals(0, queue.getMessageCount());
        provider.disconnect();
     }
  
     @Test
     @Timeout(120)
     public void testSendAtLeastOnceReceiveExactlyOnce() throws Exception {
        final MQTTClientProvider provider = getMQTTClientProvider();
        initializeConnection(provider);
        provider.subscribe("foo", EXACTLY_ONCE);
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Test Message: " + i;
           provider.publish("foo", payload.getBytes(), AT_LEAST_ONCE);
           byte[] message = provider.receive(5000);
           assertNotNull(message, "Should get a message");
           assertEquals(payload, new String(message));
        }
        provider.disconnect();
     }
  
     @Test
     @Timeout(120)
     public void testSendAtLeastOnceReceiveAtMostOnce() throws Exception {
        final MQTTClientProvider provider = getMQTTClientProvider();
        initializeConnection(provider);
        provider.subscribe("foo", AT_MOST_ONCE);
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Test Message: " + i;
           provider.publish("foo", payload.getBytes(), AT_LEAST_ONCE);
           byte[] message = provider.receive(5000);
           assertNotNull(message, "Should get a message");
           assertEquals(payload, new String(message));
        }
        provider.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testSendAndReceiveAtMostOnce() throws Exception {
        final MQTTClientProvider provider = getMQTTClientProvider();
        initializeConnection(provider);
        provider.subscribe("foo", AT_MOST_ONCE);
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Test Message: " + i;
           provider.publish("foo", payload.getBytes(), AT_MOST_ONCE);
           byte[] message = provider.receive(5000);
           assertNotNull(message, "Should get a message");
           assertEquals(payload, new String(message));
        }
        provider.disconnect();
     }
  
     @Test
     @Timeout(120)
     public void testSendAndReceiveAtLeastOnce() throws Exception {
        final MQTTClientProvider provider = getMQTTClientProvider();
        initializeConnection(provider);
        provider.subscribe("foo", AT_LEAST_ONCE);
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Test Message: " + i;
           provider.publish("foo", payload.getBytes(), AT_LEAST_ONCE);
           byte[] message = provider.receive(5000);
           assertNotNull(message, "Should get a message");
           assertEquals(payload, new String(message));
        }
        provider.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testSendAndReceiveExactlyOnceWithInterceptors() throws Exception {
        MQTTIncomingInterceptor.clear();
        MQTTOutoingInterceptor.clear();
        final MQTTClientProvider publisher = getMQTTClientProvider();
        initializeConnection(publisher);
  
        final MQTTClientProvider subscriber = getMQTTClientProvider();
        initializeConnection(subscriber);
  
        subscriber.subscribe("foo", EXACTLY_ONCE);
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String payload = "Test Message: " + i;
           publisher.publish("foo", payload.getBytes(), EXACTLY_ONCE);
           byte[] message = subscriber.receive(5000);
           assertNotNull(message, "Should get a message + [" + i + "]");
           assertEquals(payload, new String(message));
        }
        subscriber.disconnect();
        publisher.disconnect();
        assertEquals(NUM_MESSAGES, MQTTIncomingInterceptor.getMessageCount());
        assertEquals(NUM_MESSAGES, MQTTOutoingInterceptor.getMessageCount());
     }
  
     @Disabled
     @Test
     @Timeout(600)
     public void testSendMoreThanUniqueId() throws Exception {
        int messages = MQTTUtil.TWO_BYTE_INT_MAX;
  
        final MQTTClientProvider publisher = getMQTTClientProvider();
        initializeConnection(publisher);
  
        final MQTTClientProvider subscriber = getMQTTClientProvider();
        initializeConnection(subscriber);
  
        int count = 0;
        subscriber.subscribe("foo", EXACTLY_ONCE);
        for (int i = 0; i < messages; i++) {
           String payload = "Test Message: " + i;
           publisher.publish("foo", payload.getBytes(), EXACTLY_ONCE);
           byte[] message = subscriber.receive(5000);
           assertNotNull(message, "Should get a message + [" + i + "]");
           assertEquals(payload, new String(message));
           count++;
        }
  
        assertEquals(messages, count);
        subscriber.disconnect();
        publisher.disconnect();
     }
  
     @Test
     @Timeout(180)
     public void testNoMessageIdReuseBeforeAcknowledgment() throws Exception {
        final int messages = MQTTUtil.TWO_BYTE_INT_MAX;
  
        final MQTTClientProvider publisher = getMQTTClientProvider();
        initializeConnection(publisher);
  
        final MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("foo");
        final short[] messageId = new short[1];
        mqtt.setTracer(new Tracer() {
           @Override
           public void onReceive(MQTTFrame frame) {
              if (frame.messageType() == PUBLISH.TYPE) {
                 try {
                    PUBLISH publish = new PUBLISH();
                    publish.decode(frame);
                    messageId[0] = publish.messageId();
                 } catch (ProtocolException e) {
                    fail("Error decoding publish " + e.getMessage());
                 }
              }
           }
        });
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        connection.subscribe(new Topic[]{new Topic("foo", QoS.EXACTLY_ONCE)});
  
        publisher.publish("foo", "First Message".getBytes(), EXACTLY_ONCE);
        final Message firstMessage = connection.receive(5000, TimeUnit.MILLISECONDS);
        final short firstMessageId = messageId[0];
  
        publisher.publish("foo", "Second Message".getBytes(), EXACTLY_ONCE);
        final Message secondMessage = connection.receive(5000, TimeUnit.MILLISECONDS);
        final short secondMessageId = messageId[0];
  
        int count = 0;
        for (int i = 0; i < messages; i++) {
           String payload = "Test Message: " + i;
           publisher.publish("foo", payload.getBytes(), EXACTLY_ONCE);
           Message message = connection.receive(5000, TimeUnit.MILLISECONDS);
           assertNotNull(message, "Should get a message + [" + i + "]");
           assertEquals(payload, new String(message.getPayload()));
           assertFalse(firstMessageId == messageId[0] || secondMessageId == messageId[0], "Message ID must not be reused until previous message acknowledgment");
           message.ack();
           count++;
        }
  
        firstMessage.ack();
        secondMessage.ack();
  
        assertEquals(messages, count);
        connection.unsubscribe(new String[]{"foo"});
        connection.disconnect();
        publisher.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testSendAndReceiveLargeMessages() throws Exception {
        byte[] payload = new byte[1024 * 32];
        for (int i = 0; i < payload.length; i++) {
           payload[i] = '2';
        }
        final MQTTClientProvider publisher = getMQTTClientProvider();
        initializeConnection(publisher);
  
        final MQTTClientProvider subscriber = getMQTTClientProvider();
        initializeConnection(subscriber);
  
        subscriber.subscribe("foo", AT_LEAST_ONCE);
        for (int i = 0; i < 10; i++) {
           publisher.publish("foo", payload, AT_LEAST_ONCE);
           byte[] message = subscriber.receive(5000);
           assertNotNull(message, "Should get a message");
  
           assertArrayEquals(payload, message);
        }
        subscriber.disconnect();
        publisher.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testSendAndReceiveRetainedMessages() throws Exception {
        final MQTTClientProvider publisher = getMQTTClientProvider();
        initializeConnection(publisher);
  
        final MQTTClientProvider subscriber = getMQTTClientProvider();
        initializeConnection(subscriber);
  
        String RETAINED = "retained";
        publisher.publish("foo", RETAINED.getBytes(), AT_LEAST_ONCE, true);
  
        List<String> messages = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
           messages.add("TEST MESSAGE:" + i);
        }
  
        subscriber.subscribe("foo", AT_LEAST_ONCE);
  
        for (int i = 0; i < 10; i++) {
           publisher.publish("foo", messages.get(i).getBytes(), AT_LEAST_ONCE);
        }
        byte[] msg = subscriber.receive(5000);
        assertNotNull(msg);
        assertEquals(RETAINED, new String(msg));
  
        for (int i = 0; i < 10; i++) {
           msg = subscriber.receive(5000);
           assertNotNull(msg);
           assertEquals(messages.get(i), new String(msg));
        }
        subscriber.disconnect();
        publisher.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testSendAndReceiveRetainedLargeMessage() throws Exception {
        try (AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler()) {
           byte[] payload = new byte[ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE * 2];
           for (int i = 0; i < payload.length; i++) {
              payload[i] = '2';
           }
           String body = "message";
  
           String smallRetain = "retain";
           final MQTTClientProvider publisher = getMQTTClientProvider();
           initializeConnection(publisher);
  
           final MQTTClientProvider subscriber = getMQTTClientProvider();
           initializeConnection(subscriber);
  
           publisher.publish("foo", payload, AT_LEAST_ONCE, true);
  
           subscriber.subscribe("foo", AT_LEAST_ONCE);
  
           publisher.publish("foo", body.getBytes(), AT_LEAST_ONCE, false);
           byte[] msg = subscriber.receive(5000);
           assertNotNull(msg);
           assertEquals(msg.length, payload.length);
  
           msg = subscriber.receive(5000);
           assertNotNull(msg);
           assertEquals(msg.length, body.length());
  
           subscriber.disconnect();
  
           final MQTTClientProvider subscriber2 = getMQTTClientProvider();
           initializeConnection(subscriber2);
           subscriber2.subscribe("foo", AT_LEAST_ONCE);
           msg = subscriber2.receive(5000);
           assertNotNull(msg);
           assertEquals(msg.length, payload.length);
           subscriber2.disconnect();
           publisher.publish("foo", smallRetain.getBytes(), AT_LEAST_ONCE, true);
           final MQTTClientProvider subscriber3 = getMQTTClientProvider();
           initializeConnection(subscriber3);
           subscriber3.subscribe("foo", AT_LEAST_ONCE);
           msg = subscriber3.receive(5000);
           assertNotNull(msg);
           assertEquals(msg.length, smallRetain.getBytes().length);
           subscriber3.disconnect();
           publisher.disconnect();
  
           assertFalse(loggerHandler.findText("Exception"));
        }
  
     }
  
     @Test
     @Timeout(30)
     public void testValidZeroLengthClientId() throws Exception {
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("");
        mqtt.setCleanSession(true);
  
        BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        connection.disconnect();
     }
  
     @Test
     @Timeout(120)
     public void testMQTTPathPatterns() throws Exception {
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("");
        mqtt.setCleanSession(true);
  
        BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
  
        final String RETAINED = "RETAINED";
        String[] topics = {"TopicA", "/TopicA", "/", "TopicA/", "//"};
        for (String topic : topics) {
           // test retained message
           connection.publish(topic, (RETAINED + topic).getBytes(), QoS.AT_LEAST_ONCE, true);
  
           connection.subscribe(new Topic[]{new Topic(topic, QoS.AT_LEAST_ONCE)});
           Message msg = connection.receive(5, TimeUnit.SECONDS);
           assertNotNull(msg, "No message for " + topic);
           assertEquals(RETAINED + topic, new String(msg.getPayload()));
           msg.ack();
  
           // test non-retained message
           connection.publish(topic, topic.getBytes(), QoS.AT_LEAST_ONCE, false);
           msg = connection.receive(1000, TimeUnit.MILLISECONDS);
           assertNotNull(msg);
           assertEquals(topic, new String(msg.getPayload()));
           msg.ack();
  
           connection.unsubscribe(new String[]{topic});
        }
        connection.disconnect();
  
        // test wildcard patterns with above topics
        String[] wildcards = {"#", "+", "+/#", "/+", "+/", "+/+", "+/+/", "+/+/+"};
        for (String wildcard : wildcards) {
           final Pattern pattern = Pattern.compile(wildcard.replaceAll("/?#", "(/?.*)*").replaceAll("\\+", "[^/]*"));
  
           connection = mqtt.blockingConnection();
           connection.connect();
           final byte[] qos = connection.subscribe(new Topic[]{new Topic(wildcard, QoS.AT_LEAST_ONCE)});
           assertNotEquals((byte) 0x80, qos[0], "Subscribe failed " + wildcard);
  
           // test retained messages
           Message msg = connection.receive(5, TimeUnit.SECONDS);
           do {
              assertNotNull(msg, "RETAINED null " + wildcard);
              String msgPayload = new String(msg.getPayload());
              assertTrue(msgPayload.startsWith(RETAINED), "RETAINED prefix " + wildcard + " msg " + msgPayload);
              assertTrue(pattern.matcher(msg.getTopic()).matches(), "RETAINED matching " + wildcard + " " + msg.getTopic());
              msg.ack();
              msg = connection.receive(500, TimeUnit.MILLISECONDS);
           }
           while (msg != null);
  
           // test non-retained message
           for (String topic : topics) {
              connection.publish(topic, topic.getBytes(), QoS.AT_LEAST_ONCE, false);
           }
           msg = connection.receive(1000, TimeUnit.MILLISECONDS);
           do {
              assertNotNull(msg, "Non-retained Null " + wildcard);
              assertTrue(pattern.matcher(msg.getTopic()).matches(), "Non-retained matching " + wildcard + " " + msg.getTopic());
              msg.ack();
              msg = connection.receive(500, TimeUnit.MILLISECONDS);
           }
           while (msg != null);
  
           connection.unsubscribe(new String[]{wildcard});
           connection.disconnect();
        }
     }
  
     @Test
     @Timeout(60)
     public void testMQTTRetainQoS() throws Exception {
        String[] topics = {"AT_MOST_ONCE", "AT_LEAST_ONCE", "EXACTLY_ONCE"};
        for (int i = 0; i < topics.length; i++) {
           final String topic = topics[i];
  
           MQTT mqtt = createMQTTConnection();
           mqtt.setClientId("foo");
           mqtt.setKeepAlive((short) 2);
  
           final int[] actualQoS = {-1};
           mqtt.setTracer(new Tracer() {
              @Override
              public void onReceive(MQTTFrame frame) {
                 // validate the QoS
                 if (frame.messageType() == PUBLISH.TYPE) {
                    actualQoS[0] = frame.qos().ordinal();
                 }
              }
           });
  
           final BlockingConnection connection = mqtt.blockingConnection();
           connection.connect();
           connection.publish(topic, topic.getBytes(), QoS.EXACTLY_ONCE, true);
           connection.subscribe(new Topic[]{new Topic(topic, QoS.valueOf(topic))});
  
           final Message msg = connection.receive(5000, TimeUnit.MILLISECONDS);
           assertNotNull(msg);
           assertEquals(topic, new String(msg.getPayload()));
  
           Wait.assertEquals(i, () -> actualQoS[0]);
           msg.ack();
  
           connection.unsubscribe(new String[]{topic});
           connection.disconnect();
        }
  
     }
  
     @Test
     @Timeout(60)
     public void testDuplicateSubscriptions() throws Exception {
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("foo");
        mqtt.setKeepAlive((short) 20);
  
        final int[] actualQoS = {-1};
        mqtt.setTracer(new Tracer() {
           @Override
           public void onReceive(MQTTFrame frame) {
              // validate the QoS
              if (frame.messageType() == PUBLISH.TYPE) {
                 actualQoS[0] = frame.qos().ordinal();
              }
           }
        });
  
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
  
        final String RETAIN = "RETAIN";
        connection.publish("TopicA", RETAIN.getBytes(), QoS.EXACTLY_ONCE, true);
  
        QoS[] qoss = {QoS.AT_MOST_ONCE, QoS.AT_MOST_ONCE, QoS.AT_LEAST_ONCE, QoS.EXACTLY_ONCE};
        for (QoS qos : qoss) {
           connection.subscribe(new Topic[]{new Topic("TopicA", qos)});
  
           final Message msg = connection.receive(5000, TimeUnit.MILLISECONDS);
           assertNotNull(msg, "No message for " + qos);
           assertEquals(RETAIN, new String(msg.getPayload()));
           msg.ack();
           Wait.assertEquals(qos.ordinal(), () -> actualQoS[0]);
           actualQoS[0] = -1;
        }
  
        connection.unsubscribe(new String[]{"TopicA"});
        connection.disconnect();
  
     }
  
     @Test
     @Timeout(120)
     public void testRetainedMessage() throws Exception {
        MQTT mqtt = createMQTTConnection();
        mqtt.setKeepAlive((short) 60);
  
        final String RETAIN = "RETAIN";
        final String TOPICA = "TopicA";
  
        final String[] clientIds = {null, "foo", "durable"};
        for (String clientId : clientIds) {
           logger.debug("Testing now with Client ID: {}", clientId);
  
           mqtt.setClientId(clientId);
           mqtt.setCleanSession(!"durable".equals(clientId));
  
           BlockingConnection connection = mqtt.blockingConnection();
           connection.connect();
  
           // set retained message and check
           connection.publish(TOPICA, RETAIN.getBytes(), QoS.EXACTLY_ONCE, true);
           connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)});
           Message msg = connection.receive(5000, TimeUnit.MILLISECONDS);
           assertNotNull(msg, "No retained message for " + clientId);
           assertEquals(RETAIN, new String(msg.getPayload()));
           msg.ack();
           assertNull(connection.receive(100, TimeUnit.MILLISECONDS));
  
           // test duplicate subscription
           connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)});
           msg = connection.receive(15000, TimeUnit.MILLISECONDS);
           assertNotNull(msg, "No retained message on duplicate subscription for " + clientId);
           assertEquals(RETAIN, new String(msg.getPayload()));
           msg.ack();
           assertNull(connection.receive(100, TimeUnit.MILLISECONDS));
           connection.unsubscribe(new String[]{TOPICA});
  
           // clear retained message and check that we don't receive it
           connection.publish(TOPICA, "".getBytes(), QoS.AT_MOST_ONCE, true);
           connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)});
           msg = connection.receive(100, TimeUnit.MILLISECONDS);
           assertNull(msg, "Retained message not cleared for " + clientId);
           connection.unsubscribe(new String[]{TOPICA});
  
           // set retained message again and check
           connection.publish(TOPICA, RETAIN.getBytes(), QoS.EXACTLY_ONCE, true);
           connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)});
           msg = connection.receive(5000, TimeUnit.MILLISECONDS);
           assertNotNull(msg, "No reset retained message for " + clientId);
           assertEquals(RETAIN, new String(msg.getPayload()));
           msg.ack();
           assertNull(connection.receive(100, TimeUnit.MILLISECONDS));
  
           // re-connect and check
           connection.disconnect();
           connection = mqtt.blockingConnection();
           connection.connect();
           connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)});
           msg = connection.receive(5000, TimeUnit.MILLISECONDS);
           assertNotNull(msg, "No reset retained message for " + clientId);
           assertEquals(RETAIN, new String(msg.getPayload()));
           msg.ack();
           assertNull(connection.receive(100, TimeUnit.MILLISECONDS));
  
           connection.unsubscribe(new String[]{TOPICA});
           connection.disconnect();
        }
     }
  
     @Disabled
     @Test
     @Timeout(120)
     public void testRetainedMessageOnVirtualTopics() throws Exception {
        MQTT mqtt = createMQTTConnection();
        mqtt.setKeepAlive((short) 60);
  
        final String RETAIN = "RETAIN";
        final String TOPICA = "VirtualTopic/TopicA";
  
        final String[] clientIds = {null, "foo", "durable"};
        for (String clientId : clientIds) {
           logger.debug("Testing now with Client ID: {}", clientId);
  
           mqtt.setClientId(clientId);
           mqtt.setCleanSession(!"durable".equals(clientId));
  
           BlockingConnection connection = mqtt.blockingConnection();
           connection.connect();
  
           // set retained message and check
           connection.publish(TOPICA, RETAIN.getBytes(), QoS.EXACTLY_ONCE, true);
           connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)});
           Message msg = connection.receive(5000, TimeUnit.MILLISECONDS);
           assertNotNull(msg, "No retained message for " + clientId);
           assertEquals(RETAIN, new String(msg.getPayload()));
           msg.ack();
           assertNull(connection.receive(100, TimeUnit.MILLISECONDS));
  
           // test duplicate subscription
           connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)});
           msg = connection.receive(15000, TimeUnit.MILLISECONDS);
           assertNotNull(msg, "No retained message on duplicate subscription for " + clientId);
           assertEquals(RETAIN, new String(msg.getPayload()));
           msg.ack();
           assertNull(connection.receive(100, TimeUnit.MILLISECONDS));
           connection.unsubscribe(new String[]{TOPICA});
  
           // clear retained message and check that we don't receive it
           connection.publish(TOPICA, "".getBytes(), QoS.AT_MOST_ONCE, true);
           connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)});
           msg = connection.receive(500, TimeUnit.MILLISECONDS);
           assertNull(msg, "Retained message not cleared for " + clientId);
           connection.unsubscribe(new String[]{TOPICA});
  
           // set retained message again and check
           connection.publish(TOPICA, RETAIN.getBytes(), QoS.EXACTLY_ONCE, true);
           connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)});
           msg = connection.receive(5000, TimeUnit.MILLISECONDS);
           assertNotNull(msg, "No reset retained message for " + clientId);
           assertEquals(RETAIN, new String(msg.getPayload()));
           msg.ack();
           assertNull(connection.receive(100, TimeUnit.MILLISECONDS));
  
           // re-connect and check
           connection.disconnect();
           connection = mqtt.blockingConnection();
           connection.connect();
           connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)});
           msg = connection.receive(5000, TimeUnit.MILLISECONDS);
           assertNotNull(msg, "No reset retained message for " + clientId);
           assertEquals(RETAIN, new String(msg.getPayload()));
           msg.ack();
           assertNull(connection.receive(100, TimeUnit.MILLISECONDS));
  
           logger.debug("Test now unsubscribing from: {} for the last time", TOPICA);
           connection.unsubscribe(new String[]{TOPICA});
           connection.disconnect();
        }
     }
  
     @Test
     @Timeout(60)
     public void testUniqueMessageIds() throws Exception {
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("foo");
        mqtt.setKeepAlive((short) 2);
        mqtt.setCleanSession(true);
  
        final List<PUBLISH> publishList = new ArrayList<>();
        mqtt.setTracer(new Tracer() {
           @Override
           public void onReceive(MQTTFrame frame) {
              logger.debug("Client received:\n{}", frame);
              if (frame.messageType() == PUBLISH.TYPE) {
                 PUBLISH publish = new PUBLISH();
                 try {
                    publish.decode(frame);
                 } catch (ProtocolException e) {
                    fail("Error decoding publish " + e.getMessage());
                 }
                 publishList.add(publish);
              }
           }
  
           @Override
           public void onSend(MQTTFrame frame) {
              logger.debug("Client sent:\n{}", frame);
           }
        });
  
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
  
        // create overlapping subscriptions with different QoSs
        QoS[] qoss = {QoS.AT_MOST_ONCE, QoS.AT_LEAST_ONCE, QoS.EXACTLY_ONCE};
        final String TOPIC = "TopicA/";
  
        // publish retained message
        connection.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, true);
  
        String[] subs = {TOPIC, "TopicA/#", "TopicA/+"};
        for (int i = 0; i < qoss.length; i++) {
           connection.subscribe(new Topic[]{new Topic(subs[i], qoss[i])});
        }
  
        // publish non-retained message
        connection.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, false);
        int received = 0;
  
        Message msg = connection.receive(5000, TimeUnit.MILLISECONDS);
        do {
           assertNotNull(msg);
           assertEquals(TOPIC, new String(msg.getPayload()));
           msg.ack();
           int waitCount = 0;
           while (publishList.size() <= received && waitCount < 10) {
              Thread.sleep(1000);
              waitCount++;
           }
           msg = connection.receive(5000, TimeUnit.MILLISECONDS);
        }
        while (msg != null && received++ < subs.length * 2);
        assertEquals(subs.length * 2, received + 1, "Unexpected number of messages");
  
        // make sure we received distinct ids for QoS != AT_MOST_ONCE, and 0 for
        // AT_MOST_ONCE
        for (int i = 0; i < publishList.size(); i++) {
           for (int j = i + 1; j < publishList.size(); j++) {
              final PUBLISH publish1 = publishList.get(i);
              final PUBLISH publish2 = publishList.get(j);
              boolean qos0 = false;
              if (publish1.qos() == QoS.AT_MOST_ONCE) {
                 qos0 = true;
                 assertEquals(0, publish1.messageId());
              }
              if (publish2.qos() == QoS.AT_MOST_ONCE) {
                 qos0 = true;
                 assertEquals(0, publish2.messageId());
              }
              if (!qos0) {
                 assertNotEquals(publish1.messageId(), publish2.messageId());
              }
           }
        }
  
        connection.unsubscribe(subs);
        connection.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testResendMessageId() throws Exception {
        final MQTT mqtt = createMQTTConnection("resend", false);
        mqtt.setKeepAlive((short) 5);
  
        final List<PUBLISH> publishList = new ArrayList<>();
        mqtt.setTracer(new Tracer() {
           @Override
           public void onReceive(MQTTFrame frame) {
              logger.debug("Client received:\n{}", frame);
              if (frame.messageType() == PUBLISH.TYPE) {
                 PUBLISH publish = new PUBLISH();
                 try {
                    publish.decode(frame);
                 } catch (ProtocolException e) {
                    fail("Error decoding publish " + e.getMessage());
                 }
                 publishList.add(publish);
              }
           }
  
           @Override
           public void onSend(MQTTFrame frame) {
              logger.debug("Client sent:\n{}", frame);
           }
        });
  
        BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        final String TOPIC = "TopicA/";
        final String[] topics = new String[]{TOPIC, "TopicA/+"};
        connection.subscribe(new Topic[]{new Topic(topics[0], QoS.AT_LEAST_ONCE), new Topic(topics[1], QoS.EXACTLY_ONCE)});
  
        // publish non-retained message
        connection.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, false);
  
        assertTrue(Wait.waitFor(() -> publishList.size() == 2, 5000));
  
        connection.disconnect();
  
        connection = mqtt.blockingConnection();
        connection.connect();
  
        assertTrue(Wait.waitFor(() -> publishList.size() == 4, 5000));
  
        // TODO Investigate if receiving the same ID for overlapping subscriptions is actually spec compliant.
        // In Artemis we send a new ID for every copy of the message.
  
        // make sure we received duplicate message ids
        //      assertTrue(publishList.get(0).messageId() == publishList.get(2).messageId() || publishList.get(0).messageId() == publishList.get(3).messageId());
        //      assertTrue(publishList.get(1).messageId() == publishList.get(3).messageId() || publishList.get(1).messageId() == publishList.get(2).messageId());
        //      assertTrue(publishList.get(2).dup() && publishList.get(3).dup());
  
        connection.unsubscribe(topics);
        connection.disconnect();
     }
  
     @Test
     @Timeout(90)
     public void testPacketIdGeneratorNonCleanSession() throws Exception {
        final MQTT mqtt = createMQTTConnection("nonclean-packetid", false);
        mqtt.setKeepAlive((short) 15);
  
        final Map<Short, PUBLISH> publishMap = new ConcurrentHashMap<>();
        mqtt.setTracer(new Tracer() {
           @Override
           public void onReceive(MQTTFrame frame) {
              logger.debug("Client received:\n{}", frame);
              if (frame.messageType() == PUBLISH.TYPE) {
                 PUBLISH publish = new PUBLISH();
                 try {
                    publish.decode(frame);
                    logger.debug("PUBLISH {}", publish);
                 } catch (ProtocolException e) {
                    fail("Error decoding publish " + e.getMessage());
                 }
                 if (publishMap.get(publish.messageId()) != null) {
                    assertTrue(publish.dup());
                 }
                 publishMap.put(publish.messageId(), publish);
              }
           }
  
           @Override
           public void onSend(MQTTFrame frame) {
              logger.debug("Client sent:\n{}", frame);
           }
        });
  
        BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        final String TOPIC = "TopicA/";
        connection.subscribe(new Topic[]{new Topic(TOPIC, QoS.EXACTLY_ONCE)});
  
        // publish non-retained messages
        final int TOTAL_MESSAGES = 10;
        for (int i = 0; i < TOTAL_MESSAGES; i++) {
           connection.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, false);
        }
  
        // receive half the messages in this session
        for (int i = 0; i < TOTAL_MESSAGES / 2; i++) {
           final Message msg = connection.receive(1000, TimeUnit.MILLISECONDS);
           assertNotNull(msg);
           assertEquals(TOPIC, new String(msg.getPayload()));
           msg.ack();
        }
  
        connection.disconnect();
        // resume session
        connection = mqtt.blockingConnection();
        connection.connect();
        // receive rest of the messages
        Message msg = null;
        do {
           msg = connection.receive(1000, TimeUnit.MILLISECONDS);
           if (msg != null) {
              assertEquals(TOPIC, new String(msg.getPayload()));
              msg.ack();
           }
        }
        while (msg != null);
  
        // make sure we received all message ids
        for (short id = 1; id <= TOTAL_MESSAGES; id++) {
           assertNotNull(publishMap.get(id), "No message for id " + id);
        }
  
        connection.unsubscribe(new String[]{TOPIC});
        connection.disconnect();
     }
  
     @Disabled
     @Test
     @Timeout(90)
     // TODO ActiveMQ Classic does not reset the message id generator even after a clean session.  In Artemis we always reset.
     // If there is a good reason for this we should follow ActiveMQ.
     public void testPacketIdGeneratorCleanSession() throws Exception {
        final String[] cleanClientIds = new String[]{"", "clean-packetid", null};
        final Map<Short, PUBLISH> publishMap = new ConcurrentHashMap<>();
        MQTT[] mqtts = new MQTT[cleanClientIds.length];
        for (int i = 0; i < cleanClientIds.length; i++) {
           mqtts[i] = createMQTTConnection("", true);
           mqtts[i].setKeepAlive((short) 15);
  
           mqtts[i].setTracer(new Tracer() {
              @Override
              public void onReceive(MQTTFrame frame) {
                 logger.debug("Client received:\n{}", frame);
                 if (frame.messageType() == PUBLISH.TYPE) {
                    PUBLISH publish = new PUBLISH();
                    try {
                       publish.decode(frame);
                       logger.debug("PUBLISH {}", publish);
                    } catch (ProtocolException e) {
                       fail("Error decoding publish " + e.getMessage());
                    }
                    if (publishMap.get(publish.messageId()) != null) {
                       assertTrue(publish.dup());
                    }
                    publishMap.put(publish.messageId(), publish);
                 }
              }
  
              @Override
              public void onSend(MQTTFrame frame) {
                 logger.debug("Client sent:\n{}", frame);
              }
           });
        }
  
        final Random random = new Random();
        for (short i = 0; i < 10; i++) {
           BlockingConnection connection = mqtts[random.nextInt(cleanClientIds.length)].blockingConnection();
           connection.connect();
           final String TOPIC = "TopicA/";
           connection.subscribe(new Topic[]{new Topic(TOPIC, QoS.EXACTLY_ONCE)});
  
           // publish non-retained message
           connection.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, false);
           Message msg = connection.receive(1000, TimeUnit.MILLISECONDS);
           assertNotNull(msg);
           assertEquals(TOPIC, new String(msg.getPayload()));
           msg.ack();
  
           assertEquals(1, publishMap.size());
           final short id = (short) (i + 1);
           assertNotNull(publishMap.get(id), "No message for id " + id);
           publishMap.clear();
  
           connection.disconnect();
        }
  
     }
  
     @Test
     @Timeout(60)
     public void testClientConnectionFailure() throws Exception {
        MQTT mqtt = createMQTTConnection("reconnect", false);
        mqtt.setKeepAlive((short) 1);
  
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        Wait.waitFor(() -> connection.isConnected());
  
        final String TOPIC = "TopicA";
        final byte[] qos = connection.subscribe(new Topic[]{new Topic(TOPIC, QoS.EXACTLY_ONCE)});
        assertEquals(QoS.EXACTLY_ONCE.ordinal(), qos[0]);
        connection.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, false);
        // kill transport
        connection.kill();
  
        final BlockingConnection newConnection = mqtt.blockingConnection();
        newConnection.connect();
        Wait.waitFor(() -> newConnection.isConnected());
  
        assertEquals(QoS.EXACTLY_ONCE.ordinal(), qos[0]);
        Message msg = newConnection.receive(1000, TimeUnit.MILLISECONDS);
        assertNotNull(msg);
        assertEquals(TOPIC, new String(msg.getPayload()));
        msg.ack();
        newConnection.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testClientConnectionFailureSendsWillMessage() throws Exception {
        getServer().createQueue(QueueConfiguration.of("will"));
  
        MQTT mqtt = createMQTTConnection("1", false);
        mqtt.setKeepAlive((short) 1);
        mqtt.setWillMessage("test message");
        mqtt.setWillTopic("will");
        mqtt.setWillQos(QoS.AT_LEAST_ONCE);
  
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        Wait.waitFor(() -> connection.isConnected());
  
        MQTT mqtt2 = createMQTTConnection("2", false);
        BlockingConnection connection2 = mqtt2.blockingConnection();
        connection2.connect();
        connection2.subscribe(new Topic[]{new Topic("will", QoS.AT_LEAST_ONCE)});
  
        // kill transport
        connection.kill();
  
        Message m = connection2.receive(1000, TimeUnit.MILLISECONDS);
        assertNotNull(m);
        assertEquals("test message", new String(m.getPayload()));
     }
  
     @Test
     @Timeout(60)
     public void testWillMessageIsRetained() throws Exception {
        getServer().createQueue(QueueConfiguration.of("will"));
  
        MQTT mqtt = createMQTTConnection("1", false);
        mqtt.setKeepAlive((short) 1);
        mqtt.setWillMessage("test message");
        mqtt.setWillTopic("will");
        mqtt.setWillQos(QoS.AT_LEAST_ONCE);
        mqtt.setWillRetain(true);
  
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        Wait.waitFor(() -> connection.isConnected());
  
        // kill transport
        connection.kill();
  
        Thread.sleep(10000);
  
        MQTT mqtt2 = createMQTTConnection("2", false);
        BlockingConnection connection2 = mqtt2.blockingConnection();
        connection2.connect();
        connection2.subscribe(new Topic[]{new Topic("will", QoS.AT_LEAST_ONCE)});
  
        Message m = connection2.receive(1000, TimeUnit.MILLISECONDS);
        assertNotNull(m);
        m.ack();
        assertEquals("test message", new String(m.getPayload()));
     }
  
     @Test
     @Timeout(60)
     public void testCleanSessionForSubscriptions() throws Exception {
        final String CLIENTID = "cleansession";
        final MQTT mqttNotClean = createMQTTConnection(CLIENTID, false);
        BlockingConnection notClean = mqttNotClean.blockingConnection();
        final String TOPIC = "TopicA";
        notClean.connect();
        notClean.subscribe(new Topic[]{new Topic(TOPIC, QoS.EXACTLY_ONCE)});
        notClean.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, false);
        notClean.disconnect();
  
        assertEquals(1, getSessions().size());
  
        // MUST receive message from existing subscription from previous not clean session
        notClean = mqttNotClean.blockingConnection();
        notClean.connect();
        Message msg = notClean.receive(10000, TimeUnit.MILLISECONDS);
        assertNotNull(msg);
        assertEquals(TOPIC, new String(msg.getPayload()));
        msg.ack();
        notClean.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, false);
        notClean.disconnect();
  
        assertEquals(1, getSessions().size());
  
        // MUST NOT receive message from previous not clean session as existing subscription should be gone
        final MQTT mqttClean = createMQTTConnection(CLIENTID, true);
        final BlockingConnection clean = mqttClean.blockingConnection();
        clean.connect();
        msg = clean.receive(100, TimeUnit.MILLISECONDS);
        assertNull(msg);
        clean.subscribe(new Topic[]{new Topic(TOPIC, QoS.EXACTLY_ONCE)});
        clean.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, false);
        clean.disconnect();
  
        assertEquals(0, getSessions().size());
  
        // MUST NOT receive message from previous clean session as existing subscription should be gone
        notClean = mqttNotClean.blockingConnection();
        notClean.connect();
        msg = notClean.receive(100, TimeUnit.MILLISECONDS);
        assertNull(msg);
        notClean.disconnect();
  
        assertEquals(1, getSessions().size());
     }
  
     @Test
     @Timeout(60)
     public void testCleanSessionForMessages() throws Exception {
        final String CLIENTID = "cleansession";
        final MQTT mqttNotClean = createMQTTConnection(CLIENTID, false);
        BlockingConnection notClean = mqttNotClean.blockingConnection();
        final String TOPIC = "TopicA";
        notClean.connect();
        notClean.subscribe(new Topic[]{new Topic(TOPIC, QoS.EXACTLY_ONCE)});
        notClean.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, false);
        notClean.disconnect();
  
        assertEquals(1, getSessions().size());
  
        // MUST NOT receive message from previous not clean session even when creating a new subscription
        final MQTT mqttClean = createMQTTConnection(CLIENTID, true);
        final BlockingConnection clean = mqttClean.blockingConnection();
        clean.connect();
        clean.subscribe(new Topic[]{new Topic(TOPIC, QoS.EXACTLY_ONCE)});
        Message msg = clean.receive(100, TimeUnit.MILLISECONDS);
        assertNull(msg);
        clean.publish(TOPIC, TOPIC.getBytes(), QoS.EXACTLY_ONCE, false);
        clean.disconnect();
  
        assertEquals(0, getSessions().size());
     }
  
     @Test
     @Timeout(60)
     public void testSendMQTTReceiveJMS() throws Exception {
        doTestSendMQTTReceiveJMS("foo.*", "foo/bar");
     }
  
     @Test
     @Timeout(60)
     public void testLinkRouteAmqpReceiveMQTT() throws Exception {
  
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("TestClient");
        BlockingConnection blockingConnection = mqtt.blockingConnection();
        blockingConnection.connect();
        Topic t = new Topic("test", QoS.AT_LEAST_ONCE);
        blockingConnection.subscribe(new Topic[]{t});
  
        AmqpClient client = new AmqpClient(new URI(AMQP_URI), null, null);
        AmqpConnection connection = client.connect();
  
        try {
           AmqpSession session = connection.createSession();
           AmqpSender sender = session.createSender("test", true);
           AmqpMessage message = new AmqpMessage();
           message.setText("Test-Message");
           sender.send(message);
           sender.close();
        } finally {
           connection.close();
        }
  
        try {
           blockingConnection.subscribe(new Topic[]{t});
           assertNotNull(blockingConnection.receive(5, TimeUnit.SECONDS));
        } finally {
           blockingConnection.kill();
        }
     }
  
     public void doTestSendMQTTReceiveJMS(String jmsTopicAddress, String mqttAddress) throws Exception {
        final MQTTClientProvider provider = getMQTTClientProvider();
        initializeConnection(provider);
  
        // send retained message
        final String address = mqttAddress;
        final String RETAINED = "RETAINED";
  
        final byte[] payload = RETAINED.getBytes();
  
        Connection connection = cf.createConnection();
        // MUST set to true to receive retained messages
        connection.start();
  
        Session s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        javax.jms.Topic jmsTopic = s.createTopic(jmsTopicAddress);
        MessageConsumer consumer = s.createConsumer(jmsTopic);
  
        provider.publish(address, RETAINED.getBytes(), AT_LEAST_ONCE, true);
  
        // check whether we received retained message on JMS subscribe
        BytesMessage message = (BytesMessage) consumer.receive(5000);
        assertNotNull(message, "Should get retained message");
  
        byte[] b = new byte[8];
        message.readBytes(b);
        assertArrayEquals(payload, b);
  
        for (int i = 0; i < NUM_MESSAGES; i++) {
           String p = "Test Message: " + i;
           provider.publish(address, p.getBytes(), AT_LEAST_ONCE);
           message = (BytesMessage) consumer.receive(5000);
           assertNotNull(message, "Should get a message");
  
           byte[] bytePayload = new byte[p.getBytes().length];
           message.readBytes(bytePayload);
           assertArrayEquals(payload, b);
        }
  
        connection.close();
        provider.disconnect();
     }
  
     @Test
     @Timeout(120)
     public void testSendJMSReceiveMQTT() throws Exception {
        doTestSendJMSReceiveMQTT("foo.far");
     }
  
     public void doTestSendJMSReceiveMQTT(String destinationName) throws Exception {
        final MQTTClientProvider provider = getMQTTClientProvider();
        initializeConnection(provider);
        provider.subscribe("foo/+", AT_MOST_ONCE);
  
        Connection connection = cf.createConnection();
        connection.start();
  
        Session s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        javax.jms.Topic topic = s.createTopic(destinationName);
        MessageProducer producer = s.createProducer(topic);
  
        // send retained message from JMS
        final byte[] bytes = new byte[]{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
        BytesMessage bytesMessage = s.createBytesMessage();
        bytesMessage.writeBytes(bytes);
        producer.send(bytesMessage);
  
        byte[] message = provider.receive(10000);
        assertNotNull(message, "Should get retained message");
        assertArrayEquals(bytes, message);
  
        provider.disconnect();
        connection.close();
     }
  
     @Test
     @Timeout(60)
     public void testPingKeepsInactivityMonitorAlive() throws Exception {
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("foo");
        mqtt.setKeepAlive((short) 2);
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
  
        assertTrue(Wait.waitFor(() -> connection.isConnected()), "KeepAlive didn't work properly");
  
        connection.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testTurnOffInactivityMonitor() throws Exception {
        stopBroker();
        protocolConfig = "transport.useInactivityMonitor=false";
        startBroker();
  
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("foo3");
        mqtt.setKeepAlive((short) 2);
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
  
        assertTrue(Wait.waitFor(() -> connection.isConnected()), "KeepAlive didn't work properly");
  
        connection.disconnect();
     }
  
     @Disabled
     @Test
     @Timeout(60)
     // TODO Make dollar topics configurable in code base.
     public void testPublishDollarTopics() throws Exception {
        MQTT mqtt = createMQTTConnection();
        final String clientId = "publishDollar";
        mqtt.setClientId(clientId);
        mqtt.setKeepAlive((short) 2);
        BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
  
        final String DOLLAR_TOPIC = "$TopicA";
        connection.subscribe(new Topic[]{new Topic(DOLLAR_TOPIC, QoS.EXACTLY_ONCE)});
        connection.publish(DOLLAR_TOPIC, DOLLAR_TOPIC.getBytes(), QoS.EXACTLY_ONCE, true);
  
        Message message = connection.receive(100, TimeUnit.MILLISECONDS);
        assertNull(message, "Publish enabled for $ Topics by default");
        connection.disconnect();
  
        stopBroker();
        protocolConfig = "transport.publishDollarTopics=true";
        startBroker();
  
        mqtt = createMQTTConnection();
        mqtt.setClientId(clientId);
        mqtt.setKeepAlive((short) 2);
        connection = mqtt.blockingConnection();
        connection.connect();
  
        connection.subscribe(new Topic[]{new Topic(DOLLAR_TOPIC, QoS.EXACTLY_ONCE)});
        connection.publish(DOLLAR_TOPIC, DOLLAR_TOPIC.getBytes(), QoS.EXACTLY_ONCE, true);
  
        message = connection.receive(10, TimeUnit.SECONDS);
        assertNotNull(message);
        message.ack();
        assertEquals(DOLLAR_TOPIC, new String(message.getPayload()), "Message body");
  
        connection.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testDuplicateClientId() throws Exception {
        final String clientId = "duplicateClient";
        MQTT mqtt = createMQTTConnection(clientId, false);
        mqtt.setKeepAlive((short) 2);
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        final String TOPICA = "TopicA";
        connection.publish(TOPICA, TOPICA.getBytes(), QoS.EXACTLY_ONCE, true);
  
        MQTT mqtt1 = createMQTTConnection(clientId, false);
        mqtt1.setKeepAlive((short) 2);
        final BlockingConnection connection1 = mqtt1.blockingConnection();
        connection1.connect();
  
        assertTrue(Wait.waitFor(() -> connection1.isConnected()), "Duplicate client disconnected");
  
        assertTrue(Wait.waitFor(() -> !connection.isConnected()), "Old client still connected");
  
        connection1.publish(TOPICA, TOPICA.getBytes(), QoS.EXACTLY_ONCE, true);
        connection1.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testRepeatedLinkStealing() throws Exception {
        final String clientId = "duplicateClient";
        final AtomicReference<BlockingConnection> oldConnection = new AtomicReference<>();
        final String TOPICA = "TopicA";
  
        for (int i = 1; i <= 10; ++i) {
  
           logger.debug("Creating MQTT Connection {}", i);
  
           MQTT mqtt = createMQTTConnection(clientId, false);
           mqtt.setKeepAlive((short) 2);
           final BlockingConnection connection = mqtt.blockingConnection();
           connection.connect();
           connection.publish(TOPICA, TOPICA.getBytes(), QoS.EXACTLY_ONCE, true);
  
           assertTrue(Wait.waitFor(() -> connection.isConnected(), 3000, 200), "Client connect failed for attempt: " + i);
  
           if (oldConnection.get() != null) {
              assertTrue(Wait.waitFor(() -> !oldConnection.get().isConnected(), 3000, 200), "Old client still connected on attempt: " + i);
           }
  
           oldConnection.set(connection);
        }
  
        oldConnection.get().publish(TOPICA, TOPICA.getBytes(), QoS.EXACTLY_ONCE, true);
        oldConnection.get().disconnect();
     }
  
     @Test
     @Timeout(30)
     public void testJmsMapping() throws Exception {
        doTestJmsMapping("test.foo");
     }
  
     // TODO As with other tests, this should be enabled as part of the cross protocol support with MQTT.
     public void doTestJmsMapping(String destinationName) throws Exception {
        // start up jms consumer
        Connection jmsConn = cf.createConnection();
        Session session = jmsConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination dest = session.createQueue(destinationName);
        MessageConsumer consumer = session.createConsumer(dest);
        jmsConn.start();
  
        // set up mqtt producer
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("foo3");
        mqtt.setKeepAlive((short) 2);
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
  
        int messagesToSend = 5;
  
        // publish
        for (int i = 0; i < messagesToSend; ++i) {
           connection.publish("test/foo", "hello world".getBytes(), QoS.AT_LEAST_ONCE, false);
        }
  
        connection.disconnect();
  
        for (int i = 0; i < messagesToSend; i++) {
  
           javax.jms.Message message = consumer.receive(2 * 1000);
           assertNotNull(message);
           assertInstanceOf(BytesMessage.class, message);
           BytesMessage bytesMessage = (BytesMessage) message;
  
           int length = (int) bytesMessage.getBodyLength();
           byte[] buffer = new byte[length];
           bytesMessage.readBytes(buffer);
           assertEquals("hello world", new String(buffer));
        }
  
        jmsConn.close();
     }
  
     @Test
     @Timeout(30)
     public void testSubscribeMultipleTopics() throws Exception {
  
        byte[] payload = new byte[1024 * 32];
        for (int i = 0; i < payload.length; i++) {
           payload[i] = '2';
        }
  
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("MQTT-Client");
        mqtt.setCleanSession(false);
  
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
  
        Topic[] topics = {new Topic("Topic/A", QoS.EXACTLY_ONCE), new Topic("Topic/B", QoS.EXACTLY_ONCE)};
        Topic[] wildcardTopic = {new Topic("Topic/#", QoS.AT_LEAST_ONCE)};
        connection.subscribe(wildcardTopic);
  
        for (Topic topic : topics) {
           connection.publish(topic.name().toString(), payload, QoS.AT_LEAST_ONCE, false);
        }
  
        int received = 0;
        for (int i = 0; i < topics.length; ++i) {
           Message message = connection.receive();
           assertNotNull(message);
           received++;
           payload = message.getPayload();
           String messageContent = new String(payload);
           logger.debug("Received message from topic: {} Message content: {}", message.getTopic(), messageContent);
           message.ack();
        }
  
        assertEquals(topics.length, received, "Should have received " + topics.length + " messages");
     }
  
     @Test
     @Timeout(60)
     public void testReceiveMessageSentWhileOffline() throws Exception {
        final byte[] payload = new byte[1024 * 32];
        for (int i = 0; i < payload.length; i++) {
           payload[i] = '2';
        }
  
        int numberOfRuns = 100;
        int messagesPerRun = 2;
  
        final MQTT mqttPub = createMQTTConnection("MQTT-Pub-Client", true);
        final MQTT mqttSub = createMQTTConnection("MQTT-Sub-Client", false);
  
        final BlockingConnection connectionPub = mqttPub.blockingConnection();
        connectionPub.connect();
  
        BlockingConnection connectionSub = mqttSub.blockingConnection();
        connectionSub.connect();
  
        Topic[] topics = {new Topic("TopicA", QoS.EXACTLY_ONCE)};
        connectionSub.subscribe(topics);
  
        for (int i = 0; i < messagesPerRun; ++i) {
           connectionPub.publish(topics[0].name().toString(), payload, QoS.AT_LEAST_ONCE, false);
        }
  
        int received = 0;
        for (int i = 0; i < messagesPerRun; ++i) {
           Message message = connectionSub.receive(5, TimeUnit.SECONDS);
           assertNotNull(message);
           received++;
           assertArrayEquals(payload, message.getPayload());
           message.ack();
        }
        connectionSub.disconnect();
  
        for (int j = 0; j < numberOfRuns; j++) {
  
           for (int i = 0; i < messagesPerRun; ++i) {
              connectionPub.publish(topics[0].name().toString(), payload, QoS.AT_LEAST_ONCE, false);
           }
  
           connectionSub = mqttSub.blockingConnection();
           connectionSub.connect();
           connectionSub.subscribe(topics);
  
           for (int i = 0; i < messagesPerRun; ++i) {
              Message message = connectionSub.receive(5, TimeUnit.SECONDS);
              assertNotNull(message);
              received++;
              assertArrayEquals(payload, message.getPayload());
              message.ack();
           }
           connectionSub.disconnect();
        }
        assertEquals((messagesPerRun * (numberOfRuns + 1)), received, "Should have received " + (messagesPerRun * (numberOfRuns + 1)) + " messages");
     }
  
     @Test
     @Timeout(60)
     public void testDefaultSessionExpiryInterval() throws Exception {
        final MQTT mqttSub = createMQTTConnection("MQTT-Sub-Client", false);
  
        BlockingConnection connectionSub = mqttSub.blockingConnection();
        connectionSub.connect();
  
        assertEquals(1, getSessions().size());
  
        Topic[] topics = {new Topic("TopicA", QoS.EXACTLY_ONCE)};
        connectionSub.subscribe(topics);
        connectionSub.disconnect();
  
        // session shouldn't expire by default
        Wait.assertEquals(1, () -> getSessions().size(), 10000, 100);
     }
  
     @Test
     @Timeout(30)
     public void testDefaultKeepAliveWhenClientSpecifiesZero() throws Exception {
        stopBroker();
        protocolConfig = "transport.defaultKeepAlive=2000";
        startBroker();
  
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("foo");
        mqtt.setKeepAlive((short) 0);
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
  
        assertTrue(Wait.waitFor(() -> connection.isConnected()), "KeepAlive didn't work properly");
     }
  
     @Test
     @Timeout(60)
     public void testReuseConnection() throws Exception {
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("Test-Client");
  
        {
           BlockingConnection connection = mqtt.blockingConnection();
           connection.connect();
           connection.disconnect();
           Thread.sleep(100);
        }
        {
           BlockingConnection connection = mqtt.blockingConnection();
           connection.connect();
           connection.disconnect();
           Thread.sleep(100);
        }
     }
  
     @Test
     @Timeout(60)
     public void testNoMessageReceivedAfterUnsubscribeMQTT() throws Exception {
        Topic[] topics = {new Topic("TopicA", QoS.EXACTLY_ONCE)};
  
        MQTT mqttPub = createMQTTConnection("MQTTPub-Client", true);
        // mqttPub.setVersion("3.1.1");
  
        MQTT mqttSub = createMQTTConnection("MQTTSub-Client", false);
        // mqttSub.setVersion("3.1.1");
  
        BlockingConnection connectionPub = mqttPub.blockingConnection();
        connectionPub.connect();
  
        BlockingConnection connectionSub = mqttSub.blockingConnection();
        connectionSub.connect();
        connectionSub.subscribe(topics);
        connectionSub.disconnect();
  
        for (int i = 0; i < 5; i++) {
           String payload = "Message " + i;
           connectionPub.publish(topics[0].name().toString(), payload.getBytes(), QoS.EXACTLY_ONCE, false);
        }
  
        connectionSub = mqttSub.blockingConnection();
        connectionSub.connect();
  
        int received = 0;
        for (int i = 0; i < 5; ++i) {
           Message message = connectionSub.receive(5, TimeUnit.SECONDS);
           assertNotNull(message, "Missing message " + i);
           logger.debug("Message is {}", new String(message.getPayload()));
           received++;
           message.ack();
        }
        assertEquals(5, received);
  
        // unsubscribe from topic
        connectionSub.unsubscribe(new String[]{"TopicA"});
  
        // send more messages
        for (int i = 0; i < 5; i++) {
           String payload = "Message " + i;
           connectionPub.publish(topics[0].name().toString(), payload.getBytes(), QoS.EXACTLY_ONCE, false);
        }
  
        // these should not be received
        assertNull(connectionSub.receive(100, TimeUnit.MILLISECONDS));
  
        connectionSub.disconnect();
        connectionPub.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testMQTT311Connection() throws Exception {
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("foo");
        mqtt.setVersion("3.1.1");
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        connection.disconnect();
     }
  
     // TODO This should be reworked to align with Artemis recovery.
     //   @Test(timeout = 60 * 1000)
     //   public void testActiveMQRecoveryPolicy() throws Exception {
     //      // test with ActiveMQ LastImageSubscriptionRecoveryPolicy
     //      final PolicyMap policyMap = new PolicyMap();
     //      final PolicyEntry policyEntry = new PolicyEntry();
     //      policyEntry.setSubscriptionRecoveryPolicy(new LastImageSubscriptionRecoveryPolicy());
     //      policyMap.put(new ActiveMQTopic(">"), policyEntry);
     //      brokerService.setDestinationPolicy(policyMap);
     //
     //      MQTT mqtt = createMQTTConnection("pub-sub", true);
     //      final int[] retain = new int[1];
     //      final int[] nonretain  = new int[1];
     //      mqtt.setTracer(new Tracer() {
     //         @Override
     //         public void onReceive(MQTTFrame frame) {
     //            if (frame.messageType() == PUBLISH.TYPE) {
     //               logger.debug("Received message with retain={}", frame.retain());
     //               if (frame.retain()) {
     //                  retain[0]++;
     //               } else {
     //                  nonretain[0]++;
     //               }
     //            }
     //         }
     //      });
     //
     //      BlockingConnection connection = mqtt.blockingConnection();
     //      connection.connect();
     //      final String RETAINED = "RETAINED";
     //      connection.publish("one", RETAINED.getBytes(), QoS.AT_LEAST_ONCE, true);
     //      connection.publish("two", RETAINED.getBytes(), QoS.AT_LEAST_ONCE, true);
     //
     //      final String NONRETAINED = "NONRETAINED";
     //      connection.publish("one", NONRETAINED.getBytes(), QoS.AT_LEAST_ONCE, false);
     //      connection.publish("two", NONRETAINED.getBytes(), QoS.AT_LEAST_ONCE, false);
     //
     //      connection.subscribe(new Topic[]{new Topic("#", QoS.AT_LEAST_ONCE)});
     //      for (int i = 0; i < 4; i++) {
     //         final Message message = connection.receive(30, TimeUnit.SECONDS);
     //         assertNotNull("Should receive 4 messages", message);
     //         message.ack();
     //      }
     //      assertEquals("Should receive 2 retained messages", 2, retain[0]);
     //      assertEquals("Should receive 2 non-retained messages", 2, nonretain[0]);
     //   }
  
     // TODO As with other tests, this should be enabled as part of the cross protocol support with MQTT.
     //   @Test(timeout = 60 * 1000)
     //   public void testSendMQTTReceiveJMSVirtualTopic() throws Exception {
     //
     //      final MQTTClientProvider provider = getMQTTClientProvider();
     //      initializeConnection(provider);
     //      final String DESTINATION_NAME = "Consumer.jms.VirtualTopic.TopicA";
     //
     //      // send retained message
     //      final String RETAINED = "RETAINED";
     //      final String MQTT_DESTINATION_NAME = "VirtualTopic/TopicA";
     //      provider.publish(MQTT_DESTINATION_NAME, RETAINED.getBytes(), AT_LEAST_ONCE, true);
     //
     //      ActiveMQConnection activeMQConnection = (ActiveMQConnection) new ActiveMQConnectionFactory(jmsUri).createConnection();
     //      // MUST set to true to receive retained messages
     //      activeMQConnection.setUseRetroactiveConsumer(true);
     //      activeMQConnection.start();
     //      Session s = activeMQConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
     //      Queue jmsQueue = s.createQueue(DESTINATION_NAME);
     //      MessageConsumer consumer = s.createConsumer(jmsQueue);
     //
     //      // check whether we received retained message on JMS subscribe
     //      ActiveMQMessage message = (ActiveMQMessage) consumer.receive(5000);
     //      assertNotNull("Should get retained message", message);
     //      ByteSequence bs = message.getContent();
     //      assertEquals(RETAINED, new String(bs.data, bs.offset, bs.length));
     //      assertTrue(message.getBooleanProperty(RetainedMessageSubscriptionRecoveryPolicy.RETAINED_PROPERTY));
     //
     //      for (int i = 0; i < NUM_MESSAGES; i++) {
     //         String payload = "Test Message: " + i;
     //         provider.publish(MQTT_DESTINATION_NAME, payload.getBytes(), AT_LEAST_ONCE);
     //         message = (ActiveMQMessage) consumer.receive(5000);
     //         assertNotNull("Should get a message", message);
     //         bs = message.getContent();
     //         assertEquals(payload, new String(bs.data, bs.offset, bs.length));
     //      }
     //
     //      // re-create consumer and check we received retained message again
     //      consumer.close();
     //      consumer = s.createConsumer(jmsQueue);
     //      message = (ActiveMQMessage) consumer.receive(5000);
     //      assertNotNull("Should get retained message", message);
     //      bs = message.getContent();
     //      assertEquals(RETAINED, new String(bs.data, bs.offset, bs.length));
     //      assertTrue(message.getBooleanProperty(RetainedMessageSubscriptionRecoveryPolicy.RETAINED_PROPERTY));
     //
     //      activeMQConnection.close();
     //      provider.disconnect();
     //   }
  
     @Test
     @Timeout(60)
     public void testPingOnMQTT() throws Exception {
        stopBroker();
        protocolConfig = "maxInactivityDuration=-1";
        startBroker();
  
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId("test-mqtt");
        mqtt.setKeepAlive((short) 2);
        final BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        assertTrue(Wait.waitFor(() -> connection.isConnected()), "KeepAlive didn't work properly");
  
        connection.disconnect();
     }
  
     @Test
     @Timeout(60)
     public void testClientDisconnectedOnMaxConsumerLimitReached() throws Exception {
        Exception peerDisconnectedException = null;
        try {
           String clientId = "test.client";
           String coreAddress = MQTTUtil.getCoreAddressFromMqttTopic("foo/bar", server.getConfiguration().getWildcardConfiguration());
           Topic[] mqttSubscription = new Topic[]{new Topic("foo/bar", QoS.AT_LEAST_ONCE)};
  
           getServer().createQueue(QueueConfiguration.of(MQTTUtil.getCoreQueueFromMqttTopic("foo/bar", clientId, server.getConfiguration().getWildcardConfiguration())).setAddress(coreAddress).setRoutingType(RoutingType.MULTICAST).setDurable(false).setTemporary(true).setMaxConsumers(0));
  
           MQTT mqtt = createMQTTConnection();
           mqtt.setClientId(clientId);
           mqtt.setKeepAlive((short) 2);
           final BlockingConnection connection = mqtt.blockingConnection();
           connection.connect();
           connection.subscribe(mqttSubscription);
        } catch (EOFException e) {
           peerDisconnectedException = e;
        }
        assertNotNull(peerDisconnectedException);
        assertTrue(peerDisconnectedException.getMessage().contains("Peer disconnected"));
     }
  
     @Test
     @Timeout(60)
     public void testAnycastPrefixWorksWithMQTT() throws Exception {
        String clientId = "testMqtt";
  
        String anycastAddress = "anycast:foo/bar";
        String sendAddress = "foo/bar";
        Topic[] mqttSubscription = new Topic[]{new Topic(anycastAddress, QoS.AT_LEAST_ONCE)};
  
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId(clientId);
        BlockingConnection connection1 = mqtt.blockingConnection();
        connection1.connect();
        connection1.subscribe(mqttSubscription);
  
        MQTT mqtt2 = createMQTTConnection();
        mqtt2.setClientId(clientId + "2");
        BlockingConnection connection2 = mqtt2.blockingConnection();
        connection2.connect();
        connection2.subscribe(mqttSubscription);
  
        String message1 = "TestMessage1";
        String message2 = "TestMessage2";
  
        connection1.publish(sendAddress, message1.getBytes(), QoS.AT_LEAST_ONCE, false);
        connection2.publish(sendAddress, message2.getBytes(), QoS.AT_LEAST_ONCE, false);
  
        assertNotNull(connection1.receive(1000, TimeUnit.MILLISECONDS));
        assertNull(connection1.receive(100, TimeUnit.MILLISECONDS));
  
        assertNotNull(connection2.receive(1000, TimeUnit.MILLISECONDS));
        assertNull(connection2.receive(100, TimeUnit.MILLISECONDS));
     }
  
     @Test
     @Timeout(60)
     public void testAnycastAddressWorksWithMQTT() throws Exception {
        String anycastAddress = "foo/bar";
  
        getServer().addAddressInfo(new AddressInfo(SimpleString.of("foo.bar"), RoutingType.ANYCAST));
        String clientId = "testMqtt";
  
        Topic[] mqttSubscription = new Topic[]{new Topic(anycastAddress, QoS.AT_LEAST_ONCE)};
  
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId(clientId);
        BlockingConnection connection1 = mqtt.blockingConnection();
        connection1.connect();
        connection1.subscribe(mqttSubscription);
  
        MQTT mqtt2 = createMQTTConnection();
        mqtt2.setClientId(clientId + "2");
        BlockingConnection connection2 = mqtt2.blockingConnection();
        connection2.connect();
        connection2.subscribe(mqttSubscription);
  
        String message1 = "TestMessage1";
        String message2 = "TestMessage2";
  
        connection1.publish(anycastAddress, message1.getBytes(), QoS.AT_LEAST_ONCE, false);
        connection2.publish(anycastAddress, message2.getBytes(), QoS.AT_LEAST_ONCE, false);
  
        assertNotNull(connection1.receive(1000, TimeUnit.MILLISECONDS));
        assertNull(connection1.receive(100, TimeUnit.MILLISECONDS));
  
        assertNotNull(connection2.receive(1000, TimeUnit.MILLISECONDS));
        assertNull(connection2.receive(100, TimeUnit.MILLISECONDS));
     }
  
     @Test
     @Timeout(60)
     public void testAmbiguousRoutingWithMQTT() throws Exception {
        String anycastAddress = "foo/bar";
  
        EnumSet<RoutingType> routingTypeSet = EnumSet.of(RoutingType.ANYCAST, RoutingType.MULTICAST);
  
        getServer().addAddressInfo(new AddressInfo(SimpleString.of("foo.bar"), routingTypeSet));
        String clientId = "testMqtt";
  
        Topic[] mqttSubscription = new Topic[]{new Topic(anycastAddress, QoS.AT_LEAST_ONCE)};
  
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId(clientId);
        BlockingConnection connection1 = mqtt.blockingConnection();
        connection1.connect();
        connection1.subscribe(mqttSubscription);
  
        MQTT mqtt2 = createMQTTConnection();
        mqtt2.setClientId(clientId + "2");
        BlockingConnection connection2 = mqtt2.blockingConnection();
        connection2.connect();
        connection2.subscribe(mqttSubscription);
  
        String message1 = "TestMessage1";
        String message2 = "TestMessage2";
  
        connection1.publish(anycastAddress, message1.getBytes(), QoS.AT_LEAST_ONCE, false);
        connection2.publish(anycastAddress, message2.getBytes(), QoS.AT_LEAST_ONCE, false);
  
        assertNotNull(connection1.receive(1000, TimeUnit.MILLISECONDS));
        assertNotNull(connection1.receive(1000, TimeUnit.MILLISECONDS));
  
        assertNotNull(connection2.receive(1000, TimeUnit.MILLISECONDS));
        assertNotNull(connection2.receive(1000, TimeUnit.MILLISECONDS));
     }
  
     @Test
     public void testRetainedMessagesAreCorrectlyFormedAfterRestart() throws Exception {
        String clientId = "testMqtt";
        String address = "testAddress";
        String payload = "This is a test message";
  
        // Create address
        getServer().addAddressInfo(new AddressInfo(SimpleString.of(address), RoutingType.MULTICAST));
  
        // Send MQTT Retain Message
        Topic[] mqttTopic = new Topic[]{new Topic(address, QoS.AT_LEAST_ONCE)};
  
        MQTT mqtt = createMQTTConnection();
        mqtt.setClientId(clientId);
        BlockingConnection connection1 = mqtt.blockingConnection();
        connection1.connect();
        connection1.publish(address, payload.getBytes(), QoS.AT_LEAST_ONCE, true);
  
        getServer().fail(false);
        getServer().start();
        waitForServerToStart(getServer());
  
        MQTT mqtt2 = createMQTTConnection();
        mqtt2.setClientId(clientId + "2");
        BlockingConnection connection2 = mqtt2.blockingConnection();
        connection2.connect();
        connection2.subscribe(mqttTopic);
  
        Message message = connection2.receive(5000, TimeUnit.MILLISECONDS);
        assertEquals(payload, new String(message.getPayload()));
     }
  
     @Test
     @Timeout(60)
     public void testBrokerRestartAfterSubHashWithConfigurationQueues() throws Exception {
  
        // Add some pre configured queues
        CoreAddressConfiguration coreAddressConfiguration = new CoreAddressConfiguration();
        coreAddressConfiguration.setName("DLA");
        coreAddressConfiguration.addRoutingType(RoutingType.ANYCAST);
        coreAddressConfiguration.addQueueConfiguration(QueueConfiguration.of("DLQ")
                                                          .setAddress("DLA")
                                                          .setRoutingType(RoutingType.ANYCAST));
  
        getServer().getConfiguration().getAddressConfigurations().add(coreAddressConfiguration);
  
        getServer().stop();
        getServer().start();
        getServer().waitForActivation(10, TimeUnit.SECONDS);
  
        for (int i = 0; i < 2; i++) {
           MQTT mqtt = createMQTTConnection("myClient", false);
           BlockingConnection connection = mqtt.blockingConnection();
           connection.connect();
           connection.subscribe(new Topic[]{new Topic("#", QoS.AT_MOST_ONCE)});
           connection.disconnect();
  
           getServer().stop();
           getServer().start();
           getServer().waitForActivation(10, TimeUnit.SECONDS);
        }
  
     }
  
     @Test
     public void testDoubleBroker() throws Exception {
        /*
         * Start two embedded server instances for MQTT and connect to them
         * with the same MQTT client id. As those are two different instances
         * connecting to them with the same client ID must succeed.
         */
  
        final int port1 = 1884;
        final int port2 = 1885;
  
        final Configuration cfg1 = createDefaultConfig(1, false);
        cfg1.setResolveProtocols(true).addAcceptorConfiguration("mqtt1", "tcp://localhost:" + port1 + "?protocols=MQTT");
  
        final Configuration cfg2 = createDefaultConfig(2, false);
        cfg2.setResolveProtocols(true).addAcceptorConfiguration("mqtt2", "tcp://localhost:" + port2 + "?protocols=MQTT");
  
        final ActiveMQServer server1 = createServer(cfg1);
        server1.start();
        final ActiveMQServer server2 = createServer(cfg2);
        server2.start();
  
        final String clientId = "client1";
        final MQTT mqtt1 = createMQTTConnection(clientId, true);
        final MQTT mqtt2 = createMQTTConnection(clientId, true);
  
        mqtt1.setHost("localhost", port1);
        mqtt2.setHost("localhost", port2);
  
        final BlockingConnection connection1 = mqtt1.blockingConnection();
        final BlockingConnection connection2 = mqtt2.blockingConnection();
  
        try {
           connection1.connect();
           connection2.connect();
        } catch (Exception e) {
           fail("Connections should have worked.");
        } finally {
           if (connection1.isConnected())
              connection1.disconnect();
           if (connection2.isConnected())
              connection2.disconnect();
        }
     }
  
     @Test
     public void autoDestroyAddress() throws Exception {
        AddressSettings addressSettings = new AddressSettings();
        addressSettings.setAutoDeleteAddresses(true);
        server.getAddressSettingsRepository().addMatch("foo.bar", addressSettings);
  
        final MQTTClientProvider subscriptionProvider = getMQTTClientProvider();
        initializeConnection(subscriptionProvider);
        subscriptionProvider.subscribe("foo/bar", AT_MOST_ONCE);
        assertNotNull(server.getAddressInfo(SimpleString.of("foo.bar")));
  
        subscriptionProvider.disconnect();
  
        Wait.assertTrue(() -> server.getAddressInfo(SimpleString.of("foo.bar")) == null);
     }
  
     @Test
     @Timeout(60)
     public void testAutoDeleteRetainedQueue() throws Exception {
        final String TOPIC = "/abc/123";
        final String RETAINED_QUEUE = MQTTUtil.getCoreRetainAddressFromMqttTopic(TOPIC, server.getConfiguration().getWildcardConfiguration());
        final MQTTClientProvider publisher = getMQTTClientProvider();
        final MQTTClientProvider subscriber = getMQTTClientProvider();
  
        server.getAddressSettingsRepository().addMatch(MQTTUtil.getCoreAddressFromMqttTopic("#", server.getConfiguration().getWildcardConfiguration()), new AddressSettings().setExpiryDelay(500L).setAutoDeleteQueues(true).setAutoDeleteAddresses(true));
  
        initializeConnection(publisher);
        initializeConnection(subscriber);
  
        String RETAINED = "retained";
        publisher.publish(TOPIC, RETAINED.getBytes(), AT_LEAST_ONCE, true);
  
        List<String> messages = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
           messages.add("TEST MESSAGE:" + i);
        }
  
        subscriber.subscribe(TOPIC, AT_LEAST_ONCE);
  
        for (int i = 0; i < 10; i++) {
           publisher.publish(TOPIC, messages.get(i).getBytes(), AT_LEAST_ONCE);
        }
  
        byte[] msg = subscriber.receive(5000);
        assertNotNull(msg);
        assertEquals(RETAINED, new String(msg));
  
        for (int i = 0; i < 10; i++) {
           msg = subscriber.receive(5000);
           assertNotNull(msg);
           assertEquals(messages.get(i), new String(msg));
        }
  
        subscriber.disconnect();
        publisher.disconnect();
  
        Wait.assertTrue(() -> server.locateQueue(RETAINED_QUEUE).getMessageCount() == 0, 2000, 50);
        Wait.assertTrue(() -> server.locateQueue(RETAINED_QUEUE) == null, 2000, 50);
  
        // now that we auto-deleted do it again to ensure it is recreated and auto-deleted properly
  
        initializeConnection(publisher);
        initializeConnection(subscriber);
  
        publisher.publish(TOPIC, RETAINED.getBytes(), AT_LEAST_ONCE, true);
  
        subscriber.subscribe(TOPIC, AT_LEAST_ONCE);
  
        for (int i = 0; i < 10; i++) {
           publisher.publish(TOPIC, messages.get(i).getBytes(), AT_LEAST_ONCE);
        }
  
        msg = subscriber.receive(5000);
        assertNotNull(msg);
        assertEquals(RETAINED, new String(msg));
  
        for (int i = 0; i < 10; i++) {
           msg = subscriber.receive(5000);
           assertNotNull(msg);
           assertEquals(messages.get(i), new String(msg));
        }
  
        subscriber.disconnect();
        publisher.disconnect();
  
        Wait.assertTrue(() -> server.locateQueue(RETAINED_QUEUE).getMessageCount() == 0, 3000, 50);
        Wait.assertTrue(() -> server.locateQueue(RETAINED_QUEUE) == null, 3000, 50);
     }
  
     /*
      * [MQTT-3.3.1-9] When sending a PUBLISH Packet to a Client the Server...MUST set the RETAIN flag to 0 when a PUBLISH
      * Packet is sent to a Client because it matches an *established* subscription regardless of how the flag was set in
      * the message it received.
      */
     @Test
     @Timeout(60)
     public void testRetainFlagOnEstablishedSubscription() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);
        final String topic = getTopicName();
  
        MqttClient subscriber = createPaho3_1_1Client("subscriber");
        subscriber.setCallback(new DefaultMqtt3Callback() {
           @Override
           public void messageArrived(String topic, MqttMessage message) throws Exception {
              if (!message.isRetained()) {
                 latch.countDown();
              }
           }
        });
        subscriber.connect();
        subscriber.subscribe(topic, 1);
  
        MqttClient publisher = createPaho3_1_1Client("publisher");
        publisher.connect();
        publisher.publish(topic, "retained".getBytes(StandardCharsets.UTF_8), 1, true);
        publisher.disconnect();
        publisher.close();
  
        assertTrue(latch.await(1, TimeUnit.SECONDS), "Did not receive expected message within timeout");
  
        subscriber.disconnect();
        subscriber.close();
     }
  
     /*
      * [MQTT-3.3.1-8] When sending a PUBLISH Packet to a Client the Server MUST set the RETAIN flag to 1 if a message is
      * sent as a result of a new subscription being made by a Client.
      */
     @Test
     @Timeout(60)
     public void testRetainFlagOnNewSubscription() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);
        final String topic = getTopicName();
  
        MqttClient publisher = createPaho3_1_1Client("publisher");
        publisher.connect();
        publisher.publish(topic, "retained".getBytes(StandardCharsets.UTF_8), 1, true);
        publisher.disconnect();
        publisher.close();
  
        MqttClient subscriber = createPaho3_1_1Client("subscriber");
        subscriber.setCallback(new DefaultMqtt3Callback() {
           @Override
           public void messageArrived(String topic, MqttMessage message) throws Exception {
              if (message.isRetained()) {
                 latch.countDown();
              }
           }
        });
        subscriber.connect();
        subscriber.subscribe(topic, 1);
  
        assertTrue(latch.await(1, TimeUnit.SECONDS), "Did not receive expected message within timeout");
  
        subscriber.disconnect();
        subscriber.close();
     }
  }
  