MQTT Connector

The MQTT connector provides Akka Stream sources to connect to AMQP servers.

Artifacts

sbt
libraryDependencies += "com.lightbend.akka" %% "akka-stream-alpakka-mqtt" % "0.3"
Maven
<dependency>
  <groupId>com.lightbend.akka</groupId>
  <artifactId>akka-stream-alpakka-mqtt_2.11</artifactId>
  <version>0.3</version>
</dependency>
Gradle
dependencies {
  compile group: "com.lightbend.akka", name: "akka-stream-alpakka-mqtt_2.11", version: "0.3"
}

Usage

First we need to define various settings, that are required when connecting to an MQTT server.

Scala
val settings = MqttSourceSettings(
  MqttConnectionSettings(
    "tcp://localhost:1883",
    "test-client",
    new MemoryPersistence
  ),
  Map("topic1" -> MqttQoS.AtMostOnce, "topic2" -> MqttQoS.AtMostOnce)
)
Java
final MqttSourceSettings settings = MqttSourceSettings.create(
  MqttConnectionSettings.create(
    "tcp://localhost:1883",
    "test-client",
    new MemoryPersistence()
  )
).withSubscriptions(
  Pair.create("topic1", MqttQoS.atMostOnce()),
  Pair.create("topic2", MqttQoS.atMostOnce())
);

Here we used MqttSourceSettings factory to set the address of the server, client ID, which needs to be unique for every client, and client persistence implementation (MemoryPersistence) which allows to control reliability guarantees.

Then let’s create a source that is going to connect to the MQTT server upon materialization and receive messages that are sent to the subscribed topics.

Scala
val mqttSource = MqttSource(settings, bufferSize = 8)
Java
final Integer bufferSize = 8;
final Source<MqttMessage, CompletionStage<Done>> mqttSource =
  MqttSource.create(settings, bufferSize);

And finally run the source.

Scala
val (subscriptionFuture, result) = mqttSource
  .map(m => s"${m.topic}_${m.payload.utf8String}")
  .take(messageCount * 2)
  .toMat(Sink.seq)(Keep.both)
  .run()
Java
final Pair<CompletionStage<Done>, CompletionStage<List<String>>> result = mqttSource
  .map(m -> m.topic() + "_" + m.payload().utf8String())
  .take(messageCount * 2)
  .toMat(Sink.seq(), Keep.both())
  .run(materializer);

This source has a materialized value (Future in Scala API and CompletionStage in Java API) which is completed when the subscription to the MQTT broker has been completed.

Running the example code

The code in this guide is part of runnable tests of this project. You are welcome to edit the code and run it in sbt.

Scala
sbt
> mqtt/testOnly *.MqttSourceSpec
Java
sbt
> mqtt/testOnly *.MqttSourceTest