mqtt support

Author: pelikhan

package.json

@@ -41,6 +41,7 @@
     "fastify": "^4.9.2",
     "http-errors": "^2.0.0",
     "json5": "^2.2.3",
+    "mqtt": "^4.3.7",
     "node-fetch": "^2.6.7",
     "openapi-types": "^12.1.0",
     "pino-pretty": "^9.1.1",

src/apidevices.ts

@@ -114,7 +114,7 @@ export async function sendJSON(id: DeviceId, topic: string, json: any) {
     await pubToDevice(id, msg)
 }
 
-async function sendBinary(id: DeviceId, topic: string, buf: Buffer) {
+export async function sendBinary(id: DeviceId, topic: string, buf: Buffer) {
     if (buf.length > MAX_WSSK_SIZE)
         throwStatus(
             413,

src/azure/appinsights.ts

@@ -41,7 +41,7 @@ export async function setup() {
     registerMessageSink({
         name: "app insights events",
         topicName: "tev",
-        ingest: async (message, device) => {
+        ingest: async (topic, message, device) => {
             const {
                 n: name,
                 p: properties,
@@ -58,7 +58,7 @@ export async function setup() {
     registerMessageSink({
         name: "app insights metrics",
         topicName: "tme",
-        ingest: async (message, device) => {
+        ingest: async (topic, message, device) => {
             const {
                 n: name,
                 v: value,

src/azure/eventhub.ts

@@ -21,13 +21,14 @@ export async function setup() {
     registerMessageSink({
         name: "event hub",
         topicName: "*",
-        ingest: async (message, device) => {
+        ingest: async (topic, message, device) => {
             const batch = await producer.createBatch()
             const correlationId = device.sessionId
             const body = {
                 context: {
                     deviceId: device.id,
                     deviceName: device.dev.name,
+                    topic,
                 },
                 data: message,
             }

src/azure/storagequeue.ts

@@ -14,18 +14,20 @@ export async function setup() {
         return
     }
 
+    console.log("registering Azure Storage Queue message sink")
     const queueServiceClient = QueueServiceClient.fromConnectionString(connStr)
     await queueServiceClient.createQueue("messages")
     const queueClient = queueServiceClient.getQueueClient("messages")
     registerMessageSink({
         name: "storage queue",
         topicName: "*",
-        ingest: async (message, device) => {
+        ingest: async (topic, message, device) => {
             const correlationId = device.sessionId
             const body = {
                 context: {
                     deviceId: device.id,
                     deviceName: device.dev.name,
+                    topic,
                     correlationId,
                 },
                 data: message,

src/env.ts

@@ -6,7 +6,7 @@ export async function setup() {
     registerMessageSink({
         name: "environment variables",
         topicName: "env",
-        ingest: async (message: EnvironmentFromDevice, device) => {
+        ingest: async (topic: string, message: EnvironmentFromDevice, device) => {
             const { dev, id } = device
             const { envJSON } = dev
             const { fields } = message

src/index.ts

@@ -17,6 +17,7 @@ import { setup as eventHubSetup } from "./azure/eventhub"
 import { setup as storageSetup, defaultPartition } from "./storage"
 import { setup as envSetup } from "./env"
 import { setup as storageQueueSetup } from "./azure/storagequeue"
+import { setup as mqttSetup } from "./mqtt"
 
 async function initAuth(server: FastifyInstance) {
     console.log(`starting gateway...`)
@@ -126,6 +127,7 @@ async function main() {
     await storageSetup()
     await eventHubSetup()
     await storageQueueSetup()
+    await mqttSetup()
     await envSetup()
     await initAuth(server)
     await wsskInit(server)

src/messages.ts

@@ -16,7 +16,11 @@ export interface MessageSink {
      * Ingest incoming message
      * @param message
      */
-    ingest(message: Message, device: ConnectedDevice): Promise<void>
+    ingest(
+        topic: string,
+        message: Message,
+        device: ConnectedDevice
+    ): Promise<void>
 }
 
 const messageSinks: MessageSink[] = []
@@ -36,19 +40,19 @@ export function registerMessageSink(sink: MessageSink) {
  * @returns
  */
 export async function ingestMessage(
-    topicName: string,
+    topic: string,
     message: Message,
     device: ConnectedDevice
 ) {
     if (!message) return
 
     // collect sinks interrested
-    let sinks = messageSinks.filter(({ topicName: type }) => type === topicName)
+    let sinks = messageSinks.filter(({ topicName: type }) => type === topic)
     if (!sinks.length)
         sinks = messageSinks.filter(({ topicName }) => topicName === "*")
 
     // dispatch
-    await Promise.all(sinks.map(sink => sink.ingest(message, device)))
+    await Promise.all(sinks.map(sink => sink.ingest(topic, message, device)))
 }
 
 export type LogSink = (logs: string[], device: ConnectedDevice) => Promise<void>

src/mqtt.ts

@@ -0,0 +1,70 @@
+import { connect } from "mqtt" // import connect from mqtt
+import { serverTelemetry } from "./azure/appinsights"
+import { registerMessageSink } from "./messages"
+import { createSecretClient } from "./secrets"
+import { pubToDevice } from "./devutil"
+import { defaultPartition, getDevice } from "./storage"
+import { sendBinary, sendJSON } from "./apidevices"
+import { DeviceId } from "./schema"
+
+export async function setup() {
+    const secrets = createSecretClient()
+    const connectionStringSecretName =
+        process.env.DEVS_MQTT_CONNECTION_STRING_SECRET || "mqttConnectionString"
+    const connStrSecret = await secrets.getSecret(connectionStringSecretName)
+    const connStr = connStrSecret.value
+    if (!connStr) {
+        console.log("no MQTT connection string secret, skipping registration")
+        return
+    }
+
+    console.log("registering MQTT broker")
+    const telemetry = serverTelemetry()
+    const client = connect(connStr) // create a client
+
+    // device to mqtt
+    registerMessageSink({
+        name: "mqtt",
+        topicName: "*",
+        ingest: async (topic, message, device) => {
+            if (!client.connected) return
+
+            const mqTopic = `devs/from/${device.id}/${topic}`
+            client.publish(
+                mqTopic,
+                Buffer.from(JSON.stringify(message), "utf-8"),
+                { qos: 0 },
+                err => {
+                    if (err) {
+                        console.error(err)
+                        telemetry?.trackException({ exception: err })
+                    }
+                }
+            )
+        },
+    })
+
+    // dispatch messages to devices
+    client.on("connect", () => {
+        // devs/deviceid/json/topic
+        client.subscribe("devs/to/+/#", err => {
+            if (err) {
+                console.error(err)
+                telemetry?.trackException({ exception: err })
+            }
+        })
+    })
+    client.on("message", async function (topic, message) {
+        const { deviceId, msgTopic } =
+            /^devs\/(?<deviceId>.+?)\/(?<msgTopic>.+)$/.exec(topic)?.groups ||
+            {}
+        if (!deviceId || !msgTopic) return // unknown topic
+        const did: DeviceId = {
+            partitionKey: defaultPartition,
+            rowKey: deviceId,
+        }
+        const dev = await getDevice(did)
+        if (!dev) return // unknown device
+        await sendJSON(did, msgTopic, JSON.parse(message.toString("utf-8")))
+    })
+}