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