English | 简体中文

rmqtt-http-api

RESTful HTTP API plugin. Provides broker management endpoints, health checks, node info, client/subscription/route listing, MQTT operations, plugin management, statistics, and Prometheus metrics.

Overview

Uses the salvo HTTP framework to serve API endpoints. Starts the HTTP server during plugin construction (new()). Supports hot-reload: when config changes require a restart (listen address changed), the old server is shut down after the new one starts. Forwards cluster-wide queries via gRPC.

Usage

Build

Add the dependency in rmqttd/Cargo.toml:

rmqtt-http-api = "0.21"

Requires rmqtt features: plugin, metrics, stats, grpc, shared-subscription.

Register

rmqtt_http_api::register(&scx, true, false).await?;
// or with explicit name:
rmqtt_http_api::register_named(&scx, "rmqtt-http-api", true, false).await?;

Parameters: (scx, default_startup, immutable).

Configuration

File: rmqtt-http-api.toml (in the plugin config directory). Loaded via scx.plugins.read_config_default::<PluginConfig>("rmqtt-http-api").

Configuration Source

Supports the standard RMQTT plugin config chain:

1. {plugins.dir}/rmqtt-http-api.toml (file, optional — uses defaults if missing)
2. rmqtt_plugin_rmqtt_http_api_* environment variables
3. Inline config via ServerContext::plugins_config_map_add()

Example

# rmqtt-http-api.toml

max_row_limit = 10_000

http_laddr = "0.0.0.0:6060"

http_request_log = false

message_expiry_interval = "5m"

prometheus_metrics_cache_interval = "5s"

API Endpoints

All endpoints are prefixed with /api/v1.

Retained Message Query

GET /retains queries retained messages:

Response example:

{
  "items": [
    {
      "topic": "/iot/b/x",
      "msg_id": 1024,
      "from": { "typ": "client", "id": { "node_id": 1, "client_id": "c1" } },
      "publish": {
        "topic": "/iot/b/x",
        "qos": 1,
        "retain": true,
        "dup": false,
        "payload": "<base64 encoded>",
        "create_time": 1780000000000,
        "properties": null
      },
      "remaining_ttl": 3599
    }
  ],
  "has_more": false
}

Note: the topic_filter=# (full-range) path paginates at the storage layer and includes remaining_ttl (seconds); the filtered path paginates in memory with remaining_ttl as null. Retained messages are broadcast-synced across cluster nodes, so a single-node query already covers the whole cluster.

Feature Support Query

GET /features returns the feature support state of every cluster node plus a cluster-wide consistency summary:

{
  "consistent": false,
  "node_count": 3,
  "conflicts": [
    {
      "feature": "retain",
      "values": [
        { "value": true,  "node_ids": [1, 2] },
        { "value": false, "node_ids": [3] }
      ]
    }
  ],
  "nodes": [
    {
      "node_id": 1,
      "node_name": "rmqtt@127.0.0.1",
      "features": {
        "retain": true,
        "message_storage": false,
        "session_storage": false,
        "delayed": true,
        "shared_subscription": true,
        "auto_subscription": false
      }
    }
  ]
}

• consistent: whether all reachable nodes report identical feature flags; false indicates config drift or a partially-failed plugin on some node
• conflicts: feature fields whose values differ, grouped by value with node lists (empty when consistent is true)
• nodes: per-node details; unreachable nodes appear as error strings and are excluded from the consistency check
• A features inconsistent across cluster warning log is emitted when an inconsistency is detected

Authentication

If http_bearer_token is configured, all API requests (except health check) require an Authorization: Bearer <token> header.

Dependencies

rmqtt (features: plugin, metrics, stats, grpc, shared-subscription), salvo, tokio, serde, serde_json, anyhow, base64, futures

License

MIT OR Apache-2.0