Scalable IoT integration using Apache ActiveMQ and MQTT

I have been doing a lot of work on MQTT support in Apache ActiveMQ recently, starting with hardening and adding support for MQTT 3.1.1 in ActiveMQ for the MQTT Interop Day Event I mentioned in a previous post.

I like MQTT as a simple protocol for IoT. It's easy to implement in devices, and is not overly complicated as protocols go. However, as an experienced JMS architect, the first thing that struck me is that it uses the publish-subscribe model. And as I expected AT_LEAST_ONCE and EXACTLY_ONCE subscriptions in MQTT are mapped to durable subscriptions in ActiveMQ.

This means MQTT consumers are limited to creating a single subscription with a fixed client-id for those QoSs, if they don't want to have to deal with duplicates. Essentially it has the same limitation when it comes to scaling consumers for JMS Topics.

If you aren't already familiar with it, the ActiveMQ documentation describes the issue in more detail. The documentation there also describes the ActiveMQ Virtual Topics feature to solve this problem using logical Topics which are mapped to physical Queues. Messages on these Queues can then be load balanced across multiple connections and consumers without having to worry about duplicates.

Compared to durable subscriptions, Queues also make management and monitoring easier. For instance, monitoring tools can be used to raise an alert when Queue size becomes too large, signaling that messages are piling up in the Broker. This alert could also be used to create more consumer process instances, etc. Apache Camel JMS endpoints, and other JMS utilities such as JMS listeners in Spring Framework can automatically increase or decrease the number of consumers based on demand.

The same documentation page also describes ActiveMQ's Composite Destinations feature for routing messages, which can come in handy as a wire-tap for audit logs, etc.

I have recently submitted a couple of major fixes in AMQ-5160 and AMQ-5187. AMQ-5160 started as an issue with wildcard authorization in ActiveMQ, which I first fixed for non-retained messages. Dejan Bosanac's suggestion of using Subscription Recovery Policy for retained messages, together with my fix for AMQ-5187, now makes it possible to do what I wanted to be able to do early on when I started fixing issues in ActiveMQ MQTT transport, i.e. process MQTT messages using Virtual Topics.

Also, the fix for AMQ-5160 basically adds Retained Messages as a Broker level feature in ActiveMQ. So non-MQTT Topic clients can set the ActiveMQ.Retain boolean property to true to mark a message to be retained in the Topic, and the Broker sets the boolean property ActiveMQ.Retained to true to mark a message as having been recovered as a retained message in a Topic. Note that the Broker always uses RetainedMessageSubscriptionRecoveryPolicy and any user supplied policies are simply added to retained message recovery. So, the user doesn't have to do anything special in the configuration for retained message support.

Retained messages work for mapped JMS Queues by recovering the retained message from the Virtual Topic for the first Queue consumer, so there are no duplicate recovered messages. The retained message will have the property ActiveMQ.Retained set to true.

The patches are waiting further testing and validation and should be applied to ActiveMQ trunk soon, to be included in the 5.10 release.

The highly scalable MQTT solution basically consists of MQTT producers sending messages using the MQTT protocol to ActiveMQ Virtual Topics, which are configured trivially using name patterns. These Virtual Topics are mapped to Queue names used by regular ActiveMQ Java JMS consumers. MQTT messages are mapped to JMS BytesMessages. Java developers should be happy to be able to use their favorite language on the server/consumer side.

Although the ActiveMQ Broker completely manages the QoS flow with the MQTT producer, the JMS BytesMessage will have the property ActiveMQ.MQTT.QoS set to sender's QoS. The JMS consumer does not have to do anything special with it, besides the standard JMS message acknowledgement. This property can also be used by JMS producers as the MQTT QoS for MQTT consumers. Also, JMS consumers can use JMS transactions to include other transactional resources such as databases, either using Idempotent Consumers or in the worst case, XA transactions.

Hopefully, users will have as much fun using these new capabilities in Apache ActiveMQ as I have had developing them. Cheers and good luck with your super scalable MQTT deployments with ActiveMQ. 

MQTT 3.1.1 support in JBoss A-MQ 6.1, Apache ActiveMQ 5.10-SNAPSHOT and Apache Camel 2.13.0

I had the good fortune of recently attending the MQTT Interoperability Test Day during the recent EclipseCon in Burlingame, California on March 17th 2014. The event was held by the Eclipse Foundation and the Eclipse IoT Working Group.

It's aim was to prove spec maturity of MQTT 3.1.1 by demonstrating industry adoption and interoperability among products that support them, and to potentially iron out any issues in the spec that might show up through the exercise. MQTT is a key protocol for the rapidly growing IoT approach. If you haven't heard of MQTT before you should definitely check out http://mqtt.org/.

At the event I was representing Red Hat Inc. and its JBoss A-MQ 6.1 (Early Access build 367) product, Apache ActiveMQ 5.10-SNAPSHOT, and Apache Camel 2.13.0. As part of the exercise Ian Craggs from the Eclipse Paho team had built a very useful mock client and server to check compliance with the draft MQTT 3.1.1 spec. That Python kit proved very valuable to the Fuse team and I in helping find and address several issues in Apache ActiveMQ's MQTT protocol implementation. More information on Ian's test kit can be found at https://wiki.eclipse.org/Paho/MQTT_Interop_Testing_Day.

As a result of all the testing and fixes, the MQTT implementation in ActiveMQ has improved by leaps and bounds. I also wrote test Java clients and server using the Fuse mqtt-client library, Apache Camel and the Apache ActiveMQ broker in JBoss A-MQ 6.1. The test client I wrote mirrors tests executed by Ian's Python client. It verified compliance with several key improvements in MQTT 3.1.1 listed below:

1. Basic publish subscribe
2. Retained messages
3. Offline message queueing
4. Will messages for client disconnects
5. Overlapping subscriptions with MQTT wildcards
6. Connection keep alive
7. Redelivery of messages on reconnect
8. Zero length client id (optional)
9. Dollar topics (optional)
10. Subscription failure (optional)

The improvements in ActiveMQ MQTT support were demonstrated when both the client and server passed all the above tests with flying colors when tested against the Python compliance test client and server as well as several other MQTT products tested for interoperability during the Interop Test Day. Since I implemented several fixes in the MQTT transport for ActiveMQ as well as some critical fixes in the Broker for supporting MQTT topics and wildcards, I can state that there is only one spec requirement (MQTT-3.1.4-2) that isn't supported at the moment. 

That requirement is questionable since it mandates that MQTT Brokers MUST disconnect an existing client connection when another connection sends a CONNECT packet with the same client id. This will cause issues in client libraries such as Fuse mqtt-client, which automatically reconnect when disconnected from the Broker. So ActiveMQ chooses to reject the new connection instead. 

The code for my Java JBoss A-MQ test clients and server configuration can be found at https://github.com/dhirajsb/jboss-fuse-mqtt-test. The instructions for running them are simple and found in the README.md files. 

All in all it was a very fruitful event, personally for me since I put in a lot of work to get all the MQTT issues fixed in Apache ActiveMQ, and for Red Hat to be able to now proudly say that we support MQTT 3.1.1 spec in a soon to be GA product JBoss Fuse A-MQ 6.1. 

Of course this wouldn't have been possible without the hard work by Ian Skerret from the Eclipse Foundation in organizing the Interop Test Day, and all the help and support I received from my colleagues in Fuse engineering team at Red Hat.

I hope you take the time to check out the Java test clients and Broker and have lots of fun using MQTT in your applications. 

Salesforce integration using Apache Camel

This is the first in a series of articles on Salesforce integration using Apache Camel.

The need for Salesforce support was obvious, and to be fair to Camel, it did have the fundamental SOAP/REST, CometD support needed to integrate with Salesforce from scratch. But that requires a considerable amount of work by the developer to setup the low level protocol details, and mapping to and from domain objects to Salesforce message format.

So, the primary goals when designing the Salesforce component were:

1. Easy Salesforce connection setup (OAuth 2.0 authentication) and management (session re-establishment on expiry, etc.)
2. Support for Salesforce domain object (SObjects in Salesforce terminology) generation and mapping
3. Support for all or majority of Salesforce integration APIs (SOAP, REST, Bulk, Streaming)

Salesforce integration APIs for SOAP and REST have their own quirks which make working with them interesting. The SOAP API uses a WSDL and is always tied to a specific Salesforce platform version, and comes with all the pros and cons of a SOAP interface. The REST API is not quite pure REST, more like RESTish, and uses some ad-hoc features like using a different root name (SObject name) for what is essentially the same object (SObject description).

But the REST interface does not require a new WSDL to be compiled with every Salesforce release, and is backward compatible. More importantly, the REST interface also makes it very easy to generate code for Salesforce domain objects (that map to and from JSON/XML), compared to having to generate a WSDL/Schema for user objects and compiling that to add to an application. 

With these goals and ideas the Salesforce component became what it is now. A two piece solution consisting of a maven plugin to generate Salesforce domain objects and a Salesforce component that supports REST, Bulk, and Streaming APIs. 

The maven plugin generates POJOs from Salesforce objects, along with some helper classes that support Salesforce queries, picklists, etc. It uses the excellent Joda Time library for date time fields, and Jackson for JSON support and XStream for XML support. 

The Salesforce component supports easy configuration to use OAuth 2.0 User name password authentication. It uses Jetty HTTP Client to manage Salesforce connections and transparently handles session token refresh. It supports working with both raw JSON/XML and domain objects generated by the Camel Salesforce maven plugin. 

Hopefully Apache Camel users will find this component easy to use and it drastically reduces the time it takes them to integrate with Salesforce. 

Future posts in this series will focus on project setup and using the rich set of features supported by the component. For the impatient among us, there are extensive unit tests for every component feature here.   Enjoy.