Source code

001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.transport.mqtt;
018
019import java.io.IOException;
020import java.util.Map;
021import java.util.concurrent.ConcurrentHashMap;
022import java.util.concurrent.ConcurrentMap;
023import java.util.concurrent.atomic.AtomicBoolean;
024import java.util.zip.DataFormatException;
025import java.util.zip.Inflater;
026
027import javax.jms.Destination;
028import javax.jms.InvalidClientIDException;
029import javax.jms.JMSException;
030import javax.jms.Message;
031import javax.security.auth.login.CredentialException;
032
033import org.apache.activemq.broker.BrokerService;
034import org.apache.activemq.broker.BrokerServiceAware;
035import org.apache.activemq.broker.region.policy.RetainedMessageSubscriptionRecoveryPolicy;
036import org.apache.activemq.command.ActiveMQBytesMessage;
037import org.apache.activemq.command.ActiveMQDestination;
038import org.apache.activemq.command.ActiveMQMapMessage;
039import org.apache.activemq.command.ActiveMQMessage;
040import org.apache.activemq.command.ActiveMQTextMessage;
041import org.apache.activemq.command.Command;
042import org.apache.activemq.command.ConnectionError;
043import org.apache.activemq.command.ConnectionId;
044import org.apache.activemq.command.ConnectionInfo;
045import org.apache.activemq.command.ExceptionResponse;
046import org.apache.activemq.command.MessageAck;
047import org.apache.activemq.command.MessageDispatch;
048import org.apache.activemq.command.MessageId;
049import org.apache.activemq.command.ProducerId;
050import org.apache.activemq.command.ProducerInfo;
051import org.apache.activemq.command.Response;
052import org.apache.activemq.command.SessionId;
053import org.apache.activemq.command.SessionInfo;
054import org.apache.activemq.command.ShutdownInfo;
055import org.apache.activemq.transport.mqtt.strategy.MQTTSubscriptionStrategy;
056import org.apache.activemq.util.ByteArrayOutputStream;
057import org.apache.activemq.util.ByteSequence;
058import org.apache.activemq.util.FactoryFinder;
059import org.apache.activemq.util.IOExceptionSupport;
060import org.apache.activemq.util.IdGenerator;
061import org.apache.activemq.util.JMSExceptionSupport;
062import org.apache.activemq.util.LRUCache;
063import org.apache.activemq.util.LongSequenceGenerator;
064import org.fusesource.hawtbuf.Buffer;
065import org.fusesource.hawtbuf.UTF8Buffer;
066import org.fusesource.mqtt.client.QoS;
067import org.fusesource.mqtt.client.Topic;
068import org.fusesource.mqtt.codec.CONNACK;
069import org.fusesource.mqtt.codec.CONNECT;
070import org.fusesource.mqtt.codec.DISCONNECT;
071import org.fusesource.mqtt.codec.MQTTFrame;
072import org.fusesource.mqtt.codec.PINGREQ;
073import org.fusesource.mqtt.codec.PINGRESP;
074import org.fusesource.mqtt.codec.PUBACK;
075import org.fusesource.mqtt.codec.PUBCOMP;
076import org.fusesource.mqtt.codec.PUBLISH;
077import org.fusesource.mqtt.codec.PUBREC;
078import org.fusesource.mqtt.codec.PUBREL;
079import org.fusesource.mqtt.codec.SUBACK;
080import org.fusesource.mqtt.codec.SUBSCRIBE;
081import org.fusesource.mqtt.codec.UNSUBACK;
082import org.fusesource.mqtt.codec.UNSUBSCRIBE;
083import org.slf4j.Logger;
084import org.slf4j.LoggerFactory;
085
086public class MQTTProtocolConverter {
087
088    private static final Logger LOG = LoggerFactory.getLogger(MQTTProtocolConverter.class);
089
090    public static final String QOS_PROPERTY_NAME = "ActiveMQ.MQTT.QoS";
091    public static final int V3_1 = 3;
092    public static final int V3_1_1 = 4;
093
094    private static final IdGenerator CONNECTION_ID_GENERATOR = new IdGenerator();
095    private static final MQTTFrame PING_RESP_FRAME = new PINGRESP().encode();
096    private static final double MQTT_KEEP_ALIVE_GRACE_PERIOD = 0.5;
097    static final int DEFAULT_CACHE_SIZE = 5000;
098
099    private final ConnectionId connectionId = new ConnectionId(CONNECTION_ID_GENERATOR.generateId());
100    private final SessionId sessionId = new SessionId(connectionId, -1);
101    private final ProducerId producerId = new ProducerId(sessionId, 1);
102    private final LongSequenceGenerator publisherIdGenerator = new LongSequenceGenerator();
103
104    private final ConcurrentMap<Integer, ResponseHandler> resposeHandlers = new ConcurrentHashMap<Integer, ResponseHandler>();
105    private final Map<String, ActiveMQDestination> activeMQDestinationMap = new LRUCache<String, ActiveMQDestination>(DEFAULT_CACHE_SIZE);
106    private final Map<Destination, String> mqttTopicMap = new LRUCache<Destination, String>(DEFAULT_CACHE_SIZE);
107
108    private final Map<Short, MessageAck> consumerAcks = new LRUCache<Short, MessageAck>(DEFAULT_CACHE_SIZE);
109    private final Map<Short, PUBREC> publisherRecs = new LRUCache<Short, PUBREC>(DEFAULT_CACHE_SIZE);
110
111    private final MQTTTransport mqttTransport;
112    private final BrokerService brokerService;
113
114    private final Object commnadIdMutex = new Object();
115    private int lastCommandId;
116    private final AtomicBoolean connected = new AtomicBoolean(false);
117    private final ConnectionInfo connectionInfo = new ConnectionInfo();
118    private CONNECT connect;
119    private String clientId;
120    private long defaultKeepAlive;
121    private int activeMQSubscriptionPrefetch = -1;
122    private final MQTTPacketIdGenerator packetIdGenerator;
123    private boolean publishDollarTopics;
124
125    public int version;
126
127    private final FactoryFinder STRATAGY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/transport/strategies/");
128
129    /*
130     * Subscription strategy configuration element.
131     *   > mqtt-default-subscriptions
132     *   > mqtt-virtual-topic-subscriptions
133     */
134    private String subscriptionStrategyName = "mqtt-default-subscriptions";
135    private MQTTSubscriptionStrategy subsciptionStrategy;
136
137    public MQTTProtocolConverter(MQTTTransport mqttTransport, BrokerService brokerService) {
138        this.mqttTransport = mqttTransport;
139        this.brokerService = brokerService;
140        this.packetIdGenerator = MQTTPacketIdGenerator.getMQTTPacketIdGenerator(brokerService);
141        this.defaultKeepAlive = 0;
142    }
143
144    int generateCommandId() {
145        synchronized (commnadIdMutex) {
146            return lastCommandId++;
147        }
148    }
149
150    public void sendToActiveMQ(Command command, ResponseHandler handler) {
151
152        // Lets intercept message send requests..
153        if (command instanceof ActiveMQMessage) {
154            ActiveMQMessage msg = (ActiveMQMessage) command;
155            try {
156                if (!getPublishDollarTopics() && findSubscriptionStrategy().isControlTopic(msg.getDestination())) {
157                    // We don't allow users to send to $ prefixed topics to avoid failing MQTT 3.1.1
158                    // specification requirements for system assigned destinations.
159                    if (handler != null) {
160                        try {
161                            handler.onResponse(this, new Response());
162                        } catch (IOException e) {
163                            LOG.warn("Failed to send command " + command, e);
164                        }
165                    }
166                    return;
167                }
168            } catch (IOException e) {
169                LOG.warn("Failed to send command " + command, e);
170            }
171        }
172
173        command.setCommandId(generateCommandId());
174        if (handler != null) {
175            command.setResponseRequired(true);
176            resposeHandlers.put(command.getCommandId(), handler);
177        }
178        getMQTTTransport().sendToActiveMQ(command);
179    }
180
181    void sendToMQTT(MQTTFrame frame) {
182        try {
183            mqttTransport.sendToMQTT(frame);
184        } catch (IOException e) {
185            LOG.warn("Failed to send frame " + frame, e);
186        }
187    }
188
189    /**
190     * Convert a MQTT command
191     */
192    public void onMQTTCommand(MQTTFrame frame) throws IOException, JMSException {
193        switch (frame.messageType()) {
194            case PINGREQ.TYPE:
195                LOG.debug("Received a ping from client: " + getClientId());
196                checkConnected();
197                sendToMQTT(PING_RESP_FRAME);
198                LOG.debug("Sent Ping Response to " + getClientId());
199                break;
200            case CONNECT.TYPE:
201                CONNECT connect = new CONNECT().decode(frame);
202                onMQTTConnect(connect);
203                LOG.debug("MQTT Client {} connected. (version: {})", getClientId(), connect.version());
204                break;
205            case DISCONNECT.TYPE:
206                LOG.debug("MQTT Client {} disconnecting", getClientId());
207                onMQTTDisconnect();
208                break;
209            case SUBSCRIBE.TYPE:
210                onSubscribe(new SUBSCRIBE().decode(frame));
211                break;
212            case UNSUBSCRIBE.TYPE:
213                onUnSubscribe(new UNSUBSCRIBE().decode(frame));
214                break;
215            case PUBLISH.TYPE:
216                onMQTTPublish(new PUBLISH().decode(frame));
217                break;
218            case PUBACK.TYPE:
219                onMQTTPubAck(new PUBACK().decode(frame));
220                break;
221            case PUBREC.TYPE:
222                onMQTTPubRec(new PUBREC().decode(frame));
223                break;
224            case PUBREL.TYPE:
225                onMQTTPubRel(new PUBREL().decode(frame));
226                break;
227            case PUBCOMP.TYPE:
228                onMQTTPubComp(new PUBCOMP().decode(frame));
229                break;
230            default:
231                handleException(new MQTTProtocolException("Unknown MQTTFrame type: " + frame.messageType(), true), frame);
232        }
233    }
234
235    void onMQTTConnect(final CONNECT connect) throws MQTTProtocolException {
236        if (connected.get()) {
237            throw new MQTTProtocolException("Already connected.");
238        }
239        this.connect = connect;
240
241        // The Server MUST respond to the CONNECT Packet with a CONNACK return code 0x01
242        // (unacceptable protocol level) and then disconnect the Client if the Protocol Level
243        // is not supported by the Server [MQTT-3.1.2-2].
244        if (connect.version() < 3 || connect.version() > 4) {
245            CONNACK ack = new CONNACK();
246            ack.code(CONNACK.Code.CONNECTION_REFUSED_UNACCEPTED_PROTOCOL_VERSION);
247            try {
248                getMQTTTransport().sendToMQTT(ack.encode());
249                getMQTTTransport().onException(IOExceptionSupport.create("Unsupported or invalid protocol version", null));
250            } catch (IOException e) {
251                getMQTTTransport().onException(IOExceptionSupport.create(e));
252            }
253            return;
254        }
255
256        String clientId = "";
257        if (connect.clientId() != null) {
258            clientId = connect.clientId().toString();
259        }
260
261        String userName = null;
262        if (connect.userName() != null) {
263            userName = connect.userName().toString();
264        }
265        String passswd = null;
266        if (connect.password() != null) {
267            passswd = connect.password().toString();
268        }
269
270        version = connect.version();
271
272        configureInactivityMonitor(connect.keepAlive());
273
274        connectionInfo.setConnectionId(connectionId);
275        if (clientId != null && !clientId.isEmpty()) {
276            connectionInfo.setClientId(clientId);
277        } else {
278            // Clean Session MUST be set for 0 length Client Id
279            if (!connect.cleanSession()) {
280                CONNACK ack = new CONNACK();
281                ack.code(CONNACK.Code.CONNECTION_REFUSED_IDENTIFIER_REJECTED);
282                try {
283                    getMQTTTransport().sendToMQTT(ack.encode());
284                    getMQTTTransport().onException(IOExceptionSupport.create("Invalid Client ID", null));
285                } catch (IOException e) {
286                    getMQTTTransport().onException(IOExceptionSupport.create(e));
287                }
288                return;
289            }
290            connectionInfo.setClientId("" + connectionInfo.getConnectionId().toString());
291        }
292
293        connectionInfo.setResponseRequired(true);
294        connectionInfo.setUserName(userName);
295        connectionInfo.setPassword(passswd);
296        connectionInfo.setTransportContext(mqttTransport.getPeerCertificates());
297
298        sendToActiveMQ(connectionInfo, new ResponseHandler() {
299            @Override
300            public void onResponse(MQTTProtocolConverter converter, Response response) throws IOException {
301
302                if (response.isException()) {
303                    // If the connection attempt fails we close the socket.
304                    Throwable exception = ((ExceptionResponse) response).getException();
305                    //let the client know
306                    CONNACK ack = new CONNACK();
307                    if (exception instanceof InvalidClientIDException) {
308                        ack.code(CONNACK.Code.CONNECTION_REFUSED_IDENTIFIER_REJECTED);
309                    } else if (exception instanceof SecurityException) {
310                        ack.code(CONNACK.Code.CONNECTION_REFUSED_NOT_AUTHORIZED);
311                    } else if (exception instanceof CredentialException) {
312                        ack.code(CONNACK.Code.CONNECTION_REFUSED_BAD_USERNAME_OR_PASSWORD);
313                    } else {
314                        ack.code(CONNACK.Code.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
315                    }
316                    getMQTTTransport().sendToMQTT(ack.encode());
317                    getMQTTTransport().onException(IOExceptionSupport.create(exception));
318                    return;
319                }
320
321                final SessionInfo sessionInfo = new SessionInfo(sessionId);
322                sendToActiveMQ(sessionInfo, null);
323
324                final ProducerInfo producerInfo = new ProducerInfo(producerId);
325                sendToActiveMQ(producerInfo, new ResponseHandler() {
326                    @Override
327                    public void onResponse(MQTTProtocolConverter converter, Response response) throws IOException {
328
329                        if (response.isException()) {
330                            // If the connection attempt fails we close the socket.
331                            Throwable exception = ((ExceptionResponse) response).getException();
332                            CONNACK ack = new CONNACK();
333                            ack.code(CONNACK.Code.CONNECTION_REFUSED_BAD_USERNAME_OR_PASSWORD);
334                            getMQTTTransport().sendToMQTT(ack.encode());
335                            getMQTTTransport().onException(IOExceptionSupport.create(exception));
336                            return;
337                        }
338
339                        CONNACK ack = new CONNACK();
340                        ack.code(CONNACK.Code.CONNECTION_ACCEPTED);
341                        connected.set(true);
342                        getMQTTTransport().sendToMQTT(ack.encode());
343
344                        if (connect.cleanSession()) {
345                            packetIdGenerator.stopClientSession(getClientId());
346                        } else {
347                            packetIdGenerator.startClientSession(getClientId());
348                        }
349
350                        findSubscriptionStrategy().onConnect(connect);
351                    }
352                });
353            }
354        });
355    }
356
357    void onMQTTDisconnect() throws MQTTProtocolException {
358        if (connected.compareAndSet(true, false)) {
359            sendToActiveMQ(connectionInfo.createRemoveCommand(), null);
360            sendToActiveMQ(new ShutdownInfo(), null);
361        }
362        stopTransport();
363    }
364
365    void onSubscribe(SUBSCRIBE command) throws MQTTProtocolException {
366        checkConnected();
367        LOG.trace("MQTT SUBSCRIBE message:{} client:{} connection:{}",
368                  command.messageId(), clientId, connectionInfo.getConnectionId());
369        Topic[] topics = command.topics();
370        if (topics != null) {
371            byte[] qos = new byte[topics.length];
372            for (int i = 0; i < topics.length; i++) {
373                try {
374                    qos[i] = findSubscriptionStrategy().onSubscribe(topics[i]);
375                } catch (IOException e) {
376                    throw new MQTTProtocolException("Failed to process subscription request", true, e);
377                }
378            }
379            SUBACK ack = new SUBACK();
380            ack.messageId(command.messageId());
381            ack.grantedQos(qos);
382            try {
383                getMQTTTransport().sendToMQTT(ack.encode());
384            } catch (IOException e) {
385                LOG.warn("Couldn't send SUBACK for " + command, e);
386            }
387        } else {
388            LOG.warn("No topics defined for Subscription " + command);
389        }
390    }
391
392    public void onUnSubscribe(UNSUBSCRIBE command) throws MQTTProtocolException {
393        checkConnected();
394        if (command.qos() != QoS.AT_LEAST_ONCE && (version != V3_1 || publishDollarTopics != true)) {
395            throw new MQTTProtocolException("Failed to process unsubscribe request", true, new Exception("UNSUBSCRIBE frame not properly formatted, QoS"));
396        }
397        UTF8Buffer[] topics = command.topics();
398        if (topics != null) {
399            for (UTF8Buffer topic : topics) {
400                try {
401                    findSubscriptionStrategy().onUnSubscribe(topic.toString());
402                } catch (IOException e) {
403                    throw new MQTTProtocolException("Failed to process unsubscribe request", true, e);
404                }
405            }
406        }
407        UNSUBACK ack = new UNSUBACK();
408        ack.messageId(command.messageId());
409        sendToMQTT(ack.encode());
410    }
411
412    /**
413     * Dispatch an ActiveMQ command
414     */
415    public void onActiveMQCommand(Command command) throws Exception {
416        if (command.isResponse()) {
417            Response response = (Response) command;
418            ResponseHandler rh = resposeHandlers.remove(Integer.valueOf(response.getCorrelationId()));
419            if (rh != null) {
420                rh.onResponse(this, response);
421            } else {
422                // Pass down any unexpected errors. Should this close the connection?
423                if (response.isException()) {
424                    Throwable exception = ((ExceptionResponse) response).getException();
425                    handleException(exception, null);
426                }
427            }
428        } else if (command.isMessageDispatch()) {
429            MessageDispatch md = (MessageDispatch) command;
430            MQTTSubscription sub = findSubscriptionStrategy().getSubscription(md.getConsumerId());
431            if (sub != null) {
432                MessageAck ack = sub.createMessageAck(md);
433                PUBLISH publish = sub.createPublish((ActiveMQMessage) md.getMessage());
434                switch (publish.qos()) {
435                    case AT_LEAST_ONCE:
436                    case EXACTLY_ONCE:
437                        publish.dup(publish.dup() ? true : md.getMessage().isRedelivered());
438                    case AT_MOST_ONCE:
439                }
440                if (ack != null && sub.expectAck(publish)) {
441                    synchronized (consumerAcks) {
442                        consumerAcks.put(publish.messageId(), ack);
443                    }
444                }
445                LOG.trace("MQTT Snd PUBLISH message:{} client:{} connection:{}",
446                          publish.messageId(), clientId, connectionInfo.getConnectionId());
447                getMQTTTransport().sendToMQTT(publish.encode());
448                if (ack != null && !sub.expectAck(publish)) {
449                    getMQTTTransport().sendToActiveMQ(ack);
450                }
451            }
452        } else if (command.getDataStructureType() == ConnectionError.DATA_STRUCTURE_TYPE) {
453            // Pass down any unexpected async errors. Should this close the connection?
454            Throwable exception = ((ConnectionError) command).getException();
455            handleException(exception, null);
456        } else if (command.isBrokerInfo()) {
457            //ignore
458        } else {
459            LOG.debug("Do not know how to process ActiveMQ Command {}", command);
460        }
461    }
462
463    void onMQTTPublish(PUBLISH command) throws IOException, JMSException {
464        checkConnected();
465        LOG.trace("MQTT Rcv PUBLISH message:{} client:{} connection:{}",
466                  command.messageId(), clientId, connectionInfo.getConnectionId());
467        ActiveMQMessage message = convertMessage(command);
468        message.setProducerId(producerId);
469        message.onSend();
470        sendToActiveMQ(message, createResponseHandler(command));
471    }
472
473    void onMQTTPubAck(PUBACK command) {
474        short messageId = command.messageId();
475        LOG.trace("MQTT Rcv PUBACK message:{} client:{} connection:{}",
476                  messageId, clientId, connectionInfo.getConnectionId());
477        packetIdGenerator.ackPacketId(getClientId(), messageId);
478        MessageAck ack;
479        synchronized (consumerAcks) {
480            ack = consumerAcks.remove(messageId);
481        }
482        if (ack != null) {
483            getMQTTTransport().sendToActiveMQ(ack);
484        }
485    }
486
487    void onMQTTPubRec(PUBREC commnand) {
488        //from a subscriber - send a PUBREL in response
489        PUBREL pubrel = new PUBREL();
490        pubrel.messageId(commnand.messageId());
491        sendToMQTT(pubrel.encode());
492    }
493
494    void onMQTTPubRel(PUBREL command) {
495        PUBREC ack;
496        synchronized (publisherRecs) {
497            ack = publisherRecs.remove(command.messageId());
498        }
499        if (ack == null) {
500            LOG.warn("Unknown PUBREL: {} received", command.messageId());
501        }
502        PUBCOMP pubcomp = new PUBCOMP();
503        pubcomp.messageId(command.messageId());
504        sendToMQTT(pubcomp.encode());
505    }
506
507    void onMQTTPubComp(PUBCOMP command) {
508        short messageId = command.messageId();
509        packetIdGenerator.ackPacketId(getClientId(), messageId);
510        MessageAck ack;
511        synchronized (consumerAcks) {
512            ack = consumerAcks.remove(messageId);
513        }
514        if (ack != null) {
515            getMQTTTransport().sendToActiveMQ(ack);
516        }
517    }
518
519    ActiveMQMessage convertMessage(PUBLISH command) throws JMSException {
520        ActiveMQBytesMessage msg = new ActiveMQBytesMessage();
521
522        msg.setProducerId(producerId);
523        MessageId id = new MessageId(producerId, publisherIdGenerator.getNextSequenceId());
524        msg.setMessageId(id);
525        LOG.trace("MQTT-->ActiveMQ: MQTT_MSGID:{} client:{} connection:{} ActiveMQ_MSGID:{}",
526                command.messageId(), clientId, connectionInfo.getConnectionId(), msg.getMessageId());
527        msg.setTimestamp(System.currentTimeMillis());
528        msg.setPriority((byte) Message.DEFAULT_PRIORITY);
529        msg.setPersistent(command.qos() != QoS.AT_MOST_ONCE && !command.retain());
530        msg.setIntProperty(QOS_PROPERTY_NAME, command.qos().ordinal());
531        if (command.retain()) {
532            msg.setBooleanProperty(RetainedMessageSubscriptionRecoveryPolicy.RETAIN_PROPERTY, true);
533        }
534
535        ActiveMQDestination destination;
536        synchronized (activeMQDestinationMap) {
537            destination = activeMQDestinationMap.get(command.topicName());
538            if (destination == null) {
539                String topicName = MQTTProtocolSupport.convertMQTTToActiveMQ(command.topicName().toString());
540                try {
541                    destination = findSubscriptionStrategy().onSend(topicName);
542                } catch (IOException e) {
543                    throw JMSExceptionSupport.create(e);
544                }
545
546                activeMQDestinationMap.put(command.topicName().toString(), destination);
547            }
548        }
549
550        msg.setJMSDestination(destination);
551        msg.writeBytes(command.payload().data, command.payload().offset, command.payload().length);
552        return msg;
553    }
554
555    public PUBLISH convertMessage(ActiveMQMessage message) throws IOException, JMSException, DataFormatException {
556        PUBLISH result = new PUBLISH();
557        // packet id is set in MQTTSubscription
558        QoS qoS;
559        if (message.propertyExists(QOS_PROPERTY_NAME)) {
560            int ordinal = message.getIntProperty(QOS_PROPERTY_NAME);
561            qoS = QoS.values()[ordinal];
562
563        } else {
564            qoS = message.isPersistent() ? QoS.AT_MOST_ONCE : QoS.AT_LEAST_ONCE;
565        }
566        result.qos(qoS);
567        if (message.getBooleanProperty(RetainedMessageSubscriptionRecoveryPolicy.RETAINED_PROPERTY)) {
568            result.retain(true);
569        }
570
571        String topicName;
572        synchronized (mqttTopicMap) {
573            topicName = mqttTopicMap.get(message.getJMSDestination());
574            if (topicName == null) {
575                String amqTopicName = findSubscriptionStrategy().onSend(message.getDestination());
576                topicName = MQTTProtocolSupport.convertActiveMQToMQTT(amqTopicName);
577                mqttTopicMap.put(message.getJMSDestination(), topicName);
578            }
579        }
580        result.topicName(new UTF8Buffer(topicName));
581
582        if (message.getDataStructureType() == ActiveMQTextMessage.DATA_STRUCTURE_TYPE) {
583            ActiveMQTextMessage msg = (ActiveMQTextMessage) message.copy();
584            msg.setReadOnlyBody(true);
585            String messageText = msg.getText();
586            if (messageText != null) {
587                result.payload(new Buffer(messageText.getBytes("UTF-8")));
588            }
589        } else if (message.getDataStructureType() == ActiveMQBytesMessage.DATA_STRUCTURE_TYPE) {
590            ActiveMQBytesMessage msg = (ActiveMQBytesMessage) message.copy();
591            msg.setReadOnlyBody(true);
592            byte[] data = new byte[(int) msg.getBodyLength()];
593            msg.readBytes(data);
594            result.payload(new Buffer(data));
595        } else if (message.getDataStructureType() == ActiveMQMapMessage.DATA_STRUCTURE_TYPE) {
596            ActiveMQMapMessage msg = (ActiveMQMapMessage) message.copy();
597            msg.setReadOnlyBody(true);
598            Map<String, Object> map = msg.getContentMap();
599            if (map != null) {
600                result.payload(new Buffer(map.toString().getBytes("UTF-8")));
601            }
602        } else {
603            ByteSequence byteSequence = message.getContent();
604            if (byteSequence != null && byteSequence.getLength() > 0) {
605                if (message.isCompressed()) {
606                    Inflater inflater = new Inflater();
607                    inflater.setInput(byteSequence.data, byteSequence.offset, byteSequence.length);
608                    byte[] data = new byte[4096];
609                    int read;
610                    ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
611                    while ((read = inflater.inflate(data)) != 0) {
612                        bytesOut.write(data, 0, read);
613                    }
614                    byteSequence = bytesOut.toByteSequence();
615                    bytesOut.close();
616                }
617                result.payload(new Buffer(byteSequence.data, byteSequence.offset, byteSequence.length));
618            }
619        }
620        LOG.trace("ActiveMQ-->MQTT:MQTT_MSGID:{} client:{} connection:{} ActiveMQ_MSGID:{}",
621                result.messageId(), clientId, connectionInfo.getConnectionId(), message.getMessageId());
622        return result;
623    }
624
625    public MQTTTransport getMQTTTransport() {
626        return mqttTransport;
627    }
628
629    boolean willSent = false;
630    public void onTransportError() {
631        if (connect != null) {
632            if (connected.get()) {
633                if (connect.willTopic() != null && connect.willMessage() != null && !willSent) {
634                    willSent = true;
635                    try {
636                        PUBLISH publish = new PUBLISH();
637                        publish.topicName(connect.willTopic());
638                        publish.qos(connect.willQos());
639                        publish.messageId(packetIdGenerator.getNextSequenceId(getClientId()));
640                        publish.payload(connect.willMessage());
641                        publish.retain(connect.willRetain());
642                        ActiveMQMessage message = convertMessage(publish);
643                        message.setProducerId(producerId);
644                        message.onSend();
645
646                        sendToActiveMQ(message, null);
647                    } catch (Exception e) {
648                        LOG.warn("Failed to publish Will Message " + connect.willMessage());
649                    }
650                }
651                // remove connection info
652                sendToActiveMQ(connectionInfo.createRemoveCommand(), null);
653            }
654        }
655    }
656
657    void configureInactivityMonitor(short keepAliveSeconds) {
658        MQTTInactivityMonitor monitor = getMQTTTransport().getInactivityMonitor();
659
660        // If the user specifically shuts off the InactivityMonitor with transport.useInactivityMonitor=false,
661        // then ignore configuring it because it won't exist
662        if (monitor == null) {
663            return;
664        }
665
666        // Client has sent a valid CONNECT frame, we can stop the connect checker.
667        monitor.stopConnectChecker();
668
669        long keepAliveMS = keepAliveSeconds * 1000;
670
671        LOG.debug("MQTT Client {} requests heart beat of {} ms", getClientId(), keepAliveMS);
672
673        try {
674            // if we have a default keep-alive value, and the client is trying to turn off keep-alive,
675
676            // we'll observe the server-side configured default value (note, no grace period)
677            if (keepAliveMS == 0 && defaultKeepAlive > 0) {
678                keepAliveMS = defaultKeepAlive;
679            }
680
681            long readGracePeriod = (long) (keepAliveMS * MQTT_KEEP_ALIVE_GRACE_PERIOD);
682
683            monitor.setProtocolConverter(this);
684            monitor.setReadKeepAliveTime(keepAliveMS);
685            monitor.setReadGraceTime(readGracePeriod);
686            monitor.startReadChecker();
687
688            LOG.debug("MQTT Client {} established heart beat of  {} ms ({} ms + {} ms grace period)",
689                      new Object[] { getClientId(), keepAliveMS, keepAliveMS, readGracePeriod });
690        } catch (Exception ex) {
691            LOG.warn("Failed to start MQTT InactivityMonitor ", ex);
692        }
693    }
694
695    void handleException(Throwable exception, MQTTFrame command) {
696        LOG.warn("Exception occurred processing: \n" + command + ": " + exception.toString());
697        LOG.debug("Exception detail", exception);
698
699        if (connected.get() && connectionInfo != null) {
700            connected.set(false);
701            sendToActiveMQ(connectionInfo.createRemoveCommand(), null);
702        }
703        stopTransport();
704    }
705
706    void checkConnected() throws MQTTProtocolException {
707        if (!connected.get()) {
708            throw new MQTTProtocolException("Not connected.");
709        }
710    }
711
712    private void stopTransport() {
713        try {
714            getMQTTTransport().stop();
715        } catch (Throwable e) {
716            LOG.debug("Failed to stop MQTT transport ", e);
717        }
718    }
719
720    ResponseHandler createResponseHandler(final PUBLISH command) {
721        if (command != null) {
722            return new ResponseHandler() {
723                @Override
724                public void onResponse(MQTTProtocolConverter converter, Response response) throws IOException {
725                    if (response.isException()) {
726                        Throwable error = ((ExceptionResponse) response).getException();
727                        LOG.warn("Failed to send MQTT Publish: ", command, error.getMessage());
728                        LOG.trace("Error trace: {}", error);
729                    }
730
731                    switch (command.qos()) {
732                        case AT_LEAST_ONCE:
733                            PUBACK ack = new PUBACK();
734                            ack.messageId(command.messageId());
735                            LOG.trace("MQTT Snd PUBACK message:{} client:{} connection:{}",
736                                      command.messageId(), clientId, connectionInfo.getConnectionId());
737                            converter.getMQTTTransport().sendToMQTT(ack.encode());
738                            break;
739                        case EXACTLY_ONCE:
740                            PUBREC req = new PUBREC();
741                            req.messageId(command.messageId());
742                            synchronized (publisherRecs) {
743                                publisherRecs.put(command.messageId(), req);
744                            }
745                            LOG.trace("MQTT Snd PUBREC message:{} client:{} connection:{}",
746                                      command.messageId(), clientId, connectionInfo.getConnectionId());
747                            converter.getMQTTTransport().sendToMQTT(req.encode());
748                            break;
749                        default:
750                            break;
751                    }
752                }
753            };
754        }
755        return null;
756    }
757
758    public long getDefaultKeepAlive() {
759        return defaultKeepAlive;
760    }
761
762    /**
763     * Set the default keep alive time (in milliseconds) that would be used if configured on server side
764     * and the client sends a keep-alive value of 0 (zero) on a CONNECT frame
765     * @param keepAlive the keepAlive in milliseconds
766     */
767    public void setDefaultKeepAlive(long keepAlive) {
768        this.defaultKeepAlive = keepAlive;
769    }
770
771    public int getActiveMQSubscriptionPrefetch() {
772        return activeMQSubscriptionPrefetch;
773    }
774
775    /**
776     * set the default prefetch size when mapping the MQTT subscription to an ActiveMQ one
777     * The default = 1
778     *
779     * @param activeMQSubscriptionPrefetch
780     *        set the prefetch for the corresponding ActiveMQ subscription
781     */
782    public void setActiveMQSubscriptionPrefetch(int activeMQSubscriptionPrefetch) {
783        this.activeMQSubscriptionPrefetch = activeMQSubscriptionPrefetch;
784    }
785
786    public MQTTPacketIdGenerator getPacketIdGenerator() {
787        return packetIdGenerator;
788    }
789
790    public void setPublishDollarTopics(boolean publishDollarTopics) {
791        this.publishDollarTopics = publishDollarTopics;
792    }
793
794    public boolean getPublishDollarTopics() {
795        return publishDollarTopics;
796    }
797
798    public ConnectionId getConnectionId() {
799        return connectionId;
800    }
801
802    public SessionId getSessionId() {
803        return sessionId;
804    }
805
806    public boolean isCleanSession() {
807        return this.connect.cleanSession();
808    }
809
810    public String getSubscriptionStrategy() {
811        return subscriptionStrategyName;
812    }
813
814    public void setSubscriptionStrategy(String name) {
815        this.subscriptionStrategyName = name;
816    }
817
818    public String getClientId() {
819        if (clientId == null) {
820            if (connect != null && connect.clientId() != null) {
821                clientId = connect.clientId().toString();
822            } else {
823                clientId = "";
824            }
825        }
826        return clientId;
827    }
828
829    protected MQTTSubscriptionStrategy findSubscriptionStrategy() throws IOException {
830        if (subsciptionStrategy == null) {
831            synchronized (STRATAGY_FINDER) {
832                if (subsciptionStrategy != null) {
833                    return subsciptionStrategy;
834                }
835
836                MQTTSubscriptionStrategy strategy = null;
837                if (subscriptionStrategyName != null && !subscriptionStrategyName.isEmpty()) {
838                    try {
839                        strategy = (MQTTSubscriptionStrategy) STRATAGY_FINDER.newInstance(subscriptionStrategyName);
840                        LOG.debug("MQTT Using subscription strategy: {}", subscriptionStrategyName);
841                        if (strategy instanceof BrokerServiceAware) {
842                            ((BrokerServiceAware)strategy).setBrokerService(brokerService);
843                        }
844                        strategy.initialize(this);
845                    } catch (Exception e) {
846                        throw IOExceptionSupport.create(e);
847                    }
848                } else {
849                    throw new IOException("Invalid subscription strategy name given: " + subscriptionStrategyName);
850                }
851
852                this.subsciptionStrategy = strategy;
853            }
854        }
855        return subsciptionStrategy;
856    }
857}