From d8f9686d2a671b69bb16d1b529988dd2adfd21fb Mon Sep 17 00:00:00 2001 From: Timothy Bish Date: Thu, 31 Jul 2014 17:49:59 -0400 Subject: [PATCH] https://issues.apache.org/jira/browse/AMQ-5290 Clean up some code to make future work a bit simpler. --- .../transport/mqtt/MQTTProtocolConverter.java | 79 +++++++------------ .../transport/mqtt/MQTTProtocolSupport.java | 73 +++++++++++++++++ .../transport/mqtt/MQTTSubscription.java | 68 ++++++++++++++-- .../transport/mqtt/ResponseHandler.java | 5 +- 4 files changed, 165 insertions(+), 60 deletions(-) create mode 100644 activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolSupport.java diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolConverter.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolConverter.java index 4ca23c370f..91087d030c 100644 --- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolConverter.java +++ b/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, - MQTTSubscription mqttSubscription) throws MQTTProtocolException { + private void resendRetainedMessages(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; + } } diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolSupport.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolSupport.java new file mode 100644 index 0000000000..2def25cc9e --- /dev/null +++ b/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('.', '/'); + } +} diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTSubscription.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTSubscription.java index b4971bcd80..bbed6551e8 100644 --- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTSubscription.java +++ b/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; } - } diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/ResponseHandler.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/ResponseHandler.java index 75c218abca..f0ee0ad12d 100644 --- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/ResponseHandler.java +++ b/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; + }