Edits history of script submission #23953 for ' Publish message (mqtt)'

• bun

  Ap­pro­ved version

  Edit in Windmill

  Embed

  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()
    }
  }
  

  Submitted by hugo989 9 hours ago