# rmqtt-http-api
[](https://crates.io/crates/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`:
```toml
rmqtt-http-api = "0.21"
```
Requires `rmqtt` features: `plugin`, `metrics`, `stats`, `grpc`, `shared-subscription`.
### Register
```rust
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")`.
| `max_row_limit` | `usize` | `10000` | Maximum number of rows returned by list endpoints |
| `http_laddr` | `string` | `"0.0.0.0:6060"` | HTTP server listen address |
| `http_request_log` | `bool` | `false` | Whether to print HTTP request logs |
| `http_reuseaddr` | `bool` | `true` | Enable `SO_REUSEADDR` socket option (Unix only) |
| `http_reuseport` | `bool` | `false` | Enable `SO_REUSEPORT` socket option (Unix only) |
| `http_bearer_token` | `string` | — | Bearer token for HTTP API authentication (optional) |
| `message_type` | `u8` | `99` | gRPC message type identifier for plugin communication |
| `message_expiry_interval` | `string` | `"5m"` | Default message expiration interval for publish operations |
| `metrics_sample_interval` | `string` | `"5s"` | Metrics sampling interval |
| `prometheus_metrics_cache_interval` | `string` | `"5s"` | Prometheus metrics data caching interval |
| `storage` | `object` | — | History data storage config (optional; omit to disable history) |
| `flush_interval` | `string` | `"5s"` | History flush interval |
| `history_retention` | `string` | `"7d"` | History data retention. During warmup, each data point's timestamp is checked against this duration — expired entries are discarded from the cache and removed from storage. |
### 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
```toml
# 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`.
| GET | `/` | List all available API endpoints |
| **Brokers** | | |
| GET | `/brokers` | Return basic information of all nodes in the cluster |
| GET | `/brokers/{id}` | Return basic information of a specific node |
| **Nodes** | | |
| GET | `/nodes` | Return status of all nodes in the cluster |
| GET | `/nodes/{id}` | Return status of a specific node |
| **Features** | | |
| GET | `/features` | Return feature support state of all cluster nodes |
| GET | `/features/{id}` | Return feature support state of a specific node |
| **Health** | | |
| GET | `/health/check` | Health check for the cluster |
| GET | `/health/check/{id}` | Health check for a specific node |
| **Clients** | | |
| GET | `/clients` | Search client information from the cluster |
| GET | `/clients/{clientid}` | Get specific client information |
| DELETE | `/clients/{clientid}` | Kick client from the cluster |
| GET | `/clients/{clientid}/online` | Check if a client is online |
| GET | `/clients/offlines` | Search offline client information |
| DELETE | `/clients/offlines` | Kick offline clients from the cluster |
| **Subscriptions** | | |
| GET | `/subscriptions` | Query subscription information from the cluster |
| GET | `/subscriptions/{clientid}` | Get subscriptions for a specific client |
| **Routes** | | |
| GET | `/routes` | Return all routing information from the cluster |
| GET | `/routes/{topic}` | Get routing information for a specific topic |
| **Retained Messages** | | |
| GET | `/retains` | Query retained messages with `topic_filter` / `offset` / `limit` params |
| **MQTT Operations** | | |
| POST | `/mqtt/publish` | Publish an MQTT message |
| POST | `/mqtt/subscribe` | Subscribe to an MQTT topic for a session |
| POST | `/mqtt/unsubscribe` | Unsubscribe from MQTT topics |
| **Plugins** | | |
| GET | `/plugins` | Returns information of all plugins in the cluster |
| GET | `/plugins/{node}` | Returns plugin information for a specific node |
| GET | `/plugins/{node}/{plugin}` | Get a specific plugin's info |
| GET | `/plugins/{node}/{plugin}/config` | Get a plugin's configuration |
| PUT | `/plugins/{node}/{plugin}/config/reload` | Reload a plugin's configuration |
| PUT | `/plugins/{node}/{plugin}/load` | Load/start a plugin on a node |
| PUT | `/plugins/{node}/{plugin}/unload` | Unload/stop a plugin on a node |
| **Statistics** | | |
| GET | `/stats` | Returns all statistics from the cluster |
| GET | `/stats/sum` | Summarize all statistics from the cluster |
| GET | `/stats/{id}` | Returns statistics for a specific node |
| GET | `/stats/sys` | Returns all system statistics from the cluster |
| GET | `/stats/sys/sum` | Summarize all system statistics from the cluster |
| GET | `/stats/sys/{id}` | Returns system statistics for a specific node |
| **Metrics** | | |
| GET | `/metrics` | Returns all metrics from the cluster |
| GET | `/metrics/sum` | Summarize all metrics from the cluster |
| GET | `/metrics/{id}` | Returns metrics for a specific node |
| GET | `/metrics/prometheus` | Get Prometheus metrics from the cluster |
| GET | `/metrics/prometheus/sum` | Summarize Prometheus metrics |
| GET | `/metrics/prometheus/{id}` | Get Prometheus metrics for a specific node |
### Retained Message Query
`GET /retains` queries retained messages:
| `topic_filter` | string | `#` | Topic filter supporting `#` / `+` wildcards; empty or `#` uses the full-range pagination path |
| `offset` | usize | `0` | Pagination offset |
| `limit` | usize | `max_row_limit` | Page size, capped at `max_row_limit` |
Response example:
```json
{
"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:
```json
{
"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