MQTT Streaming

build.sbt

@@ -22,6 +22,7 @@ lazy val modules: Seq[ProjectReference] = Seq(
   kudu,
   mongodb,
   mqtt,
+  mqttStreaming,
   orientdb,
   reference,
   s3,
@@ -160,6 +161,11 @@ lazy val mongodb = alpakkaProject("mongodb", "mongodb", Dependencies.MongoDb)
 
 lazy val mqtt = alpakkaProject("mqtt", "mqtt", Dependencies.Mqtt)
 
+lazy val mqttStreaming = alpakkaProject("mqtt-streaming", "mqttStreaming", Dependencies.MqttStreaming)
+lazy val mqttStreamingBench = alpakkaProject("mqtt-streaming-bench", "mqttStreamingBench", Seq.empty)
+  .enablePlugins(JmhPlugin)
+  .dependsOn(mqtt, mqttStreaming)
+
 lazy val orientdb = alpakkaProject("orientdb",
                                    "orientdb",
                                    Dependencies.OrientDB,

docs/src/main/paradox/index.md

@@ -42,6 +42,7 @@ The [Alpakka project](https://developer.lightbend.com/docs/alpakka/current/) is
 * [JMS](jms.md)
 * [MongoDB](mongodb.md)
 * [MQTT](mqtt.md)
+* [MQTT Streaming](mqtt-streaming.md)
 * [OrientDB](orientdb.md)
 * [Pulsar](external/pulsar.md)
 * [Server-sent Events (SSE)](sse.md)

docs/src/main/paradox/mqtt-streaming.md

@@ -0,0 +1,93 @@
+# MQTT Streaming
+
+@@@ note { title="MQTT Streaming" }
+
+MQTT stands for MQ Telemetry Transport. It is a publish/subscribe messaging protocol, designed for constrained devices and low-bandwidth, high-latency or unreliable networks. The design principles are to minimize network bandwidth and device resource requirements whilst also attempting to ensure reliability and some degree of assurance of delivery. These principles also turn out to make the protocol ideal of the emerging “machine-to-machine” (M2M) or “Internet of Things” world of connected devices, and for mobile applications where bandwidth and battery power are at a premium.  
+
+Further information on [mqtt.org](https://mqtt.org/).
+
+@@@ 
+
+@@@ note { title="Paho Differences" }
+
+Alpakka contains @ref[another MQTT connector](mqtt.md) which is based on the Eclipse Paho client. Unlike the Paho version, this library has no dependencies other than those of Akka Streams i.e. it is entirely reactive. As such, there should be a significant performance advantage given its pure-Akka foundations, particularly in terms of memory usage given its diligent use of threads.
+
+This library also differs in that it separates out the concern of how MQTT is connected. Unlike Paho, where TCP is assumed, this library can join in any flow. The end result is that by using this library, Unix Domain Sockets, TCP, UDP or anything else can be used to transport MQTT.
+
+@@@
+
+The Alpakka MQTT connector provides an Akka Stream flow to connect to MQTT brokers. In addition, a flow is provided so that you can implement your own MQTT server in the case where you do not wish to use a broker--MQTT is a fine protocol for directed client/server interactions, as well as having an intermediary broker.
+
+### Reported issues
+
+[Tagged issues at Github](https://github.com/akka/alpakka/labels/p%3Amqtt-streaming)
+
+## Artifacts
+
+@@dependency [sbt,Maven,Gradle] {
+  group=com.lightbend.akka
+  artifact=akka-stream-alpakka-mqtt-streaming_$scala.binary.version$
+  version=$project.version$
+}
+
+## Flow through a client session
+
+The following code illustrates how to establish an MQTT client session and join it with a TCP connection:
+
+Scala
+: @@snip [snip](/mqtt-streaming/src/test/scala/docs/scaladsl/MqttFlowSpec.scala) { #create-streaming-flow }
+
+Java
+: @@snip [snip](/mqtt-streaming/src/test/java/docs/javadsl/MqttFlowTest.java) { #create-streaming-flow }
+
+The resulting flow's type shows how `Command`s are received and `Event`s are emitted. With `Event`, they can
+be either decoded successfully or not.
+
+Run the flow by connecting a source of messages to be published via a queue:
+
+Scala
+: @@snip [snip](/mqtt-streaming/src/test/scala/docs/scaladsl/MqttFlowSpec.scala) { #run-streaming-flow }
+
+Java
+: @@snip [snip](/mqtt-streaming/src/test/java/docs/javadsl/MqttFlowTest.java) { #run-streaming-flow }
+
+We drop the first 3 events received as they will be ACKs to our connect, subscribe and publish. The next event
+received is the publication to the topic we just subscribed to.
+
+## Flow through a server session
+
+The following code illustrates how to establish an MQTT server session and join it with a TCP binding:
+
+Scala
+: @@snip [snip](/mqtt-streaming/src/test/scala/docs/scaladsl/MqttFlowSpec.scala) { #create-streaming-bind-flow }
+
+Java
+: @@snip [snip](/mqtt-streaming/src/test/java/docs/javadsl/MqttFlowTest.java) { #create-streaming-bind-flow }
+
+The resulting source's type shows how `Event`s are received and `Command`s are queued in reply. Our example
+acknowledges a connection, subscription and publication. Upon receiving a publication, it is re-published
+from the server so that any client that is subscribed will receive it.
+
+Run the flow:
+
+Scala
+: @@snip [snip](/mqtt-streaming/src/test/scala/docs/scaladsl/MqttFlowSpec.scala) { #run-streaming-bind-flow }
+
+Java
+: @@snip [snip](/mqtt-streaming/src/test/java/docs/javadsl/MqttFlowTest.java) { #run-streaming-bind-flow }
+
+## 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-streaming/testOnly *.MqttFlowSpec
+    ```
+
+Java
+:   ```
+    sbt
+    > mqtt-streaming/testOnly *.MqttFlowTest
+    ```

docs/src/main/paradox/mqtt.md

@@ -8,6 +8,12 @@ Further information on [mqtt.org](https://mqtt.org/).
 
 @@@ 
 
+@@@ note { title="Streaming Differences" }
+
+Alpakka contains @ref[another MQTT connector](mqtt-streaming.md) which is _not_ based on the Eclipse Paho client, unlike this one. Please refer to the other connector where the differences are expanded on.
+
+@@@
+
 The Alpakka MQTT connector provides an Akka Stream source, sink and flow to connect to MQTT brokers. It is based on the [Eclipse Paho Java client](https://www.eclipse.org/paho/clients/java/).
 
 ### Reported issues

mqtt-streaming-bench/src/main/resources/application.conf

@@ -0,0 +1,5 @@
+akka {
+  loggers = []
+  stdout-loglevel = "OFF"
+  loglevel = "OFF"
+}

mqtt-streaming-bench/src/main/scala/akka/stream/alpakka/mqtt/MqttPerf.scala

@@ -0,0 +1,129 @@
+/*
+ * Copyright (C) 2016-2018 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.mqtt
+
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.locks.ReentrantLock
+
+import akka.actor.ActorSystem
+import akka.stream.alpakka.mqtt.scaladsl.MqttFlow
+import akka.stream.alpakka.mqtt.streaming.scaladsl.{ActorMqttServerSession, Mqtt}
+import akka.stream.scaladsl.{BroadcastHub, Keep, Sink, Source, Tcp}
+import akka.stream.{ActorMaterializer, Materializer, OverflowStrategy}
+import akka.util.ByteString
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
+import org.openjdk.jmh.annotations._
+
+import scala.concurrent.duration._
+import scala.concurrent.Await
+
+object MqttPerf {
+  /*
+   * An entry point for debugging purposes - invoke whatever you need to debug
+   */
+  def main(args: Array[String]): Unit = {
+    val test = new MqttPerf()
+    test.setup()
+    try {
+      for (_ <- 0 until 10000) test.serverPublish()
+    } finally {
+      test.tearDown()
+    }
+  }
+}
+
+@State(Scope.Benchmark)
+class MqttPerf {
+
+  import streaming.MqttCodec._
+
+  private implicit val system: ActorSystem = ActorSystem("mqttperf")
+  private implicit val mat: Materializer = ActorMaterializer()
+
+  private val (_, clientSource) = Source
+    .queue[MqttMessage](2, OverflowStrategy.backpressure)
+    .toMat(BroadcastHub.sink)(Keep.both)
+    .run()
+
+  private val (server, serverSource) = Source
+    .queue[streaming.Command[_]](1, OverflowStrategy.backpressure)
+    .toMat(BroadcastHub.sink)(Keep.both)
+    .run()
+
+  private val pubAckReceivedLock = new ReentrantLock()
+  private val pubAckReceived = pubAckReceivedLock.newCondition()
+
+  @Setup
+  def setup(): Unit = {
+    val host = "localhost"
+    val port = 9883
+
+    val connectionSettings = MqttConnectionSettings(s"tcp://$host:$port", "some-client-id", new MemoryPersistence)
+
+    val connAck = streaming.ConnAck(streaming.ConnAckFlags.None, streaming.ConnAckReturnCode.ConnectionAccepted)
+    val subAck = streaming.SubAck(streaming.PacketId(1), List(streaming.ControlPacketFlags.QoSAtLeastOnceDelivery))
+
+    val serverSession = ActorMqttServerSession(streaming.MqttSessionSettings())
+
+    val bound = Tcp()
+      .bind(host, port)
+      .flatMapMerge(
+        1, { connection =>
+          Source
+            .fromGraph(serverSource)
+            .via(
+              Mqtt
+                .serverSessionFlow(serverSession, ByteString(connection.remoteAddress.getAddress.getAddress))
+                .join(connection.flow)
+            )
+            .wireTap(Sink.foreach[Either[DecodeError, streaming.Event[_]]] {
+              case Right(streaming.Event(_: streaming.Connect, _)) =>
+                server.offer(streaming.Command(connAck))
+              case Right(streaming.Event(s: streaming.Subscribe, _)) =>
+                server.offer(streaming.Command(subAck.copy(packetId = s.packetId)))
+              case Right(streaming.Event(_: streaming.PubAck, _)) =>
+                pubAckReceivedLock.lock()
+                try {
+                  pubAckReceived.signal()
+                } finally {
+                  pubAckReceivedLock.unlock()
+                }
+              case _ =>
+            })
+        }
+      )
+      .toMat(Sink.ignore)(Keep.left)
+      .run()
+    Await.ready(bound, 3.seconds)
+
+    Source
+      .fromGraph(clientSource)
+      .via(
+        MqttFlow.atLeastOnce(
+          connectionSettings,
+          MqttSubscriptions("some-topic", MqttQoS.AtLeastOnce),
+          bufferSize = 8,
+          MqttQoS.AtLeastOnce
+        )
+      )
+      .mapAsync(1)(_.ack())
+      .runWith(Sink.ignore)
+  }
+
+  @Benchmark
+  def serverPublish(): Unit = {
+    server.offer(streaming.Command(streaming.Publish("some-topic", ByteString("some-payload"))))
+    pubAckReceivedLock.lock()
+    try {
+      pubAckReceived.await(3, TimeUnit.SECONDS)
+    } finally {
+      pubAckReceivedLock.unlock()
+    }
+  }
+
+  @TearDown
+  def tearDown(): Unit =
+    system.terminate()
+}