MQTT
Overview
The MQTT integration connects Nussknacker to an MQTT broker. Once configured, a source and a sink, both labeled <integration name> MQTT, are added to
the Component Palette. Both are unbounded streaming components and handle message payloads as JSON; the topic is chosen in each node.
To add the integration, open Integrations from the Designer's side menu and click Add new on the MQTT tile.
Integration parameters
| Parameter | Required | Description |
|---|---|---|
| Integration name | true | Name used to prefix names of the MQTT source and sink. |
| Broker hostname | true | MQTT broker hostname. |
| Broker port | true | MQTT broker port, a number (e.g. 1883, or 8883 for TLS). |
| Connection timeout in milliseconds | true | Connection timeout for communication with the MQTT broker, in milliseconds. |
| Use TLS | false | Enables TLS encryption of the connection with the MQTT broker. Off by default. |
| QOS | true | MQTT Quality of Service level: 0, 1 or 2. |
| Username | false | Username for authentication in the MQTT broker. |
| Password | false | Password for authentication in the MQTT broker. |
Use the check connection button to verify the configuration before saving it.
MQTT source
Subscribes to a topic and emits one record per message. Each message must be a JSON document; it is parsed and made available in the #input variable.
The type of #input is not known at design time (Unknown), so access its fields with the usual SpEL syntax, e.g. #input.temperature. A message that
is not valid JSON fails processing with Could not parse JSON input.
| Name | Description |
|---|---|
| Topic name | Topic this source subscribes to; its messages are read as JSON. Mandatory. |
Advanced parameters
| Name | Description |
|---|---|
| Event time | Expression producing the event timestamp (Instant) used for watermarks and time-based state. Left empty, the timestamp assigned by the upstream source is used, and the machine clock when it has none. |
| Max out-of-orderness | The maximum amount of time an event may be late before time-based transformations ignore it. Pre-filled with 10 seconds; with the value removed, no lateness is tolerated. |
| Idleness | The time period after which a parallel instance of this source that receives no events is marked as idle, so it stops holding back the watermark. Left empty, idleness is not detected. |
MQTT sink
Publishes one message per record to a topic.
| Name | Description |
|---|---|
| Topic name | Topic the messages are published to. Mandatory. It is fixed for the node, so the expression cannot depend on the event. |
| Value | Expression producing the message payload, published as JSON. Mandatory. Edited with the JSON template editor or as a SpEL expression. |
The parameter names used in the scenario JSON (e.g. via the API) are Topic and Value; Topic name is the label shown in the node details.