Publish message

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

  }

}