Edits history of script submission #23966 for ' Publish message (amqp)'

• bun

  Ap­pro­ved version

  Edit in Windmill

  Embed

  import amqp from "amqplib"
  
  /**
   * Publish message
   * Publish one message to an AMQP 0-9-1 exchange (RabbitMQ) and wait for the broker's confirm. Leave exchange empty to send straight to the queue named by routing_key. A string payload is sent as-is, anything else as JSON. routed is false when no queue received it.
   */
  export async function main(
    auth: RT.Amqp,
    routing_key: string,
    payload: any,
    exchange: string | undefined,
    headers: { [key: string]: string } | undefined,
    persistent: boolean = false
  ) {
    if (payload === undefined) throw new Error("payload is required")
    const anonymous = !auth.username && !auth.password
    const conn = await amqp.connect({
      protocol: auth.tls ? "amqps" : "amqp",
      hostname: auth.host,
      port: auth.port || (auth.tls ? 5671 : 5672),
      // guest/guest is amqplib's default, but only for a resource with no credentials at all.
      username: anonymous ? "guest" : (auth.username ?? ""),
      password: anonymous ? "guest" : (auth.password ?? ""),
      // amqplib unescapes the vhost: encode it, like Windmill's AMQP trigger does.
      vhost: encodeURIComponent(auth.vhost || "/"),
    })
    try {
      // amqplib rethrows an 'error' event nobody listens to, and doesn't settle pending
      // confirms when that happens: race the confirm against the connection's and the
      // channel's errors (unknown exchange, access refused, broker closing the connection).
      const failed = new Promise<never>((_, reject) => conn.once("error", reject))
      failed.catch(() => {})
      const channel = await Promise.race([conn.createConfirmChannel(), failed])
      const closed = new Promise<never>((_, reject) =>
        channel.once("error", reject)
      )
      closed.catch(() => {})
      // mandatory: the broker returns a message no queue took, before its confirm.
      let routed = true
      channel.on("return", () => (routed = false))
      const text = typeof payload === "string"
      channel.publish(
        exchange || "",
        routing_key,
        Buffer.from(text ? payload : JSON.stringify(payload)),
        {
          mandatory: true,
          persistent,
          contentType: text ? "text/plain" : "application/json",
          headers: Object.fromEntries(
            Object.entries(headers ?? {}).map(([k, v]) => [
              k,
              typeof v === "string" ? v : JSON.stringify(v),
            ])
          ),
        }
      )
      await Promise.race([channel.waitForConfirms(), failed, closed])
      return { exchange: exchange || "", routing_key, routed }
    } finally {
      // Closing an already-closed connection throws, which would hide the real error.
      await conn.close().catch(() => {})
    }
  }
  

  Submitted by hugo989 8 hours ago