work out of the box with mqtt broker

Author: pelikhan

local.env

@@ -5,4 +5,5 @@ DEVS_LOCALHOST="1"
 DEVS_STORAGE_CONNECTION_STRING="DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;QueueEndpoint=http://127.0.0.1:10001/devstoreaccount1;TableEndpoint=http://127.0.0.1:10002/devstoreaccount1;"
 DEVS_PASSWORDS_SECRET="DEVS_LOCAL_USER_PASSWORD"
 DEVS_LOCAL_USER_PASSWORD="devstoreaccount1:Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
-DEVS_CONNECTION_STRING="AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;ApiRoot=$WEBSITE_PROTOCOL://$WEBSITE_HOSTNAME"
+DEVS_CONNECTION_STRING="AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;ApiRoot=$WEBSITE_PROTOCOL://$WEBSITE_HOSTNAME"
+DEVS_MQTT_CONNETION_STRING="mqtt://public.mqtthq.com:1883"

src/azure/storagequeue.ts

@@ -3,7 +3,7 @@ import { serverTelemetry } from "./appinsights"
 import { registerMessageSink } from "../messages"
 import { readStorageConnectionString } from "../storage"
 
-const MESSAGE_TIME_TO_LIVE = 360
+const MESSAGE_TIME_TO_LIVE = 60
 
 export async function setup() {
     const connStr = await readStorageConnectionString()

src/index.ts

@@ -10,7 +10,7 @@ import * as mq from "./mq"
 import { wsskInit } from "./wssk"
 import { fwdSockInit } from "./fwdsock"
 
-import { createSecretClient } from "./secrets"
+import { getSecret } from "./secrets"
 import { generateOpenApiSpec } from "./swagger/openapi"
 import { setup as appInsightsSetup, serverTelemetry } from "./azure/appinsights"
 import { setup as eventHubSetup } from "./azure/eventhub"
@@ -21,13 +21,13 @@ import { setup as mqttSetup } from "./mqtt"
 
 async function initAuth(server: FastifyInstance) {
     console.log(`starting gateway...`)
-    const secrets = createSecretClient()
-    const passwordSecretName =
-        process.env["DEVS_PASSWORDS_SECRET"] || "passwords"
-    const passwordsSecret = await secrets.getSecret(passwordSecretName)
-    if (!passwordsSecret.value) throw new Error("passwords is empty")
+    const passwordsSecret = await getSecret(
+        "passwords",
+        "DEVS_PASSWORDS_SECRET"
+    )
+    if (!passwordsSecret) throw new Error("passwords is empty")
 
-    const passwords = passwordsSecret.value
+    const passwords = passwordsSecret
         .split(/,/)
         .map(s => s.trim())
         .filter(s => /^\w+:.+/.test(s))

src/mqtt.ts

@@ -1,18 +1,17 @@
 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 { getSecret } from "./secrets"
 import { defaultPartition, getDevice } from "./storage"
-import { sendBinary, sendJSON } from "./apidevices"
+import { 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
+    const connStr = await getSecret(
+        "mqttConnectionString",
+        "DEVS_MQTT_CONNECTION_STRING_SECRET",
+        "DEVS_MQTT_CONNETION_STRING"
+    )
     if (!connStr) {
         console.log("no MQTT connection string secret, skipping registration")
         return
@@ -51,6 +50,8 @@ export async function setup() {
             if (err) {
                 console.error(err)
                 telemetry?.trackException({ exception: err })
+            } else {
+                console.log(`mqtt: subscribe to 'devs/to/+/#'`)
             }
         })
     })
@@ -65,6 +66,7 @@ export async function setup() {
         }
         const dev = await getDevice(did)
         if (!dev) return // unknown device
+
         await sendJSON(did, msgTopic, JSON.parse(message.toString("utf-8")))
     })
 }

src/secrets.ts

@@ -17,3 +17,18 @@ export function createSecretClient(): Secrets {
         },
     }
 }
+
+export async function getSecret(
+    defaultName: string,
+    secretNameEnvironmentName: string,
+    environmentName?: string
+) {
+    const secrets = createSecretClient()
+    const connectionStringSecretName =
+        process.env[secretNameEnvironmentName] || defaultName
+    const connStrSecret = await secrets.getSecret(connectionStringSecretName)
+    const connStr =
+        connStrSecret.value ||
+        (environmentName ? process.env[environmentName] : undefined)
+    return connStr
+}

src/storage.ts

@@ -16,7 +16,7 @@ import { gunzip, gzip } from "zlib"
 import { pubToDevice } from "./devutil"
 import { DeviceId, DeviceInfo, DeviceStats, zeroDeviceStats } from "./schema"
 import { delay, throwStatus } from "./util"
-import { createSecretClient } from "./secrets"
+import { getSecret } from "./secrets"
 import { DebugInfo } from "./interop"
 import { createHash } from "crypto"
 
@@ -37,13 +37,11 @@ export interface ScriptBody {
 }
 
 export async function readStorageConnectionString() {
-    const secrets = createSecretClient()
-    const connectionStringSecretName =
-        process.env["DEVS_STORAGE_CONNECTION_STRING_SECRET"] ||
-        "storageAccountConnectionString"
-    const connStrSecret = await secrets.getSecret(connectionStringSecretName)
-    const connStr =
-        connStrSecret.value || process.env.DEVS_STORAGE_CONNECTION_STRING
+    const connStr = getSecret(
+        "storageAccountConnectionString",
+        "DEVS_STORAGE_CONNECTION_STRING_SECRET",
+        "DEVS_STORAGE_CONNECTION_STRING"
+    )
     return connStr
 }