Wraps the MQTT.js client to multiplex incoming messages to the subscribed handlers and supports querying retained topics via HTTP.
MQTT.js events can also be registered on the mqtt-topping client.
Expects that the default MQTT message payload is formatted as JSON.
- Subscribe and unsubscribe handler callbacks to individual (wildcard) topics
-
JSON.stringifyall published payloads -
JSON.parseall incoming payloads - Ignore non-JSON payloads
- Decide whether to retain a message or not depending on the topic name (retained unless topic is prefixed with
onordo) - Publishes and subscriptions are sent with quality-of-service 2
const { connectAsync } = require("@artcom/mqtt-topping")
async function main() {
const client = await connectAsync("tcp://broker.example.com")
client.on("offline", () => console.error("Client is offline. Trying to reconnect."))
await client.subscribe("my/topic", (payload, topic, packet) => {
console.log(
"Received Payload " + payload + " for Topic " + topic + " (retained = " + packet.retain + ")"
)
})
await client.publish("my/topic", "myPayload")
await client.unpublish("my/topic")
}- Works with the broker plugin "HiveMQ Retained Message Query Plugin"
- Supports single and batch queries including wildcard topics, additional options are:
-
parseJson: Parse theresult.payloadas JSON. Default istrue. -
depth: Specifies the recursive depth of the query. Adepth > 0returns subtopics inresult.children. Default is0. -
flatten: Flattens all topics into a flat array. Default isfalse.
-
- Supports single and batch json queries which:
- return entire topic trees (topics with subtopics) as one JSON object
- ignore topic payloads if subtopics exist
const { connectAsync, HttpClient } = require("@artcom/mqtt-topping")
async function main() {
const client = await connectAsync("tcp://broker.example.com")
const httpClient = new HttpClient("http://broker.example.com/query")
await client.publish("my/topic", "myPayload")
// wait a few milliseconds to ensure the data is processed on the server
const result = await httpClient.query({ topic: "my", depth: 1 })
// {
// "topic": "my",
// "children": [
// {
// "topic": "my/topic",
// "payload": "myPayload"
// }
// ]
// }
}const { connectAsync, HttpClient } = require("@artcom/mqtt-topping")
async function main() {
const client = await connectAsync("tcp://broker.example.com")
const httpClient = new HttpClient("http://broker.example.com/query")
await client.publish("my/topic1", "myPayload1")
await client.publish("my/topic2", "myPayload2")
// wait a few milliseconds to ensure the data is processed on the server
const result = await httpClient.queryBatch([{ topic: "my/topic1" }, { topic: "my/topic2" }])
// [
// {
// "topic": "my/topic1",
// "payload": "myPayload1"
// },
// {
// "topic": "my/topic2",
// "payload": "myPayload2"
// }
// ]
}const { connectAsync, HttpClient } = require("@artcom/mqtt-topping")
async function main() {
const client = await connectAsync("tcp://broker.example.com")
const httpClient = new HttpClient("http://broker.example.com/query")
await client.publish("my/topic", "myPayload")
// wait a few milliseconds to ensure the data is processed on the server
const result = await httpClient.queryJson("my")
// {
// "topic": "myPayload"
// }
}const { connectAsync, HttpClient } = require("@artcom/mqtt-topping")
async function main() {
const client = await connectAsync("tcp://broker.example.com")
const httpClient = new HttpClient("http://broker.example.com/query")
await client.publish("january/first", "eat")
await client.publish("january/second", "sleep")
await client.publish("february/first", "work")
await client.publish("february/second", "repeat")
// wait a few milliseconds to ensure the data is processed on the server
const result = await httpClient.queryJsonBatch(["january", "february"])
// [
// {
// "first": "eat"
// "second": "sleep"
// },
// {
// "first": "work"
// "second": "repeat"
// }
// ]
}const { connectAsync, HttpClient, unpublishRecursively } = require("@artcom/mqtt-topping")
async function main() {
const client = await connectAsync("tcp://broker.example.com")
const httpClient = new HttpClient("http://broker.example.com/query")
await client.publish("january/first", "eat")
await client.publish("january/second", "sleep")
await client.publish("february/first", "work")
await client.publish("february/second", "repeat")
// wait a few milliseconds to ensure the data is processed on the server
const result = await unpublishRecursively(mqttClient, httpClient, "february")
// remaining published topics on the broker
// january/first: "eat"
// january/second: "sleep"
}npm install
npm run buildThe tests require a running MQTT broker instance with the "HiveMQ Retained Message Query Plugin".
npm install
npm run build
npm run test