CrystalMQ Broker Tutorial – Build IoT Application better

Quick Overview

Bevywise CrystalMQ Broker Overview

Bevywise CrystalMQ Broker is a complete IoT Application Suite with an inbuilt MQTT Broker. It acts as a central MQTT server, that facilitates communication among MQTT-enabled edge devices / Internet of Things (IoT) devices and collects data that can be stored for data analytics and visualization.

CrystalMQ Broker uses Message Queuing Telemetry Transport (MQTT) protocol as its standard messaging protocol/communication protocol. MQTT works based on the publish-subscribe model. It provides you a fully extendable framework which helps you build powerful IoT/IIoT Application for any industrial use cases.

MQTT Versions Supported :

• MQTT 3.1
• MQTT 3.1.1
• MQTT 5

Key Functions

Topic-based Messaging: CrystalMQ uses a topic-based messaging model where messages are published to specific topics, allowing subscribers to receive only relevant data based on their subscriptions.

QoS Handling: CrystalMQ supports different Quality of Service (QoS) levels (0, 1, 2) to ensure message delivery reliability and guarantee, based on the requirements of the communication scenario.

Connection Management: It manages client connections, including establishing, maintaining, and terminating MQTT connections, as well as handling connection retries and timeouts.

Security: It provides security features such as TLS encryption, authentication mechanisms (username/password, client certificates), and access control to ensure secure and authorized communication.

Persistence: It offers message persistence options to store and manage messages, ensuring message delivery even in case of temporary network disruptions or client unavailability.

Scalability: With its ability to scale to support millions of concurrent connections, it stands out as the most scalable and high-performing MQTT Broker available.

Get Started

This section offers a comprehensive guide on downloading and installing the on-premise or self-hosted version of Bevywise CrystalMQ Broker on your chosen machines or servers. Additionally, it covers instructions on connecting devices and testing messaging scenarios (publishing and subscribing) with the broker.

Technical Details & Support

This section explains about the hardware requirements and files & directories for high level configurations.

Hardware Requirements

Hardware requirements vary based on the message rate and size.

However, please find the minimal requirements below for your reference.

• CPU - 1 core or higher
• RAM – 2 GB or higher
• Hard Disk – 50 GB or higher

Introducing MQTT Broker Folders

After completing the installation process, you will discover the following files and folders located at the specified path.

Technical Support

For any technical queries regarding MQTT Broker or need to discuss your requirements, you can always reach us at

Email : [email protected]
Message : https://www.bevywise.com/contact-us.html
Phone : India - +91 8072398868 / US - +1 707 879 8999

MQTT 5 Broker Properties

This section provides the detailed view on MQTT 5 properties and the configurations you can set to play with this protocol version.

Session Expiry

MQTT 5 introduces the concept of session and message expiry. Clients can set session expiry intervals, allowing the broker to clean up disconnected client sessions after a specified period.

You can configure and set the session expiry interval in broker.conf

You can define a specific time frame for session expiry. Once configured, disconnected client sessions will be automatically removed after the specified duration.

You have the option to set a universal session expiry interval for all devices.

Additionally, you can set individual session expiry intervals for each client by setting the value to -1. In this case, the broker will check the CONNECT packet of each client for the session expiry interval value, and it will use these values to manage session cleanup accordingly.

Receive Maximum

The RECEIVE MAXIMUM property is used to indicate the maximum number of QoS 1 and QoS 2 messages that the client can process concurrently. This enhances the flexibility and efficiency of message delivery for concurrent message processing, enabling better resource utilization and improved message handling dynamics between clients and broker.

You can configure and set the numbers in broker.conf under property SERVER_RECEIVE_MAXIMUM.

The value of SERVER_RECEIVE_MAXIMUM can range from 1 to 65,535.

A value of 0 indicates that the client can not process any QoS 1 or QoS 2 messages concurrently, effectively disabling these QoS levels for the client.

Maximum QoS

MQTT Broker can inform the client about the highest QoS level it supports by including the Maximum QoS in the CONNACK packet during connection establishment.

Upon receiving the Maximum QoS from the broker in the CONNACK packet, the client must adhere to this maximum QoS level when sending PUBLISH packets.

However, the broker can still accept SUBSCRIBE packets from clients containing a Requested QoS of 0, 1, or 2, regardless of its support for QoS 1 or QoS 2 PUBLISH packets.

You can define Maximum QoS in broker.conf

Maximum Packet Size

The Maximum Packet Size property is used by the broker to notify the client about the maximum allowed size for packets. The client is required not to send packets larger than this limit to the server. If the server receives a packet that exceeds this size limit, it's considered a Protocol Error, and the server will disconnect the client using DISCONNECT packet.

You can set the maximum packet size in broker.conf as below.

Topic Alias Maximum

The "Topic Alias Maximum" is a property used to specify the maximum number of topic aliases that a client can use during MQTT communication with the broker. This allows the broker to inform the client about the maximum number of topic aliases it can use.

Topic aliases are shorthand references to topics that help reduce the size of MQTT control packets by replacing the topic name with a shorter identifier (alias).

You can set Maximum Topic Alias in broker.conf.

A value of 0 indicates that the broker does not support topic aliases, and the client should not use topic aliases in its messages.

Server Keep Alive

The "Server Keep Alive" mechanism refers to a feature that allows the MQTT broker (server) to notify clients about the maximum time interval allowed for maintaining an active connection.

During the connection establishment phase (CONNACK packet), the MQTT broker informs the client about the "Server Keep Alive" interval. The "Server Keep Alive" interval specifies the maximum time duration (in seconds) that the client can remain idle without sending any control packets (e.g., PINGREQ) to the broker.

You can define this in mqtt.conf

If you specify the keep alive here, the client will use this value instead of the keep-alive value provided in its CONNECT packet.

However, if the value is set to -1, which means the broker does not send the Server Keep Alive property, and the broker will use the keep-alive value set by the client in its CONNECT packet.

Retain Available

The "Retain Available" feature refers to the capability of an MQTT broker to support retained messages. Retained messages are special MQTT messages that are stored on the broker and delivered to new subscribers when they subscribe to a topic.

You can enable / disable Retain in mqtt.conf

A value of 0 means that retained messages are disabled. A value of 1 means retained messages are enabled.

Wild Card Subscription Available

A wildcard subscription refers to a subscription pattern that allows clients to subscribe to multiple topics using wildcard characters (+, #).

You have the option to configure the MQTT Broker to either enable or disable the wildcard feature based on your requirements.

A value of 0 means that Wild Card Subscription is disabled. A value of 1 means that Wild Card Subscription is enabled.

Subscription Identifiers Available

Subscription identifiers are a feature that allows MQTT clients to assign identifiers to their subscriptions. Subscription identifiers are used to uniquely identify individual subscriptions made by MQTT clients to specific topics. Each subscription can be assigned a unique identifier.

A value of 0 means that subscription identifiers are disabled. A value of 1 means that subscription identifiers are enabled.

MQTT Shared Subscription

Our broker, by default, supports shared subscriptions, allowing data to be distributed among subscribed devices. When a publisher sends data to a topic, it is shared among all subscribed clients. For example, if two clients are subscribed to 'Topic 1', the broker will alternate sending messages between them.

Shared subscriptions follow the format below

You can enable / disable the same in mqtt.conf

Note : Please be aware that while shared subscriptions function within nodes in a clustered environment, they do not extend across the entire cluster.

MQTT Broker User Interface & Dashboard

CrystalMQ UI enables you to monitor, visualise & manage your connected devices & data. This section provides a deep view on how to access the dashboard and to visualise MQTT data in a way you need.

Access to the UI / Dashboard

By default, the UI of the broker starts running in the port 8080. You can access it by navigating to the browser and entering

Or

Clients

This menu displays the complete list of devices (both active & inactive) that are connected to the MQTT Broker. This provides you with a high level view on each device connected.

Let us dig deeper to know how this can help.

Dashboard can provide the latest quick events of devices. But how can we get the complete list of events published and the commands received if it is a subscriber?

How can I send an instant command to the devices?

All these questions can be answered if you can make the optimum use of Clients Menu.

• Go to Clients
• Click any client

Events

The complete list of events published by the device selected. You can also get the list of messages published along with its associated topics by the selected device.

• QoS level of each topic published. Read our blog to know more about Quality of Service (QoS) level.
• The time at which the message was published.

Commands

The complete list of commands subscribed by the device selected. You can also get,

• The list of messages received along with its associated topics by the selected device.
• QoS level of each topic subscribed.
• The time at which the message was received by the received.

Subscribe Topics

This is a list of topic filters subscribed by the particular client.

Send Command

‘Send Command’ helps you to send instant messages to client.

Follow the steps to send an instant command :

• Select topic name from the drop down menu.
• Make sure the topic is subscribed by the device to which you are sending command.
• Enter the command to be sent in the message tab.

Topics

The absolute list of topics published and subscribed by all devices connected to the broker.

Device Log

The connection of the device with the broker may be restricted/interrupted for so many reasons. The Device Log tab contains a list of errors that occurred while connecting a device to the broker. This helps you know the reason behind the failure.

The below list helps you to identify the reason for the error.

Security

Ensuring MQTT security is vital to protect sensitive data and maintain the integrity of messaging systems. By addressing the key aspects of MQTT security, organizations can establish a robust security posture for MQTT-based deployments, protect sensitive data, prevent unauthorized access, and maintain the integrity and confidentiality of MQTT communications.

Here are three levels of security employed in Bevywise MQTT Broker:

Device-Level Authentication

Device-level authentication involves validating the identity of devices connecting to the MQTT broker. This is crucial to prevent unauthorized devices from accessing the broker and ensures that only trusted devices can publish and subscribe to topics.

This enables you to connect your devices more securely with authentication credentials. This requires you to enable Authentication field in mqtt.conf file present in conf folder.

Enable Authentication :

• Go to ./CrystalMQ/conf (Path may vary)
• Open mqtt.conf
• You can find AUTHENTICATION_ENABLED field is set to DISABLED. Change it to DEFAULT.

Save the file & restart the broker.

Now go to the UI of MQTT Broker. You will find the ‘Security’ Menu displayed. You can create your Authentication credentials for secure device connection.

Create Authentication Credentials :

Add MQTT Username (Access Key) and Password (Access Token) that can be used by specific ClientIds. 'Clients' - can be modified to a single ClientId, or a list of comma separated ClientIds (wildcards are supported). '*' means that Clients with any ClientId can connect.

These credentials can also be deleted.

MQTT Username (Access Token) can be either stored plain or as a digest in the database. Setting SECURE_MQTT_PASSWORD = TRUE in conf/datastore.conf will store MQTT Passwords as a digest and will not be displayed in the UI.

Save the file & restart the broker.

Please note that switching this configuration requires dropping the existing database.

Access Control Lists (ACLs)

ACLs are used to define granular permissions and control which devices or clients can perform specific actions (e.g., publish, subscribe) on MQTT topics. This level of security helps enforce fine-grained access control policies within the MQTT broker.

With ACL enabled, you can specify which topics the devices are allowed to publish or subscribe to, as well as which topics they are restricted from accessing.

Enable ACL

To enable ACL, go to 'mqtt.conf' file in the 'conf' folder. Then set as follows : TOPIC_ACL = TRUE. Save the file and restart the broker.

MQTT Username based ACL

Enable Device Authentication to use MQTT Username based ACL.

Navigate to the 'Security' Menu

Under the Access Control List section, select MQTT Username

On creation of MQTT Username/Password, default ACL settings are automatically generated that allows PUBLISH/SUBSCRIBE of all Topics. This can be edited by clicking 'Edit' in the MQTT Username.

Comma Separated list of Topics can be configured to be allowed/denied for Publish and Subscribe of any Client connecting using this MQTT Username. Both the Publish and Subscribe Topics accept MQTT-like filters.

(eg. room/+/temperature)

Client based ACL

You can also set ACL based on the ClientId. ACL can be added only to already connected clients.

In the Security Menu, under the Access Control List Section, select 'Clients' and add ACL for a connected Client.

TLS / SSL Encryption

Transport Layer Security (TLS) or Secure Sockets Layer (SSL) encryption adds a layer of security by encrypting MQTT communications between clients and the broker. This prevents eavesdropping, tampering, or unauthorized access to MQTT messages during transmission.

Enable TLS / SSL

To enable TLS, set TLS_ENCRYPTION = TRUE in conf/mqtt.conf :

As TLS is enabled, broker will start running in the port 8883 instead of 1883.

Note : WSS_PORT_NO is to start the MQTT SSL version in Websocket.

By default, MQTTRoute comes with Self-Signed Certificate for Server. You can find the certificates inside CrystalMQ/Certificate

Root certificate (root.crt) must be uploaded on the device / client side so the client can verify that the server's certificate (server.crt) was signed by its trusted root certificate.

If you need to create your own self-signed certificate, check below.

Self-signed Certificate Creation - Linux

Generate the Certificate Authority's signing key

Enter pass phrase for the key and save it.

Result :

Generate a certificate signing request for the CA

Enter any organization name (Eg. Bevywise) and leave others empty.

Result :

Create the CA's root certificate

Result :

Create the server / MQTT Broker's key pair

Result :

Create a certificate signing request

This can be done using the server key (server.key) to send it to the Certificate Authority for identity verification

Provide any organization name (eg.Bevywise Inc) and enter the common name. Comman name (eg : tempmyaccount.mqttserver.com) should be the domain name of your server in which the MQTT Broker is running.

Or enter localhost if MQTT Broker is operating on a local machine.

Result :

Creating and signing the server certificate

This step is to create a new server certificate and sign it with the power of Certificate Authority.

Result :

Now you have the below certificates and key in hand.

• ca-root.crt (rename it to root.crt)
• server.crt
• server.key

Now, server.crt and server.key should be uploaded on the server ../MQTTRoute/Certificate/server, and root.crt should be provided on the client so the client can verify that the server's certificate (server.crt) was signed by its trusted root certificate.

Database Configuration

The options to configure how data is stored by the Broker is set in Bevywise/CrystalMQ/conf/datastore.conf.

Device Information (Device IDs and Subscriptions) and MQTT Payload (Published/Received messages) are stored by the broker in one of the relational databases, by default. The relational databases supported by the Broker are: SQLite, MySQL, MSSQL & PostgreSQL. The UI uses this database.

However, if the data needs to be stored in some other database, outside the Broker, then it can be configured too.

Enabling / Disabling Storage in Relational Database

To store both Device Information (Device IDs, Subscriptions) and Payload (Published/Received messages) in database,

• Go to Bevywise/CrystalMQ/conf/datastore.conf
• Set, RELATIONAL_PERSISTENCE_ENABLED as True and enable PAYLOAD_TO_DB

If you don’t want to store both Device Information and Payload in database, then set

In this case, CrystalMQ UI is NOT available. Hence, PAYLOAD_TO_DB is assumed to be DISABLED.

To store only the Device Information in database and NOT the MQTT Payloads, set

In this case, only device information can be seen in the UI.

Let us see the Relational Database Configurations below.

MQTT Sparkplug B support

MQTT Sparkplug B relies on an MQTT broker to facilitate communication between industrial devices, sensors, and applications like, SCADA, Historians, etc. The MQTT broker serves as the underlying communication infrastructure that adheres to the MQTT protocol standards and handles the routing of Sparkplug B-compliant messages.

Sparkplug B-compliant devices and applications publish and subscribe to topics on the MQTT broker, following the structured data formats and guidelines specified by the Sparkplug B protocol.

The MQTT broker provides the communication backbone and protocol support, while MQTT Sparkplug B defines the standards and conventions for data exchange within industrial IoT ecosystems, leveraging the capabilities of the MQTT protocol for efficient, reliable, and scalable industrial automation and monitoring.

Our broker, by default, supports Sparkplug-enabled devices without requiring changes to the configuration file. However, it's essential to verify that the 'authentication' setting is enabled in the mqtt.conf file.

MQTT Clustering

To ensure High Availability, more than 1 CrystalMQ Brokers can be used. Our CrystalMQ with the built-in Inter-Broker Communicator (IBC) enables all the brokers in the cluster to talk to each other ensuring continuous communication with the devices both ways irrespective of clients getting connecting to any broker in the cluster.

A load balancer can be set up at the device-facing edge to balance the load.

To learn more about the inter-broker communicator functionality, read our high availability blog.

To enable High Availability in CrystalMQ, you should activate the inter-broker communicator first.

Activate Inter-Broker Communicator

The built-in Inter-broker Communicator (IBC) feature ensures seamless communication among all brokers within a cluster. This guarantees continuous connectivity with devices in both directions, regardless of which broker clients connect to.

To enable IBC after installing MQTTRoute on broker machines, modify the configuration in Bevywise/MQTTRoute/conf/mqtt.conf as follows:

The illustration below demonstrates the setup:

In this arrangement, two or more brokers with IBC enabled are grouped to form a cluster. A load balancer is then configured to distribute workloads evenly across these brokers. All MQTT brokers within the cluster remain active, and IBC within each broker ensures persistent communication by utilizing a shared database (DB). This means that brokers do not directly communicate with one another. Instead, they push data to the central database, allowing other brokers to access this data. As a result, every broker has access to client details and data.

Process Supervision using Monit

To run the MQTT Broker as service, we need the Monit version 5.25. Monit is an open source dynamic monitoring tool for Linux systems which is used for monitoring and managing system processes. And also, it performs automatic maintenance or repair of a particular process (i.e., restarting the service) and execute significant informal actions in error conditions whenever necessary.

Download & Install Monit :

Extract the archive using the command below :

Copy monitrc file and paste in below location :

Enable Monit HTTP interface :

Enable the HTTP interface by un commenting the following lines in /etc/monitrc file.

If you want you can change admin:monit with username and password you want to use

Start Monit :

Open Monit UI using the below URL in your web browser http://localhost:2812

Extensions

CrystalMQ has certain extensions to make it flexible for users. You can customize it based on your needs.

These custom extensions can be used individually or in combinations to fulfil your requirements beyond an MQTT Broker.

In CrystalMQ/extensions/extension_globals.py, there are objects that can be used to query the Internal Database, to send live updates to UI and to send messages to a connected client.

These objects can be used in the extensions to yield compelling customisations.