Data Integration

Data integration

The webhook extension posts what the broker sees to an HTTP(S) endpoint. It has two sources - the MQTT messages the broker accepts, and the broker's own log - and one or more destinations, each with its own URL, body template, filters and retries. Every request carries an HMAC-SHA256 signature when a secret is configured.

It is an extension, not a core feature: a broker with no webhook loaded pays nothing for it. It is built for Linux, FreeBSD and Windows: the socket layer uses Winsock on Windows and the platform's POSIX sockets elsewhere, and TLS is OpenSSL on all of them. On Windows the installed fragment nameslib\xmq-extensions\xmq_webhook.dll; on the other platforms the library is libxmq_webhook.so.

All settings are edited on the Extensions page, which builds the form from what the extension declares: a URL is a URL, a number is a number, and a value that does not have the declared format is refused before anything is written. The same checks run in the broker, so a value written by a script is refused for the same reason. The destinations array is the exception: it is one JSON document inside one setting, and is described below.

Enabling it

The package installs 50-webhook.conf.template beside the other extension fragments. Copy it to 50-webhook.conf and setenabled to true, or switch the extension on from theExtensions page once it is configured:

{
  "extensions": [{
    "name": "webhook",
    "library": "/usr/local/lib/xmq-extensions/libxmq_webhook.so",
    "enabled": true,
    "settings": {
      "url": "https://collector.example/hook",
      "secret": "a shared secret",
      "topic_filter": "sensors/#",
      "max_payload": "1048576",
      "timeout_ms": "5000",
      "retries": "3",
      "concurrency": "4",
      "allow_http": "false"
    }
  }]
}

Sources

Messages. A message is offered to the extension when a client publishes it and the broker accepts it - that is, after authentication and after the topic permission check. A publication the broker refuses is not sent. The body is copied only after the topic filter matched, so a connector watching one topic costs nothing for the rest of the broker's traffic. Retained publications are sent like any other, with"retain": true, and a retained message that clears a topic is sent with an empty body.

Messages the broker produces itself are not tapped: the $SYScounters, retained-message synchronisation between cluster nodes, copies forwarded by a cluster node or a bridge, and the last will. This is deliberate, because one publication otherwise arrives several times - once where it was published and again on every node it was forwarded to.

Log records. The second source is the broker's own log, so one endpoint can be told both what clients sent and what the broker could not do. It is not an MQTT topic such as $SYS/error: the faults worth reporting are often the ones that stop MQTT from working - a refused connection, an unreachable account database - and routing them through the same broker and its ACLs would make the report depend on the thing that failed. Choose the lowest level worth sending and, optionally, the subjects.

Both sources are independent and share one endpoint; every body says which it is with "source". The two taps have separate queues and separate threads, so a receiver that cannot keep up with log records does not delay messages, and the other way round.

What is sent

{
  "id": "1762790400123-7",
  "source": "message",
  "version": 1,
  "timestamp": "2026-11-03T19:45:00Z",
  "topic": "sensors/water/1",
  "qos": 1,
  "retain": false,
  "client_id": "device-1",
  "username": "user7",
  "payload_base64": "aGVsbG8="
}

The payload is base64 because MQTT bodies are binary and JSON strings are not.X-XMQ-Signature: sha256=<hex> is the HMAC-SHA256 of the exact request body, keyed with secret. A receiver should compare it in constant time and reject a request without it when one is configured.

A log record looks like this:

{
  "id": "1762790400123-8",
  "source": "log",
  "version": 1,
  "timestamp": "2026-11-03T19:45:01Z",
  "level": "error",
  "subject": "server_connections",
  "message": "test-client: refused by an extension"
}

log_subjects is a comma-separated list of the broker's log categories (connect, publish, server_connections,server_events, storage_events,cluster_connections, cluster_events); empty means all of them. Refused publications and failed logins are the records that matter most, and they arrive even when no client can connect.

Configuration

These are the settings of one connector with a single destination. Each is checked on the Extensions page and again by the broker; the format column is the regular expression the value must match as a whole.

_comment keys in the shipped template are not settings and are not edited on the page; they are there to explain the file to somebody reading it.name and library identify the extension itself and are edited in the configuration file, not in the settings form.

Several destinations

One connector can post to more than one endpoint, each with its own body template and its own filters. Set destinations to a JSON array; it replaces the single-URL settings entirely:

"destinations": "[
  {"url": "https://ntfy.sh/xmq-errors",
   "sources": "log",
   "log_min_level": "error",
   "template": "[{{level}}] {{subject}}: {{message}}"},

  {"url": "https://collector.example/hook",
   "sources": "messages",
   "topic_filter": "sensors/#",
   "secret": "a shared secret"}
]"

The whole array is one setting, so its quotes are escaped in the configuration file; the first destination above sends a plain-text body, the second the standard JSON envelope. A messenger that wants JSON of its own - Slack's{"text": ...}, for instance - needs that JSON as the template, which means a string inside a string.

With no template the body is the JSON envelope shown above; with one, the body is the template with these names replaced: {{id}},{{source}}, {{timestamp}},{{topic}}, {{qos}}, {{retain}},{{client_id}}, {{username}},{{payload_base64}}, {{level}},{{subject}}, {{message}}.

Substituted values are JSON-escaped, so a quote or a newline in a message cannot break the body. An unknown name is left as written, which the endpoint will reject rather than accept as empty.

Prototype caveat. destinations is a JSON document inside a string setting: the settings screen shows it as one field, and asecret inside it is not masked the way the single-URLsecret is, because the interface masks whole values, not parts of one. Separate connector instances, each with its own masked settings, are the shape this should grow into.

Performance limits

The connector is built so that a slow or unreachable endpoint costs deliveries and not broker throughput. What that means in practice:

• The publisher never waits. An accepted message copies its body into the extension's queue and the publish path continues. There is no backpressure: a publisher cannot be slowed down by a receiver, and messages do not wait in the broker for it either.
• The message queue holds 4096 messages, and the log queue holds 4096 records, each per extension instance. When one is full the oldest work is dropped - the newest message is the one refused - and the drop is counted. The counters for each tap are shown on the Extensions page asmessages delivered / dropped and log records delivered / dropped. A non-zero drop count means the endpoint or the network is slower than the traffic that matches the filter.
• Each destination has a queue of 1024 bodies and concurrencyworker threads. A worker waits for one synchronous HTTP reply, so a destination through a 10 ms round trip carries about100 × concurrency bodies a second at best. Raiseconcurrency before raising timeouts.
• Memory is bounded by messages, not by bytes. With the defaultmax_payload of 1 MiB, a stalled destination can hold 1024 bodies of up to 1 MiB - and the tap queue 4096 more. Set max_payload to the largest message the deployment actually sends; that is what keeps a stalled endpoint from turning into a memory problem. 0 removes the limit altogether and should not be used on a broker that accepts untrusted publications.
• Base64 adds a third to the body. A 1 MiB payload travels as about 1.4 MB of JSON, and the queue counts the message once, before encoding.
• The tap is registered once for all destinations, with the widest filter any of them named. Each matching body is copied once for the extension, and the per-destination filters narrow it afterwards. That is the price of several destinations in one connector.
• Delivery is at-most-once. The queues are in memory and a restart loses what has not been sent. A 2xx is success; 5xx, 429, a timeout or a connection failure is retried with backoff; any other 4xx is refused without a retry.

For sizing: the extension costs one copy of each matching body on the publish path, plus the queued copies while a destination is behind. A broker whose only work is forwarding small messages will not notice it; a broker forwarding large binary payloads at a high rate should name a narrow topic_filter and a realistic max_payload, and watch the drop counters.

Security

• Use an HTTPS endpoint with a certificate the broker trusts; the system CA store is used and verification is mandatory.
• Treat secret as a password: it is a secret setting, masked in the interface and never logged.
• A URL set by an administrator can reach internal services; restrict the endpoint or the network if that matters.
• The payload is the message body: do not point this at an endpoint that logs bodies containing credentials.
• The destinations array is stored as one string, so a secret inside it is visible to anyone who can read the configuration file. Prefer the single-URL form, whose secret is masked, until the connector grows separate instances.

What it is not

This connector does not transform messages, does not speak Kafka or a database, and does not replay. Those are further extensions on the same message tap, not settings of this one.