search

fusesource / mqtt-client

A Java MQTT Client
http://mqtt-client.fusesource.org/
Apache License 2.0
1.27k stars 369 forks source link

readme

Overview

MQTT is a machine-to-machine (M2M)/"Internet of Things" connectivity protocol. It was designed as an extremely lightweight publish/subscribe messaging transport. It is useful for connections with remote locations where a small code footprint is required and/or network bandwidth is at a premium.

mqtt-client provides an ASL 2.0 licensed API to MQTT. It takes care of automatically reconnecting to your MQTT server and restoring your client session if any network failures occur. Applications can use a blocking API style, a futures based API, or a callback/continuations passing API style.

Using from Maven

Add the following to your maven pom.xml file.

<dependency>
  <groupId>org.fusesource.mqtt-client</groupId>
  <artifactId>mqtt-client</artifactId>
  <version>1.12</version>
</dependency>

Using from Gradle

Add the following to your gradle file.

compile 'org.fusesource.mqtt-client:mqtt-client:1.12'

Using from any Other Build System

Download the uber jar file and add it to your build. The uber contains all the stripped down dependencies which the mqtt-client depends on from other projects.

Using on Java 1.4

We also provide an java 1.4 uber jar file which is compatible with Java 1.4 JVMs. This version of the jar does not support SSL connections since the SSLEngine class used to implement SSL on NIO was not introduced until Java 1.5.

Configuring the MQTT Connection

The blocking, future, and callback APIs all share the same connection setup. You create a new instance of the MQTT class and configure it with connection and socket related options. At a minimum the setHost method be called before attempting to connect.

MQTT mqtt = new MQTT();
mqtt.setHost("localhost", 1883);
// or 
mqtt.setHost("tcp://localhost:1883");

Controlling MQTT Options

Controlling Connection Reconnects

Connection will automatically reconnect and re-establish messaging session if any network error occurs. You can control how often the reconnect is attempted and define maximum number of attempts of reconnects using the following methods:

Configuring Socket Options

You can adjust some socket options by using the following methods:

Throttling Connections

If you want slow down the read or write rate of your connections, use the following methods:

Using SSL connections

If you want to connect over SSL/TLS instead of TCP, use an "ssl://" or "tls://" URI prefix instead of "tcp://" for the host field. For finer grained control of which algorithm is used. Supported protocol values are:

The client will use the default JVM SSLContext which is configured via JVM system properties unless you configure the MQTT instance using the setSslContext method.

SSL connections perform blocking operations against internal thread pool unless you call the setBlockingExecutor method to configure that executor they will use instead.

Selecting the Dispatch Queue

A HawtDispatch dispatch queue is used to synchronize access to the connection. If an explicit queue is not configured via the setDispatchQueue method, then a new queue will be created for the connection. Setting an explicit queue might be handy if you want multiple connection to share the same queue for synchronization.

Using the Blocking API

The MQTT.connectBlocking method establishes a connection and provides you a connection with an blocking API.

BlockingConnection connection = mqtt.blockingConnection();
connection.connect();

Publish messages to a topic using the publish method:

connection.publish("foo", "Hello".getBytes(), QoS.AT_LEAST_ONCE, false);

You can subscribe to multiple topics using the the subscribe method:

Topic[] topics = {new Topic("foo", QoS.AT_LEAST_ONCE)};
byte[] qoses = connection.subscribe(topics);

Then receive and acknowledge consumption of messages using the receive, and ack methods:

Message message = connection.receive();
System.out.println(message.getTopic());
byte[] payload = message.getPayload();
// process the message then:
message.ack();

Finally to disconnect:

connection.disconnect();

Using the Future based API

The MQTT.connectFuture method establishes a connection and provides you a connection with an futures style API. All operations against the connection are non-blocking and return the result via a Future.

FutureConnection connection = mqtt.futureConnection();
Future<Void> f1 = connection.connect();
f1.await();

Future<byte[]> f2 = connection.subscribe(new Topic[]{new Topic(utf8("foo"), QoS.AT_LEAST_ONCE)});
byte[] qoses = f2.await();

// We can start future receive..
Future<Message> receive = connection.receive();

// send the message..
Future<Void> f3 = connection.publish("foo", "Hello".getBytes(), QoS.AT_LEAST_ONCE, false);

// Then the receive will get the message.
Message message = receive.await();
message.ack();

Future<Void> f4 = connection.disconnect();
f4.await();

Using the Callback/Continuation Passing based API

The MQTT.connectCallback method establishes a connection and provides you a connection with an callback style API. This is the most complex to use API style, but can provide the best performance. The future and blocking APIs use the callback api under the covers. All operations on the connection are non-blocking and results of an operation are passed to callback interfaces you implement.

Example:

final CallbackConnection connection = mqtt.callbackConnection();
connection.listener(new Listener() {

    public void onDisconnected() {
    }
    public void onConnected() {
    }

    public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) {
        // You can now process a received message from a topic.
        // Once process execute the ack runnable.
        ack.run();
    }
    public void onFailure(Throwable value) {
        connection.close(null); // a connection failure occured.
    }
})
connection.connect(new Callback<Void>() {
    public void onFailure(Throwable value) {
        result.failure(value); // If we could not connect to the server.
    }

    // Once we connect..
    public void onSuccess(Void v) {

        // Subscribe to a topic
        Topic[] topics = {new Topic("foo", QoS.AT_LEAST_ONCE)};
        connection.subscribe(topics, new Callback<byte[]>() {
            public void onSuccess(byte[] qoses) {
                // The result of the subcribe request.
            }
            public void onFailure(Throwable value) {
                connection.close(null); // subscribe failed.
            }
        });

        // Send a message to a topic
        connection.publish("foo", "Hello".getBytes(), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
            public void onSuccess(Void v) {
              // the pubish operation completed successfully.
            }
            public void onFailure(Throwable value) {
                connection.close(null); // publish failed.
            }
        });

        // To disconnect..
        connection.disconnect(new Callback<Void>() {
            public void onSuccess(Void v) {
              // called once the connection is disconnected.
            }
            public void onFailure(Throwable value) {
              // Disconnects never fail.
            }
        });
    }
});

Every connection has a HawtDispatch dispatch queue which it uses to process IO events for the socket. The dispatch queue is an Executor that provides serial execution of IO and processing events and is used to ensure synchronized access of connection.

The callbacks will be executing the dispatch queue associated with the connection so it safe to use the connection from the callback but you MUST NOT perform any blocking operations within the callback. If you need to perform some processing which MAY block, you must send it to another thread pool for processing. Furthermore, if another thread needs to interact with the connection it can only do it by using a Runnable submitted to the connection's dispatch queue.

Example of executing a Runnable on the connection's dispatch queue:

connection.getDispatchQueue().execute(new Runnable(){
    public void run() {
      connection.publish( ..... );
    }
});