= Vert.x MQTT server

This component provides a server which is able to handle connections, communication and messages exchange with remote
link:http://mqtt.org/[MQTT] clients.

Its API provides a bunch of events related to raw protocol messages received by
clients and exposes some features in order to send messages to them.

It's not a fully featured MQTT broker but can be used for building something like that or for protocol translation.

WARNING: this module has the tech preview status, this means the API can change between versions.

== Using Vert.x MQTT server

This component had officially released in the Vert.x stack, just following dependency to the _dependencies_ section
of your build descriptor:

* Maven (in your `pom.xml`):

[source,xml,subs="+attributes"]
----
<dependency>
    <groupId>io.vertx</groupId>
    <artifactId>vertx-mqtt-server</artifactId>
    <version>3.4.2</version>
</dependency>
----

* Gradle (in your `build.gradle` file):

[source,groovy,subs="+attributes"]
----
compile io.vertx:vertx-mqtt-server:3.4.2
----

== Getting Started

=== Handling client connection/disconnection

This example shows how it's possible to handle the connection request from a remote MQTT client. First, an
`link:../../apidocs/io/vertx/mqtt/MqttServer.html[MqttServer]` instance is created and the `link:../../apidocs/io/vertx/mqtt/MqttServer.html#endpointHandler-io.vertx.core.Handler-[endpointHandler]` method is used to specify the handler called
when a remote client sends a CONNECT message for connecting to the server itself. The `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html[MqttEndpoint]`
instance, provided as parameter to the handler, brings all main information related to the CONNECT message like client identifier,
username/password, "will" information, clean session flag, protocol version and the "keep alive" timeout.
Inside that handler, the _endpoint_ instance provides the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#accept-boolean-[accept]` method
for replying to the remote client with the corresponding CONNACK message : in this way, the connection is established.
Finally, the server is started using the `link:../../apidocs/io/vertx/mqtt/MqttServer.html#listen-io.vertx.core.Handler-[listen]` method with
the default behavior (on localhost and default MQTT port 1883). The same method allows to specify an handler in order
to check if the server is started properly or not.

[source,kotlin]
----

var mqttServer = MqttServer.create(vertx)
mqttServer.endpointHandler({ endpoint ->

  // shows main connect info
  println("MQTT client [${endpoint.clientIdentifier()}] request to connect, clean session = ${endpoint.isCleanSession()}")

  if (endpoint.auth() != null) {
    println("[username = ${endpoint.auth().userName()}, password = ${endpoint.auth().password()}]")
  }
  if (endpoint.will() != null) {
    println("[will topic = ${endpoint.will().willTopic()} msg = ${endpoint.will().willMessage()} QoS = ${endpoint.will().willQos()} isRetain = ${endpoint.will().isWillRetain()}]")
  }

  println("[keep alive timeout = ${endpoint.keepAliveTimeSeconds()}]")

  // accept connection from the remote client
  endpoint.accept(false)

}).listen({ ar ->

  if (ar.succeeded()) {

    println("MQTT server is listening on port ${ar.result().actualPort()}")
  } else {

    println("Error on starting the server")
    ar.cause().printStackTrace()
  }
})

----

The same _endpoint_ instance provides the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#disconnectHandler-io.vertx.core.Handler-[disconnectHandler]`
for specifying the handler called when the remote client sends a DISCONNECT message in order to disconnect from the server;
this handler takes no parameters.

[source,kotlin]
----

// handling disconnect message
endpoint.disconnectHandler({ v ->

  println("Received disconnect from client")
})

----

=== Handling client connection/disconnection with SSL/TLS support

The server has the support for accepting connection requests through the SSL/TLS protocol for authentication and encryption.
In order to do that, the `link:../../apidocs/io/vertx/mqtt/MqttServerOptions.html[MqttServerOptions]` class provides the `link:../../apidocs/io/vertx/mqtt/MqttServerOptions.html#setSsl-boolean-[setSsl]` method
for setting the usage of SSL/TLS (passing 'true' as value) and some other useful methods for providing server certificate and
related private key (as Java key store reference, PEM or PFX format). In the following example, the
`link:../../apidocs/io/vertx/mqtt/MqttServerOptions.html#setKeyCertOptions-io.vertx.core.net.KeyCertOptions-[setKeyCertOptions]` method is used in order to
pass the certificates in PEM format. This method requires an instance of the possible implementations of the
`link:../../apidocs/io/vertx/core/net/KeyCertOptions.html[KeyCertOptions]` interface and in this case the `link:../../apidocs/io/vertx/core/net/PemKeyCertOptions.html[PemKeyCertOptions]` class
is used in order to provide the path for the server certificate and the private key with the correspondent
`link:../../apidocs/io/vertx/core/net/PemKeyCertOptions.html#setCertPath-java.lang.String-[setCertPath]` and
`link:../../apidocs/io/vertx/core/net/PemKeyCertOptions.html#setKeyPath-java.lang.String-[setKeyPath]` methods.
The MQTT server is started passing the Vert.x instance as usual and the above MQTT options instance to the creation method.

[source,kotlin]
----

var options = MqttServerOptions(
  port = 8883,
  keyCertOptions = PemKeyCertOptions(
    keyPath = "./src/test/resources/tls/server-key.pem",
    certPath = "./src/test/resources/tls/server-cert.pem"),
  ssl = true)

var mqttServer = MqttServer.create(vertx, options)
mqttServer.endpointHandler({ endpoint ->

  // shows main connect info
  println("MQTT client [${endpoint.clientIdentifier()}] request to connect, clean session = ${endpoint.isCleanSession()}")

  if (endpoint.auth() != null) {
    println("[username = ${endpoint.auth().userName()}, password = ${endpoint.auth().password()}]")
  }
  if (endpoint.will() != null) {
    println("[will topic = ${endpoint.will().willTopic()} msg = ${endpoint.will().willMessage()} QoS = ${endpoint.will().willQos()} isRetain = ${endpoint.will().isWillRetain()}]")
  }

  println("[keep alive timeout = ${endpoint.keepAliveTimeSeconds()}]")

  // accept connection from the remote client
  endpoint.accept(false)

}).listen({ ar ->

  if (ar.succeeded()) {

    println("MQTT server is listening on port ${ar.result().actualPort()}")
  } else {

    println("Error on starting the server")
    ar.cause().printStackTrace()
  }
})

----

All the other stuff related to handle endpoint connection and related disconnection is managed in the same way without SSL/TLS support.

=== Handling client subscription/unsubscription request

After a connection is established between client and server, the client can send a subscription request for a topic
using the SUBSCRIBE message. The `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html[MqttEndpoint]` interface allows to specify an handler for the
incoming subscription request using the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#subscribeHandler-io.vertx.core.Handler-[subscribeHandler]` method.
Such handler receives an instance of the `link:../../apidocs/io/vertx/mqtt/messages/MqttSubscribeMessage.html[MqttSubscribeMessage]` interface which brings
the list of topics with related QoS levels as desired by the client.
Finally, the endpoint instance provides the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#subscribeAcknowledge-int-java.util.List-[subscribeAcknowledge]` method
for replying to the client with the related SUBACK message containing the granted QoS levels.

[source,kotlin]
----

// handling requests for subscriptions
endpoint.subscribeHandler({ subscribe ->

  var grantedQosLevels = mutableListOf<Any?>()
  for (s in subscribe.topicSubscriptions()) {
    println("Subscription for ${s.topicName()} with QoS ${s.qualityOfService()}")
    grantedQosLevels.add(s.qualityOfService())
  }
  // ack the subscriptions request
  endpoint.subscribeAcknowledge(subscribe.messageId(), grantedQosLevels)

})

----

In the same way, it's possible to use the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#unsubscribeHandler-io.vertx.core.Handler-[unsubscribeHandler]` method
on the endpoint in order to specify the handler called when the client sends an UNSUBSCRIBE message. This handler receives
an instance of the `link:../../apidocs/io/vertx/mqtt/messages/MqttUnsubscribeMessage.html[MqttUnsubscribeMessage]` interface as parameter with the list of topics to unsubscribe.
Finally, the endpoint instance provides the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#unsubscribeAcknowledge-int-[unsubscribeAcknowledge]` method
for replying to the client with the related UNSUBACK message.

[source,kotlin]
----

// handling requests for unsubscriptions
endpoint.unsubscribeHandler({ unsubscribe ->

  for (t in unsubscribe.topics()) {
    println("Unsubscription for ${t}")
  }
  // ack the subscriptions request
  endpoint.unsubscribeAcknowledge(unsubscribe.messageId())
})

----

=== Handling client published message

In order to handle incoming messages published by the remote client, the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html[MqttEndpoint]` interface provides
the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#publishHandler-io.vertx.core.Handler-[publishHandler]` method for specifying the handler called
when the client sends a PUBLISH message. This handler receives an instance of the `link:../../apidocs/io/vertx/mqtt/messages/MqttPublishMessage.html[MqttPublishMessage]`
interface as parameter with the payload, the QoS level, the duplicate and retain flags.

If the QoS level is 0 (AT_MOST_ONCE), there is no need from the endpoint to reply the client.

If the QoS level is 1 (AT_LEAST_ONCE), the endpoind needs to reply with a PUBACK message using the
available `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#publishAcknowledge-int-[publishAcknowledge]` method.

If the QoS level is 2 (EXACTLY_ONCE), the endpoint needs to reply with a PUBREC message using the
available `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#publishReceived-int-[publishReceived]` method; in this case the same endpoint should handle
the PUBREL message received from the client as well (the remote client sends it after receiving the PUBREC from the endpoint)
and it can do that specifying the handler through the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#publishReleaseHandler-io.vertx.core.Handler-[publishReleaseHandler]` method.
In order to close the QoS level 2 delivery, the endpoint can use the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#publishComplete-int-[publishComplete]` method
for sending the PUBCOMP message to the client.

[source,kotlin]
----

// handling incoming published messages
endpoint.publishHandler({ message ->

  println("Just received message [${message.payload().toString(java.nio.charset.Charset.defaultCharset())}] with QoS [${message.qosLevel()}]")

  if (message.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
    endpoint.publishAcknowledge(message.messageId())
  } else if (message.qosLevel() == MqttQoS.EXACTLY_ONCE) {
    endpoint.publishRelease(message.messageId())
  }

}).publishReleaseHandler({ messageId ->

  endpoint.publishComplete(messageId)
})

----

=== Publish message to the client

The endpoint can publish a message to the remote client (sending a PUBLISH message) using the
`link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#publish-java.lang.String-io.vertx.core.buffer.Buffer-io.netty.handler.codec.mqtt.MqttQoS-boolean-boolean-[publish]` method
which takes the following input parameters : the topic to publish, the payload, the QoS level, the duplicate and retain flags.

If the QoS level is 0 (AT_MOST_ONCE), the endpoint won't receiving any feedback from the client.

If the QoS level is 1 (AT_LEAST_ONCE), the endpoint needs to handle the PUBACK message received from the client
in order to receive final acknowledge of delivery. It's possible using the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#publishAcknowledgeHandler-io.vertx.core.Handler-[publishAcknowledgeHandler]` method
specifying such an handler.

If the QoS level is 2 (EXACTLY_ONCE), the endpoint needs to handle the PUBREC message received from the client.
The `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#publishReceivedHandler-io.vertx.core.Handler-[publishReceivedHandler]` method allows to specify
the handler for that. Inside that handler, the endpoint can use the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#publishRelease-int-[publishRelease]` method
for replying to the client with the PUBREL message. The last step is to handle the PUBCOMP message received from the client
as final acknowledge for the published message; it's possible using the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#publishCompleteHandler-io.vertx.core.Handler-[publishCompleteHandler]`
for specifying the handler called when the final PUBCOMP message is received.

[source,kotlin]
----

// just as example, publish a message with QoS level 2
endpoint.publish("my_topic", Buffer.buffer("Hello from the Vert.x MQTT server"), MqttQoS.EXACTLY_ONCE, false, false)

// specifing handlers for handling QoS 1 and 2
endpoint.publishAcknowledgeHandler({ messageId ->

  println("Received ack for message = ${messageId}")

}).publishReceivedHandler({ messageId ->

  endpoint.publishRelease(messageId)

}).publishCompleteHandler({ messageId ->

  println("Received ack for message = ${messageId}")
})

----

=== Be notified by client keep alive

The underlying MQTT keep alive mechanism is handled by the server internally. When the CONNECT message is received,
the server takes care of the keep alive timeout specified inside that message in order to check if the client doesn't
send messages in such timeout. At same time, for every PINGREQ received, the server replies with the related PINGRESP.

Even if there is no need for the high level application to handle that, the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html[MqttEndpoint]` interface
provides the `link:../../apidocs/io/vertx/mqtt/MqttEndpoint.html#pingHandler-io.vertx.core.Handler-[pingHandler]` method for specifying an handler
called when a PINGREQ message is received from the client. It's just a notification to the application that the client
isn't sending meaningful messages but only pings for keeping alive; in any case the PINGRESP is automatically sent
by the server internally as described above.

[source,kotlin]
----

// handling ping from client
endpoint.pingHandler({ v ->

  println("Ping received from client")
})

----

=== Closing the server

The `link:../../apidocs/io/vertx/mqtt/MqttServer.html[MqttServer]` interface provides the `link:../../apidocs/io/vertx/mqtt/MqttServer.html#close-io.vertx.core.Handler-[close]` method
that can be used for closing the server; it stops to listen for incoming connections and closes all the active connections
with remote clients. This method is asynchronous and one overload provides the possibility to specify a complention handler
that will be called when the server is really closed.

[source,kotlin]
----

mqttServer.close({ v ->

  println("MQTT server closed")
})

----

=== Automatic clean-up in verticles

If you’re creating MQTT servers from inside verticles, those servers will be automatically closed when the verticle is undeployed.

=== Scaling : sharing MQTT servers

The handlers related to the MQTT server are always executed in the same event loop thread. It means that on a system with
more cores, only one instance is deployed so only one core is used. In order to use more cores, it's possible to deploy
more instances of the MQTT server.

It's possible to do that programmatically:

[source,kotlin]
----

for (i in 0 until 10) {

  var mqttServer = MqttServer.create(vertx)
  mqttServer.endpointHandler({ endpoint ->
    // handling endpoint
  }).listen({ ar ->

    // handling start listening
  })


}


----

or using a verticle specifying the number of instances:

[source,kotlin]
----

var options = DeploymentOptions(
  instances = 10)
vertx.deployVerticle("com.mycompany.MyVerticle", options)

----

What's really happen is that even only MQTT server is deployed but as incoming connections arrive, Vert.x distributes
them in a round-robin fashion to any of the connect handlers executed on different cores.