Vert.x MQTT

At the component level, you set general and shared configurations that are, then, inherited by the endpoints. It is the highest configuration level.

For example, a component may have security settings, credentials for authentication, urls for network connection and so forth.

Some components only have a few options, and others may have many. Because components typically have pre-configured defaults that are commonly used, then you may often only need to configure a few options on a component; or none at all.

You can configure components using:

You usually spend more time setting up endpoints because they have many options. These options help you customize what you want the endpoint to do. The options are also categorized into whether the endpoint is used as a consumer (from), as a producer (to), or both.

Configuring endpoints is most often done directly in the endpoint URI as path and query parameters. You can also use the Endpoint DSL and DataFormat DSL as a type safe way of configuring endpoints and data formats in Java.

A good practice when configuring options is to use Property Placeholders.

Property placeholders provide a few benefits:

The following two sections list all the options, firstly for the component followed by the endpoint.

The brokerUrl option names the broker and how to reach it:

User information in the URL is refused. Credentials belong into the username and password options, so that they do not end up in logs together with the URL. The WebSocket schemes ws and wss are refused as well, until the Vert.x MQTT client supports the WebSocket transport.

The consumer sets the body to the payload as a byte[], the topic it arrived on is in the CamelVertxMqttTopic header. The producer sends whatever the body converts to a byte[]: text, bytes, a stream or a file. A message without a body is sent with an empty payload.

The headers the consumer sets, CamelVertxMqttQos and CamelVertxMqttRetained among them, describe the message that arrived. The producer does not read them: a route that bridges one broker to another would otherwise publish every retained message it receives as retained again. To choose the quality of service, the retained flag or the topic of a single message, the producer reads CamelVertxMqttOverrideQos, CamelVertxMqttOverrideRetained and CamelVertxMqttOverrideTopic.

The consumer and the producer of an endpoint share one connection. It is opened when the first of them starts and closed when the last of them stops. Two endpoints are two connections, and since a broker allows one connection per client identifier, two endpoints must not share a clientId. Without a clientId a connection gets a random one, which it keeps for every reconnect. A session the broker keeps (cleanStart=false) is only found again after a restart of the route with a clientId set.

With automaticReconnect (the default) a lost connection is opened again, after reconnectDelay and then with a delay that doubles on every attempt, up to maxReconnectDelay, plus up to a fifth of it at random, so that the clients of a broker that went away do not all come back at the same moment. Every attempt uses a new Vert.x client, so that nothing of the lost connection carries over. If the broker cannot be reached when a route starts, the route starts all the same and the component keeps trying. A producer waits up to connectTimeout for the connection before the exchange fails. Without automaticReconnect a broker that cannot be reached fails the start of the route.

The Vert.x client keeps no session of its own: a message the broker did not acknowledge when the connection was lost is not sent again, its exchange fails. A redelivery by the error handler of the route sends it as a new message, so across a lost connection quality of service 2 gives at least once, not exactly once: the broker can receive the message twice.

After a reconnect the exchanges that waited for the connection are sent at once. More of them than maxInflightQueue fail right away, as they would while connected.

With cleanStart=false the broker keeps the session of the client, with MQTT 5 for sessionExpiryInterval seconds. Keep maxReconnectDelay well below sessionExpiryInterval: a reconnect after a lost connection that comes later finds no session, and the messages the broker held for it are gone. A consumer that stops then keeps its subscription at the broker, which holds the messages of quality of service 1 and 2 until the consumer connects again with the same clientId. After a reconnect to a session the broker kept, the consumer does not subscribe again: the session still holds the subscription, and subscribing again would deliver the retained messages once more. With cleanStart=true (the default) a consumer subscribes after every connect, and unsubscribes when it stops.

The keep alive of the Vert.x client only sends a PINGREQ when it wrote nothing for keepAliveInterval seconds. A producer that publishes continuously therefore does not notice a broker that stopped answering, until ackTimeout fails its messages of quality of service 1 and 2.

The producer completes an exchange when the broker acknowledged the message, with PUBACK for quality of service 1 and PUBCOMP for 2. A message of quality of service 0 completes when it is written. A message without an acknowledgement within ackTimeout seconds fails, and so does a message beyond maxInflightQueue unacknowledged messages.

The producer completes the exchange on a worker thread, never on the event loop of the client: the rest of the route, and a redelivery delay of its error handler, would otherwise block the keep alive of every connection on that event loop.

The consumer acknowledges a message of quality of service 1 or 2 as soon as it arrives. With manualAcksEnabled it acknowledges it once its exchange completed successfully. A message whose exchange failed is handled as follows:

The consumer processes the exchanges in the order the messages arrive, on a Vert.x worker of one thread, never on the event loop of the client, which also keeps the connection alive. When more than maxPendingMessages messages wait for their exchange, the client stops reading from the connection until half of them are done, also across a reconnect. The broker is then held back by TCP, for every quality of service. The bound is a soft one: messages the client had already read when it paused still arrive, and are acknowledged on arrival unless manualAcksEnabled is set. With MQTT 5 receiveMaximum also limits the messages of quality of service 1 and 2 the broker sends before they are acknowledged.

The consumer and the producer of an endpoint share the connection, so while the consumer holds the client back, the acknowledgements for the producer wait as well, and a close for a redelivery fails the messages of quality of service 1 and 2 the producer is waiting for. When the consumer of an endpoint with cleanStart=false stops while its producer still uses the connection, the subscription stays at the broker, and messages that arrive meanwhile are acknowledged and dropped. Use two endpoints when both carry load.

When the consumer stops, and during a graceful shutdown, it first stops reading, then waits for the exchanges of the messages it already received, and only then unsubscribes and disconnects: a message acknowledged on arrival would otherwise be lost.

A broker URL with ssl or mqtts connects over TLS. The sslContextParameters give the trust managers, the key managers for a client certificate, the protocols and the cipher suites. Without them the trust store of the JVM is used. The host name of the broker is always verified against its certificate, there is no option to switch the check off or to trust every certificate.

The component uses the Vert.x instance set on it, or the one in the registry if there is exactly one. Otherwise it creates one of its own, configured by vertxOptions, and closes it when it shuts down.