HiveMQ Enterprise Extension for PostgreSQL

PostgreSQL is an open-source relational database management system that is widely used to safely store and manage large amounts of data. PostgreSQL supports most features of the current SQL standard. For more information, see SQL Conformance.

Features

Forward MQTT messages from IoT devices that are connected to your HiveMQ MQTT broker to one or more PostgreSQL databases.
Convert MQTT messages into PostgreSQL rows with convenient statement templates and insert statements.

Requirements

A running HiveMQ Professional or Enterprise Edition installation, version 4.17 or higher.
A running PostgreSQL database system. Version 11.20 or higher is recommended.
For production use, a valid HiveMQ Enterprise Extension for PostgreSQL license.

If you do not provide a valid license, HiveMQ automatically uses a free trial license. Trial licenses for HiveMQ Enterprise Extensions are valid for 5 hours. For more license information or to request an extended evaluation license, contact HiveMQ sales.

Installation

Place your HiveMQ Enterprise Extension for PostgreSQL license file (.elic) in the license folder of your HiveMQ installation. (Skip this step if you are using a trial version of the extension).

└─ <HiveMQ folder>
    ├─ bin
    ├─ conf
    ├─ data
    ├─ extensions
    │   ├─ hivemq-postgresql-extension
    │   └─ ...
    ├─ license
    ├─ log
    └─ ...

Before you enable the extension, you need to configure the extension to match your individual PostgreSQL setup.
For your convenience, we provide an example configuration conf/examples/config.xml that you can copy and modify as desired.
The included config.xsd file outlines the schema and elements that can be used in the XML configuration.
Your completed configuration file must be named config.xml and located in HIVEMQ_HOME/extensions/hivemq-postgresql-extension/conf/config.xml.
For detailed information on configuration options, see Configuration.
To enable the HiveMQ Enterprise Extension for PostgreSQL, locate the hivemq-postgresql-extension folder in the extensions directory of your HiveMQ installation and remove the DISABLED file (if present).

To function properly, the HiveMQ Enterprise Extension for PostgreSQL must be installed on all HiveMQ broker nodes in your HiveMQ cluster and the configuration file on each node must be identical.

Configuration

The extension configuration is divided into two sections:

PostgreSQLs: Provides information about the PostgreSQL databases to which your HiveMQ broker connects.
MQTT to PostgreSQL Routes: Defines how MQTT messages are sent from your HiveMQ broker to the configured PostgreSQL databases.

Extension Configuration File

The config.xml file for your PostgreSQL extension must be located in the hivemq-postgresql-extension/conf/ folder within the extensions folder of your HiveMQ installation.

The extension uses a simple but powerful XML-based configuration.

The conf/examples/config.xml file is a configuration example that has all the parameters you need to send MQTT messages from your HiveMQ MQTT broker to PostgreSQL.

If you copy and reuse the conf/examples/config.xml file, be sure to move the file to /conf/config.xml before you enable your extension. For more information, see Installation.

Example PostgreSQL database connection configuration

<hivemq-postgresql-extension xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                             xsi:noNamespaceSchemaLocation="config.xsd">
    <postgresqls>
        <postgresql>
            <id>my-postgresql-id</id>
            <host>my.host.com</host>
            <port>5432</port>
            <database>my-database</database>
            <username>my-user</username>
            <password>my-password</password>
        </postgresql>
    </postgresqls>

    <mqtt-to-postgresql-routes>
        <mqtt-to-postgresql-route>
            <id>my-mqtt-to-postgresql-route-insert</id>
            <postgresql-id>my-postgresql-id</postgresql-id>
            <mqtt-topic-filters>
                <mqtt-topic-filter>#</mqtt-topic-filter>
            </mqtt-topic-filters>
            <processor>
                <insert-statement>
                    <table>mqtt_to_postgresql_table</table>
                    <batching>true</batching>
                    <columns>
                        <column>
                            <name>topic</name>
                            <value>mqtt-topic</value>
                        </column>
                        <column>
                            <name>payload_utf8</name>
                            <value>mqtt-payload-utf8</value>
                        </column>
                    </columns>
                </insert-statement>
            </processor>
        </mqtt-to-postgresql-route>

        <mqtt-to-postgresql-route>
            <id>my-mqtt-to-postgresql-route-template</id>
            <postgresql-id>my-postgresql-id</postgresql-id>
            <mqtt-topic-filters>
                <mqtt-topic-filter>#</mqtt-topic-filter>
            </mqtt-topic-filters>
            <processor>
                <statement-template>path/to/my/statement-template.sql</statement-template>
            </processor>
        </mqtt-to-postgresql-route>
    </mqtt-to-postgresql-routes>
</hivemq-postgresql-extension>

PostgreSQL Databases Configuration

The <postgresqls> section of your configuration lists the PostgreSQL databases to which configured MQTT messages are routed.

You can define as many <postgresql> tags as your use case requires.

Example PostgreSQL database connection configuration

<postgresqls>
    <postgresql>
        <id>my-postgresql-id</id>
        <host>my.host.com</host>
        <port>5432</port>
        <database>my-database</database>
        <schema>my-schema</schema>
        <username>my-user</username>
        <password>my-password</password>
    </postgresql>
</postgresqls>

Example PostgreSQL database connection with a TLS configuration

<postgresqls>
    <postgresql>
        <id>my-postgresql-id</id>
        <host>my.host.com</host>
        <port>5432</port>
        <database>my-database</database>
        <username>my-user</username>
        <password>my-password</password>
        <tls>
            <root-certificate>/path/to/root-certificate.pem</root-certificate>
            <client-certificate>/path/to/client-certificate.pem</client-certificate>
            <client-private-key>/path/to/client-private-key.pem</client-private-key>
        </tls>
    </postgresql>
</postgresqls>

You can use Environment variables to map properties such as the PostgreSQL password.

Table 1. PostgreSQL configuration parameters
Parameter	Required	Type	Description
`id`		ID	The ID of the PostgreSQL connection. This string can only contain lowercase alphanumeric characters, dashes, and underscores.
`host`		String	The host name of the PostgreSQL database.
`port`		Integer	The port of the PostgreSQL database.
`database`		String	The name of the PostgreSQL database.
`schema`	-	String	Optional setting to specify one or more PostgreSQL database schemas in the search path as a comma-separated string. The default search path is `"$user",public`.
`username`		String	The username to connect with basic authentication.
`password`		String	The password to connect with basic authentication.
`tls`	-	Complex	Optional TLS configuration to establish a secure connection to PostgreSQL. `enabled`: Optional setting that defines whether the TLS configuration is used to encrypt communication between the extension and PostgreSQL. The default setting is `true`. To disable your TLS configuration, set the `enabled` parameter to `false`. `root-certificate`: Path to a PEM encoded X.509 trusted root certificates. The internal JDK CA-certificates are used if this setting is not provided. `client-certificate`: Path to a PEM encoded X.509 client certificate chain. `client-private-key`: Path to a PKCS#8 private key file in PEM format. `verify-hostname`: Optional setting that defines whether the extension checks the identity of the remote server. The default setting is `true`. To skip the verification, set the value to `false`.

MQTT to PostgreSQL Routes

The <mqtt-to-postgresql-routes> section of your extension configuration defines how MQTT messages are sent from the HiveMQ broker to PostgreSQL.

You can define as many <mqtt-to-postgresql-route> tags as your use case requires.

Example MQTT to PostgreSQL route with an insert statement configuration

<mqtt-to-postgresql-route>
    <id>my-mqtt-to-postgresql-route-insert</id>
    <postgresql-id>my-postgresql-id</postgresql-id>
    <mqtt-topic-filters>
        <mqtt-topic-filter>#</mqtt-topic-filter>
    </mqtt-topic-filters>
    <processor>
        <insert-statement>
            <table>mqtt_to_postgresql_table</table>
            <batching>true</batching>
            <columns>
                <column>
                    <name>topic</name>
                    <value>mqtt-topic</value>
                </column>
                <column>
                    <name>payload_utf8</name>
                    <value>mqtt-payload-utf8</value>
                </column>
            </columns>
        </insert-statement>
    </processor>
</mqtt-to-postgresql-route>

Example MQTT to PostgreSQL route with a statement template configuration

<mqtt-to-postgresql-route>
    <id>my-mqtt-to-postgresql-route-template</id>
    <postgresql-id>my-postgresql-id</postgresql-id>
    <mqtt-topic-filters>
        <mqtt-topic-filter>#</mqtt-topic-filter>
    </mqtt-topic-filters>
    <processor>
        <statement-template>path/to/my/statement-template.sql</statement-template>
    </processor>
</mqtt-to-postgresql-route>

Table 2. PostgreSQL parameters
Parameter	Required	Type	Description
`id`		ID	The ID of the `mqtt-to-postgresql-route`. This string can only contain lowercase alphanumeric characters, dashes, and underscores.
`postgresql-id`		IDREF	The ID of the PostgreSQL database.
`enabled`	-	Boolean	Optional setting that defines whether the selected `mqtt-to-postgresql-route` is enabled or disabled. The default setting is `true`. To disable the route, set to `false`.
`mqtt-topic-filters`		String	A list of one or more MQTT topic filters. `mqtt-topic-filter`: The source MQTT topic filters to select which MQTT messages are routed to the PostgreSQL database. You can define as many individual `<mqtt-topic-filter>` tags as your use case requires.
`processor`		Complex	Defines how HiveMQ converts MQTT messages into PostgreSQL rows. Two processor types are supported: `statement-template`: The path to an SQL statement template file with a valid PostgreSQL query. The path can be an absolute path or a relative path to the home folder of your HiveMQ extension for PostgreSQL. `insert-statement`: Use the XML configuration to define how to insert rows into PostgreSQL. The extension creates the query. `table`: The name of the table where rows are inserted. `batching`: Optional setting to enable or disable batching. The default setting is `true`. To disable batching, set to `false`. `columns`: A list of the table columns where values are inserted. `column`: A column name and value binding. `name`: The name of the column where the value is inserted. `value`: The MQTT property placeholder to replace and bind to the selected column.

Insert Statements

Insert statements help you convert MQTT messages into PostgreSQL rows, without the need to write your own SQL.

The insert statements provide the table name, target columns, and respective value placeholders. The extension fills the placeholders with MQTT PUBLISH properties and creates the insert query.

Batching is enabled by default for insert statements. To disable batching, set the <batching> tag in your <insert-statement> to false.

Example insert statement

<insert-statement>
    <table>mqtt_to_postgresql_table</table>
    <batching>true</batching>
    <columns>
        <column>
            <name>topic</name>
            <value>mqtt-topic</value>
        </column>
        <column>
            <name>payload_utf8</name>
            <value>mqtt-payload-utf8</value>
        </column>
        <column>
            <name>my_user_property</name>
            <value>mqtt-user-properties$myUserPropertyName</value>
        </column>
    </columns>
</insert-statement>

Statement Templates

Statement templates are SQL files that help you convert MQTT messages into PostgreSQL rows. These templates allow you to define placeholders that are later filled with the properties of the MQTT PUBLISH.

Placeholders start with ${ and end with }.

Example statement template

INSERT INTO mqtt_to_postgresql_table(topic, payload_utf8, qos, retain, packet_id, payload_format_indicator, response_topic,
                                     correlation_data_utf8, my_user_property, arrival_timestamp)
VALUES (${mqtt-topic},
        ${mqtt-payload-utf8},
        ${mqtt-qos},
        ${mqtt-retain},
        ${mqtt-packet-id},
        ${mqtt-payload-format-indicator},
        ${mqtt-response-topic},
        ${mqtt-correlation-data-utf8},
        ${mqtt-user-properties$myUserPropertyName},
        ${timestamp-iso-8601});

The HiveMQ extension automatically converts the template to a prepared statement and fills in all the placeholders.

Your HiveMQ extension automatically checks for the presence of unknown placeholders in your statement templates and insert statements. If an unknown placeholder is present, HiveMQ logs an error and the HiveMQ Extension for PostgreSQL does not start.

The following table lists all placeholders the HiveMQ PostgreSQL extension recognizes:

Table 3. Available placeholders
Name	Type	PostgreSQL recommend data type	Description
`mqtt-topic`	String	TEXT	The topic of the MQTT PUBLISH.
`mqtt-payload-utf8`	String	TEXT	The payload of the MQTT PUBLISH as a UTF-8 string.
`mqtt-payload-base64`	String	TEXT	The payload of the MQTT PUBLISH as a Base64 string.
`mqtt-qos`	String	TEXT	The QoS level of the MQTT PUBLISH (AT_MOST_ONCE, AT_LEAST_ONCE, EXACTLY_ONCE).
`mqtt-retain`	boolean	BOOLEAN	The retain flag of the MQTT PUBLISH.
`mqtt-packet-id`	int	INTEGER	The packet ID of the MQTT PUBLISH.
`mqtt-payload-format-indicator`	String	TEXT	The payload format indicator of the MQTT PUBLISH.
`mqtt-response-topic`	String	TEXT	The response topic of the MQTT PUBLISH.
`mqtt-correlation-data-utf8`	String	TEXT	The correlation data of the MQTT PUBLISH as a UTF-8 string.
`mqtt-correlation-data-base64`	String	TEXT	The correlation data of the MQTT PUBLISH as a Base64 string.
`mqtt-user-properties-json`	List	JSON	The user properties of the MQTT PUBLISH as a JSON array string.
`mqtt-user-properties$property-name`	String	TEXT	The value of the user property of the MQTT PUBLISH with the matching property name. Example: `mqtt-user-properties$myUserPropertyName` will bind the value of the first user property with the name `myUserPropertyName`.
`timestamp-ms`	long	BIGINT	The arrival timestamp of the PUBLISH message represented as a UNIX timestamp value in milliseconds. The timestamp of an incoming PUBLISH message records the moment the message arrived at the broker. The timestamp of a message created via the Extension SDK records the moment the message passed to the PublishService. The timestamp of a Will Publish message sent to subscribers records the moment the delivery of the message started.
`timestamp-iso-8601`	OffsetDateTime	TIMESTAMP	The same information as `timestamp-ms` represented as an ISO 8601 string.

Some properties in an MQTT PUBLISH message are optional. The number of placeholders the PostgreSQL extension fills varies based on the properties that are present in the MQTT PUBLISH message.

Environment variables

HiveMQ offers placeholders that can be replaced with the content of environment variables when the configuration file is read. For many use cases, it can be beneficial or necessary to use environment variables to configure items such as ports and bind addresses on the system on which you run HiveMQ. For example, when you run HiveMQ in a containerized environment.

You can use ${YOUR_ENVVAR_NAME} in the config.xml file. HiveMQ replaces the placeholder with the value of the specified environment variable during startup.

Example to set an environment variable

export MY_POSTGRESQL_PASSWORD=my-password

Example use of the environment variable in the configuration

<hivemq-postgresql-extension xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                             xsi:noNamespaceSchemaLocation="config.xsd">
    <postgresqls>
        <postgresql>
            <id>my-postgresql-id</id>
            <host>my.host.com</host>
            <port>5432</port>
            <database>my-database</database>
            <username>my-user</username>
            <password>${MY_POSTGRESQL_PASSWORD}</password>
        </postgresql>
    </postgresqls>
</hivemq-postgresql-extension>

Result of the example configuration in HiveMQ

<hivemq-postgresql-extension xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                             xsi:noNamespaceSchemaLocation="config.xsd">
    <postgresqls>
        <postgresql>
            <id>my-postgresql-id</id>
            <host>my.host.com</host>
            <port>5432</port>
            <database>my-database</database>
            <username>my-user</username>
            <password>my-password</password>
        </postgresql>
    </postgresqls>
</hivemq-postgresql-extension>

Make sure that HiveMQ is started in the same context as your environment variables are set, otherwise HiveMQ will not be able to access them.