https://issues.apache.org/jira/browse/AMQ-5290

Clean up some code to make future work a bit simpler.

activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolConverter.java

@@ -415,7 +415,6 @@ public class MQTTProtocolConverter {
 
         final String topicName = topic.name().toString();
         final QoS topicQoS = topic.qos();
-        ActiveMQDestination destination = new ActiveMQTopic(convertMQTTToActiveMQ(topicName));
 
         if (mqttSubscriptionByTopic.containsKey(topicName)) {
             final MQTTSubscription mqttSubscription = mqttSubscriptionByTopic.get(topicName);
@@ -424,11 +423,13 @@ public class MQTTProtocolConverter {
                 onUnSubscribe(topicName);
             } else {
                 // duplicate SUBSCRIBE packet, find all matching topics and re-send retained messages
-                resendRetainedMessages(topicName, destination, mqttSubscription);
+                resendRetainedMessages(mqttSubscription);
                 return (byte) topicQoS.ordinal();
             }
         }
 
+        ActiveMQDestination destination = new ActiveMQTopic(MQTTProtocolSupport.convertMQTTToActiveMQ(topicName));
+
         ConsumerId id = new ConsumerId(sessionId, consumerIdGenerator.getNextSequenceId());
         ConsumerInfo consumerInfo = new ConsumerInfo(id);
         consumerInfo.setDestination(destination);
@@ -439,7 +440,7 @@ public class MQTTProtocolConverter {
         if (!connect.cleanSession() && connect.clientId() != null && topicQoS.ordinal() >= QoS.AT_LEAST_ONCE.ordinal()) {
             consumerInfo.setSubscriptionName(topicQoS + ":" + topicName);
         }
-        MQTTSubscription mqttSubscription = new MQTTSubscription(this, topicQoS, consumerInfo);
+        MQTTSubscription mqttSubscription = new MQTTSubscription(this, topicName, topicQoS, consumerInfo);
 
         // optimistic add to local maps first to be able to handle commands in onActiveMQCommand
         subscriptionsByConsumerId.put(id, mqttSubscription);
@@ -469,14 +470,17 @@ public class MQTTProtocolConverter {
         return qos[0];
     }
 
-    private void resendRetainedMessages(String topicName, ActiveMQDestination destination,
+    private void resendRetainedMessages(MQTTSubscription mqttSubscription) throws MQTTProtocolException {
-                                        MQTTSubscription mqttSubscription) throws MQTTProtocolException {
+
+        ActiveMQDestination destination = mqttSubscription.getDestination();
+
         // check whether the Topic has been recovered in restoreDurableSubs
         // mark subscription available for recovery for duplicate subscription
         if (restoredSubs.remove(destination.getPhysicalName())) {
             return;
         }
 
+        String topicName = mqttSubscription.getTopicName();
         // get TopicRegion
         RegionBroker regionBroker;
         try {
@@ -550,7 +554,7 @@ public class MQTTProtocolConverter {
             // check if the durable sub also needs to be removed
             if (subs.getConsumerInfo().getSubscriptionName() != null) {
                 // also remove it from restored durable subscriptions set
-                restoredSubs.remove(convertMQTTToActiveMQ(topicName));
+                restoredSubs.remove(MQTTProtocolSupport.convertMQTTToActiveMQ(topicName));
 
                 RemoveSubscriptionInfo rsi = new RemoveSubscriptionInfo();
                 rsi.setConnectionId(connectionId);
@@ -680,7 +684,7 @@ public class MQTTProtocolConverter {
         synchronized (activeMQTopicMap) {
             topic = activeMQTopicMap.get(command.topicName());
             if (topic == null) {
-                String topicName = convertMQTTToActiveMQ(command.topicName().toString());
+                String topicName = MQTTProtocolSupport.convertMQTTToActiveMQ(command.topicName().toString());
                 topic = new ActiveMQTopic(topicName);
                 activeMQTopicMap.put(command.topicName().toString(), topic);
             }
@@ -710,7 +714,7 @@ public class MQTTProtocolConverter {
         synchronized (mqttTopicMap) {
             topicName = mqttTopicMap.get(message.getJMSDestination());
             if (topicName == null) {
-                topicName = convertActiveMQToMQTT(message.getDestination().getPhysicalName());
+                topicName = MQTTProtocolSupport.convertActiveMQToMQTT(message.getDestination().getPhysicalName());
                 mqttTopicMap.put(message.getJMSDestination(), topicName);
             }
         }
@@ -757,10 +761,6 @@ public class MQTTProtocolConverter {
         return result;
     }
 
-    private String convertActiveMQToMQTT(String physicalName) {
-        return physicalName.replace('.', '/');
-    }
-
     public MQTTTransport getMQTTTransport() {
         return mqttTransport;
     }
@@ -852,18 +852,6 @@ public class MQTTProtocolConverter {
         }
     }
 
-    String getClientId() {
-        if (clientId == null) {
-            if (connect != null && connect.clientId() != null) {
-                clientId = connect.clientId().toString();
-            }
-            else {
-                clientId = "";
-            }
-        }
-        return clientId;
-    }
-
     private void stopTransport() {
         try {
             getMQTTTransport().stop();
@@ -912,34 +900,6 @@ public class MQTTProtocolConverter {
         return null;
     }
 
-    private String convertMQTTToActiveMQ(String name) {
-        char[] chars = name.toCharArray();
-        for (int i = 0; i < chars.length; i++) {
-            switch(chars[i]) {
-                case '#':
-                    chars[i] = '>';
-                    break;
-                case '>':
-                    chars[i] = '#';
-                    break;
-                case '+':
-                    chars[i] = '*';
-                    break;
-                case '*':
-                    chars[i] = '+';
-                    break;
-                case '/':
-                    chars[i] = '.';
-                    break;
-                case '.':
-                    chars[i] = '/';
-                    break;
-            }
-        }
-        String rc = new String(chars);
-        return rc;
-    }
-
     public long getDefaultKeepAlive() {
         return defaultKeepAlive;
     }
@@ -979,4 +939,19 @@ public class MQTTProtocolConverter {
     public boolean getPublishDollarTopics() {
         return publishDollarTopics;
     }
+
+    public ConnectionId getConnectionId() {
+        return connectionId;
+    }
+
+    public String getClientId() {
+        if (clientId == null) {
+            if (connect != null && connect.clientId() != null) {
+                clientId = connect.clientId().toString();
+            } else {
+                clientId = "";
+            }
+        }
+        return clientId;
+    }
 }

activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolSupport.java

@@ -0,0 +1,73 @@
+/**
+ * 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.transport.mqtt;
+
+/**
+ * A set of static methods useful for handling MQTT based client connections.
+ */
+public class MQTTProtocolSupport {
+
+    /**
+     * Converts an MQTT formatted Topic name into a suitable ActiveMQ Destination
+     * name string.
+     *
+     * @param name
+     *        the MQTT formatted topic name.
+     *
+     * @return an destination name that fits the ActiveMQ conventions.
+     */
+    public static String convertMQTTToActiveMQ(String name) {
+        char[] chars = name.toCharArray();
+        for (int i = 0; i < chars.length; i++) {
+            switch(chars[i]) {
+                case '#':
+                    chars[i] = '>';
+                    break;
+                case '>':
+                    chars[i] = '#';
+                    break;
+                case '+':
+                    chars[i] = '*';
+                    break;
+                case '*':
+                    chars[i] = '+';
+                    break;
+                case '/':
+                    chars[i] = '.';
+                    break;
+                case '.':
+                    chars[i] = '/';
+                    break;
+            }
+        }
+        String rc = new String(chars);
+        return rc;
+    }
+
+    /**
+     * Converts an ActiveMQ destination name into a correctly formatted
+     * MQTT destination name.
+     *
+     * @param destinationName
+     *        the ActiveMQ destination name to process.
+     *
+     * @return a destination name formatted for MQTT.
+     */
+    public static String convertActiveMQToMQTT(String destinationName) {
+        return destinationName.replace('.', '/');
+    }
+}

activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTSubscription.java

@@ -18,6 +18,7 @@ package org.apache.activemq.transport.mqtt;
 
 import java.io.IOException;
 import java.util.zip.DataFormatException;
+
 import javax.jms.JMSException;
 
 import org.apache.activemq.command.ActiveMQDestination;
@@ -31,24 +32,47 @@ import org.fusesource.mqtt.codec.PUBLISH;
 /**
  * Keeps track of the MQTT client subscription so that acking is correctly done.
  */
-class MQTTSubscription {
+public class MQTTSubscription {
+
     private final MQTTProtocolConverter protocolConverter;
 
     private final ConsumerInfo consumerInfo;
-    private ActiveMQDestination destination;
+    private final String topicName;
     private final QoS qos;
 
-    public MQTTSubscription(MQTTProtocolConverter protocolConverter, QoS qos, ConsumerInfo consumerInfo) {
+    public MQTTSubscription(MQTTProtocolConverter protocolConverter, String topicName, QoS qos, ConsumerInfo consumerInfo) {
         this.protocolConverter = protocolConverter;
         this.consumerInfo = consumerInfo;
         this.qos = qos;
+        this.topicName = topicName;
     }
 
-    MessageAck createMessageAck(MessageDispatch md) {
+    /**
+     * Create a {@link MessageAck} that will acknowledge the given {@link MessageDispatch}.
+     *
+     * @param md
+     *        the {@link MessageDispatch} to acknowledge.
+     *
+     * @return a new {@link MessageAck} command to acknowledge the message.
+     */
+    public MessageAck createMessageAck(MessageDispatch md) {
         return new MessageAck(md, MessageAck.STANDARD_ACK_TYPE, 1);
     }
 
-    PUBLISH createPublish(ActiveMQMessage message) throws DataFormatException, IOException, JMSException {
+    /**
+     * Creates a PUBLISH command that can be sent to a remote client from an
+     * incoming {@link ActiveMQMessage} instance.
+     *
+     * @param message
+     *        the message to convert to a PUBLISH command.
+     *
+     * @return a new PUBLISH command that is populated from the {@link ActiveMQMessage}.
+     *
+     * @throws DataFormatException
+     * @throws IOException
+     * @throws JMSException
+     */
+    public PUBLISH createPublish(ActiveMQMessage message) throws DataFormatException, IOException, JMSException {
         PUBLISH publish = protocolConverter.convertMessage(message);
         if (publish.qos().ordinal() > this.qos.ordinal()) {
             publish.qos(this.qos);
@@ -63,6 +87,15 @@ class MQTTSubscription {
         return publish;
     }
 
+    /**
+     * Given a PUBLISH command determine if it will expect an ACK based on the
+     * QoS of the Publish command and the QoS of this subscription.
+     *
+     * @param publish
+     *        The publish command to inspect.
+     *
+     * @return true if the client will expect an PUBACK for this PUBLISH.
+     */
     public boolean expectAck(PUBLISH publish) {
         QoS publishQoS = publish.qos();
         if (publishQoS.compareTo(this.qos) > 0){
@@ -71,12 +104,35 @@ class MQTTSubscription {
         return !publishQoS.equals(QoS.AT_MOST_ONCE);
     }
 
+    /**
+     * @returns the original topic name value the client used when subscribing.
+     */
+    public String getTopicName() {
+        return this.topicName;
+    }
+
+    /**
+     * The real {@link ActiveMQDestination} that this subscription is assigned.
+     *
+     * @return the real {@link ActiveMQDestination} assigned to this subscription.
+     */
+    public ActiveMQDestination getDestination() {
+        return consumerInfo.getDestination();
+    }
+
+    /**
+     * Gets the {@link ConsumerInfo} that describes the subscription sent to ActiveMQ.
+     *
+     * @return the {@link ConsumerInfo} used to create this subscription.
+     */
     public ConsumerInfo getConsumerInfo() {
         return consumerInfo;
     }
 
+    /**
+     * @return the assigned QoS value for this subscription.
+     */
     public QoS qos() {
         return qos;
     }
-
 }

activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/ResponseHandler.java

@@ -20,10 +20,11 @@ import java.io.IOException;
 
 import org.apache.activemq.command.Response;
 
-
 /**
  * Interface used by the MQTTProtocolConverter for callbacks.
  */
-interface ResponseHandler {
+public interface ResponseHandler {
+
     void onResponse(MQTTProtocolConverter converter, Response response) throws IOException;
+
 }