Ctrl K

Publish message

Published today

Publish one message to an MQTT topic. A string payload is sent as-is, anything else as JSON. With QoS 1 or 2 it returns once the broker acknowledges. User properties need an MQTT 5 broker.

Script mqtt Verified

Use in Windmill

The script

Submitted by hugo989 Bun

Verified 8 hours ago

All edits

Permalink

import mqtt from "mqtt"

/**
 * Publish message
 * Publish one message to an MQTT topic. A string payload is sent as-is, anything else as JSON. With QoS 1 or 2 it returns once the broker acknowledges. User properties need an MQTT 5 broker.
 */
export async function main(
  auth: RT.Mqtt,
  topic: string,
  payload: any,
  qos: "0" | "1" | "2" = "1",
  retain: boolean = false,
  user_properties: { [key: string]: string } | undefined
) {
  if (payload === undefined) throw new Error("payload is required")
  if (/[+#]/.test(topic)) {
    throw new Error(
      `Cannot publish to ${topic}: + and # are subscription wildcards`
    )
  }
  const level = (qos != null && qos !== "" ? Number(qos) : 1) as 0 | 1 | 2
  const properties = Object.entries(user_properties ?? {})
  const tls = auth.tls?.enabled
  const ca = auth.tls?.ca_certificate?.trim()
  const host =
    auth.broker.includes(":") && !auth.broker.startsWith("[")
      ? `[${auth.broker}]`
      : auth.broker
  const client = await mqtt.connectAsync(
    `${tls ? "mqtts" : "mqtt"}://${host}:${auth.port}`,
    {
      // MQTT 5 only when user properties need it, so 3.1.1-only brokers still work.
      protocolVersion: properties.length ? 5 : 4,
      username: auth.credentials?.username || undefined,
      password: auth.credentials?.password || undefined,
      // An empty CA means the system CAs, as in Windmill's MQTT trigger.
      ca: tls && ca ? ca : undefined,
      pfx:
        tls && auth.tls?.pkcs12_client_certificate
          ? Buffer.from(auth.tls.pkcs12_client_certificate, "base64")
          : undefined,
      passphrase: auth.tls?.pkcs12_certificate_password || undefined,
      connectTimeout: 30_000,
      reconnectPeriod: 0,
    },
    false
  )
  try {
    const packet = await client.publishAsync(
      topic,
      typeof payload === "string" ? payload : JSON.stringify(payload),
      {
        qos: level,
        retain,
        properties: properties.length
          ? { userProperties: Object.fromEntries(properties) }
          : undefined,
      }
    )
    return { topic, qos: level, message_id: packet?.messageId ?? null }
  } finally {
    await client.endAsync()
  }
}


`1`	`import mqtt from "mqtt"`
`2`
`3`	`/**`
`4`	`* Publish message`
`5`	`* Publish one message to an MQTT topic. A string payload is sent as-is, anything else as JSON. With QoS 1 or 2 it returns once the broker acknowledges. User properties need an MQTT 5 broker.`
`6`	`*/`
`7`	`export async function main(`
`8`	`auth: RT.Mqtt,`
`9`	`topic: string,`
`10`	`payload: any,`
`11`	`qos: "0" \| "1" \| "2" = "1",`
`12`	`retain: boolean = false,`
`13`	`user_properties: { [key: string]: string } \| undefined`
`14`	`) {`
`15`	`if (payload === undefined) throw new Error("payload is required")`
`16`	`if (/[+#]/.test(topic)) {`
`17`	`throw new Error(`
`18`	`Cannot publish to ${topic}: + and # are subscription wildcards`
`19`	`)`
`20`	`}`
`21`	`const level = (qos != null && qos !== "" ? Number(qos) : 1) as 0 \| 1 \| 2`
`22`	`const properties = Object.entries(user_properties ?? {})`
`23`	`const tls = auth.tls?.enabled`
`24`	`const ca = auth.tls?.ca_certificate?.trim()`
`25`	`const host =`
`26`	`auth.broker.includes(":") && !auth.broker.startsWith("[")`
`27`	? `[${auth.broker}]`
`28`	`: auth.broker`
`29`	`const client = await mqtt.connectAsync(`
`30`	`${tls ? "mqtts" : "mqtt"}://${host}:${auth.port}`,
`31`	`{`
`32`	`// MQTT 5 only when user properties need it, so 3.1.1-only brokers still work.`
`33`	`protocolVersion: properties.length ? 5 : 4,`
`34`	`username: auth.credentials?.username \|\| undefined,`
`35`	`password: auth.credentials?.password \|\| undefined,`
`36`	`// An empty CA means the system CAs, as in Windmill's MQTT trigger.`
`37`	`ca: tls && ca ? ca : undefined,`
`38`	`pfx:`
`39`	`tls && auth.tls?.pkcs12_client_certificate`
`40`	`? Buffer.from(auth.tls.pkcs12_client_certificate, "base64")`
`41`	`: undefined,`
`42`	`passphrase: auth.tls?.pkcs12_certificate_password \|\| undefined,`
`43`	`connectTimeout: 30_000,`
`44`	`reconnectPeriod: 0,`
`45`	`},`
`46`	`false`
`47`	`)`
`48`	`try {`
`49`	`const packet = await client.publishAsync(`
`50`	`topic,`
`51`	`typeof payload === "string" ? payload : JSON.stringify(payload),`
`52`	`{`
`53`	`qos: level,`
`54`	`retain,`
`55`	`properties: properties.length`
`56`	`? { userProperties: Object.fromEntries(properties) }`
`57`	`: undefined,`
`58`	`}`
`59`	`)`
`60`	`return { topic, qos: level, message_id: packet?.messageId ?? null }`
`61`	`} finally {`
`62`	`await client.endAsync()`
`63`	`}`
`64`	`}`
`65`