pgmqtt

pgmqtt : CDC-to-MQTT broker for PostgreSQL

Overview

Packages

Source

pig build pkg pgmqtt;		# build rpm/deb

Install

Make sure PGDG and PIGSTY repo available:

pig repo add pgsql -u   # add both repo and update cache

Install this extension with pig:

pig install pgmqtt;		# install via package name, for the active PG version

pig install pgmqtt -v 18;   # install for PG 18
pig install pgmqtt -v 17;   # install for PG 17
pig install pgmqtt -v 16;   # install for PG 16
pig install pgmqtt -v 15;   # install for PG 15
pig install pgmqtt -v 14;   # install for PG 14

Create this extension with:

CREATE EXTENSION pgmqtt;

Usage

Sources: README, interfaces, configuration, limitations, Cargo.toml

pgmqtt is a pgrx extension that embeds an MQTT broker into PostgreSQL and uses change data capture to turn table changes into MQTT messages. It also supports inbound topic mappings so MQTT publishes can insert rows into PostgreSQL tables.

CREATE EXTENSION pgmqtt;

Outbound Mapping

Publish table changes to topics:

SELECT pgmqtt_add_outbound_mapping(
  'public',
  'my_table',
  'topics/{{ op | lower }}',
  '{{ columns | tojson }}',
  1
);

With that mapping, INSERT, UPDATE, and DELETE publish JSON payloads to topics such as topics/insert. The documented function signature also accepts optional qos integer DEFAULT 0 and template_type text DEFAULT 'jinja2'.

Inbound Mapping

Insert rows from MQTT publishes:

SELECT pgmqtt_add_inbound_mapping(
  'sensor/{site_id}/temperature',
  'sensor_readings',
  '{"site_id": "{site_id}", "value": "$.temperature"}'::jsonb
);

Publishing {"temperature": 22.5} to sensor/site-1/temperature inserts a row into sensor_readings.

Inbound mappings can also perform upsert and delete operations by passing op, conflict_columns, target_schema, mapping_name, and template_type. Topic patterns use {variable} captures; JSON payload fields use expressions such as $.temperature, $payload, and $topic.

Inspect and Remove Mappings

SELECT * FROM pgmqtt_list_outbound_mappings();
SELECT pgmqtt_remove_outbound_mapping('public', 'my_table');

SELECT * FROM pgmqtt_list_inbound_mappings();
SELECT pgmqtt_remove_inbound_mapping('temp_readings');

SELECT * FROM pgmqtt_status();

pgmqtt_status() reports active connections, subscriptions, retained messages, pending session messages, CDC mappings, CDC slot state, inbound mappings, pending inbound writes, and dead letters.

MQTT Client Examples

mosquitto_sub -h localhost -t 'topics/#'
mosquitto_pub -h localhost -t 'sensor/site-1/temperature' -m '{"temperature": 22.5}'

Configuration

The documented GUCs live under the pgmqtt namespace:

ALTER SYSTEM SET pgmqtt.cdc_every_n_ticks = 16;
SELECT pg_reload_conf();

Listener GUCs include pgmqtt.mqtt_enabled, pgmqtt.mqtt_port (1883), pgmqtt.ws_enabled, pgmqtt.ws_port (9001), pgmqtt.mqtts_enabled, pgmqtt.mqtts_port (8883), pgmqtt.wss_enabled, and pgmqtt.wss_port (9002). TLS and authentication settings include pgmqtt.tls_cert_file, pgmqtt.tls_key_file, pgmqtt.license_key, pgmqtt.jwt_public_key, pgmqtt.jwt_required, and pgmqtt.jwt_required_ws.

Performance and observability GUCs include pgmqtt.tick_interval_ms, pgmqtt.max_client_buffer_bytes, pgmqtt.cdc_every_n_ticks, pgmqtt.debug_log, pgmqtt.metrics_snapshot_interval, pgmqtt.metrics_retention_days, pgmqtt.metrics_connections_cache_interval, pgmqtt.metrics_hook_function, and pgmqtt.metrics_notify_channel. Listener and TLS settings are read when the MQTT background worker starts, so they require a worker restart rather than only pg_reload_conf().

Caveats

• The README requires wal_level = logical; without logical decoding the CDC side will not work.
• This project’s CSV tracks version 0.1.0, PostgreSQL versions 14-18, and a package-side pgrx 0.17.0 rebuild note. Upstream Cargo.toml still advertises older build defaults, so use the CSV as the package/platform authority here.
• MQTT 5.0 and MQTT 3.1.1 clients are documented as supported. QoS 0 and QoS 1 are supported; QoS 2 is not implemented and subscriptions requesting QoS 2 are downgraded to QoS 1.
• CDC captures INSERT, UPDATE, and DELETE; DDL changes and TRUNCATE are not captured. DELETE may require REPLICA IDENTITY FULL.