MQTT Integration

1. Overview

The FT Device Network Service integrates with IoT devices through the MQTT protocol, a lightweight publish/subscribe messaging transport designed for IoT devices. This integration allows the service to receive telemetry data, diagnostics, and other information from network devices in real-time.

The service acts as an MQTT client, connecting to an MQTT broker where devices publish their data. It subscribes to specific topics to receive device messages, processes them asynchronously, and stores the relevant information in the ClickHouse database for analysis and visualization.

MQTT was chosen for its low overhead, reliability, and efficient use of network bandwidth, making it ideal for communicating with devices that may have limited connectivity or resources.

2. Configuration

The MQTT integration is configured in the MqttConfig class, which sets up the connection to the MQTT broker and defines how messages are processed:

Unresolved include directive in modules/ROOT/pages/mqtt.adoc - include::../src/main/java/com/friendly/network/config/MqttConfig.java[]

The configuration uses the following properties from application.yml which can be overridden with environment variables:

app:
  mqtt:
    url: ${MQTT_URL:tcp://localhost:1883}
    client-id: ${MQTT_CLIENT_ID:ft-device-network-service-}
    username: ${MQTT_USERNAME:username}
    password: ${MQTT_PASSWORD:password}
    topics: ${MQTT_TOPICS:device/+/data,device/+/diagnostic}
    qos: ${MQTT_QOS:1}

  • MQTT_URL - The URL of the MQTT broker (default: tcp://localhost:1883)

  • MQTT_CLIENT_ID - The client ID prefix used for the MQTT connection (default: ft-device-network-service-)

  • MQTT_USERNAME - The username for MQTT broker authentication (default: username)

  • MQTT_PASSWORD - The password for MQTT broker authentication (default: password)

  • MQTT_TOPICS - A comma-separated list of topics to subscribe to (default: device//data,device//diagnostic)

  • MQTT_QOS - The Quality of Service level for MQTT subscriptions (default: 1)

3. Connection Management

The MQTT connection is managed through Spring Integration’s MQTT support, which provides a robust framework for handling MQTT communications:

3.1. Client Factory

The mqttClientFactory bean configures the MQTT connection options:

  • Clean Session - Set to true to ensure a fresh session on each connection

  • Automatic Reconnect - Enabled to automatically reconnect if the connection is lost

  • Keep Alive Interval - Set to 60 seconds to maintain the connection

  • Connection Timeout - Set to 30 seconds for initial connection attempts

3.2. Message Adapter

The inboundMqttMessageProducer bean creates an MQTT message adapter that:

  • Connects to the MQTT broker using the configured client ID with a random UUID suffix

  • Subscribes to the configured topics

  • Sets the Quality of Service (QoS) level

  • Directs incoming messages to the MQTT input channel

4. Message Processing

Incoming MQTT messages are processed through a channel adapter and directed to a message handler:

4.1. Message Channel

The mqttInputChannel bean creates a direct channel for MQTT messages, providing a conduit between the MQTT adapter and the message handler.

4.2. Message Handler

The mqttMessageHandler bean configures the MqttDeviceDataListener as the handler for incoming messages. This component:

  • Extracts the device serial number from the topic

  • Identifies the message type (data or diagnostic)

  • Delegates processing to appropriate service components

  • Handles any errors that occur during processing

5. Data Flow

The flow of data through the MQTT integration follows these steps:

  1. Device Publication - IoT devices publish data to specific topics on the MQTT broker

  2. Subscription - The service subscribes to these topics and receives messages

  3. Message Routing - Messages are routed to the appropriate handler based on the topic

  4. Asynchronous Processing - Message processing is performed asynchronously to avoid blocking

  5. Data Storage - Processed data is stored in the ClickHouse database

  6. Metrics Collection - Metrics about message processing are collected for monitoring

6. Topic Structure

The service subscribes to topics following these patterns:

  • device/+/data - For device telemetry data

  • device/+/diagnostic - For device diagnostic information

The + wildcard is used to subscribe to messages from all device serial numbers. The actual topic format for a specific device would be:

  • device/{serial}/data - For telemetry data from a specific device

  • device/{serial}/diagnostic - For diagnostic information from a specific device

7. Security Considerations

The MQTT integration implements several security measures:

  • Authentication - Username/password authentication with the MQTT broker

  • Encryption - Support for TLS/SSL connections when configured with an appropriate broker URL

  • Client ID - Unique client IDs to prevent connection conflicts

  • Access Control - Relies on the MQTT broker’s access control mechanisms to restrict topic access

8. Monitoring and Resilience

The MQTT integration includes features for monitoring and ensuring resilience:

  • Automatic Reconnection - The client automatically reconnects if the connection is lost

  • Error Handling - Robust error handling prevents message processing failures from affecting the connection

  • Metrics - Message processing metrics are collected for monitoring

  • Logging - Comprehensive logging of connection events and message processing

← Back | Main Page