Publish message

import {

  connect,

  headers as natsHeaders,

  jwtAuthenticator,

  nkeyAuthenticator,

} from "nats"

/**

 * Publish message

 * Publish one message to a NATS subject, with optional headers. A string payload is sent as-is, anything else as JSON. With jetstream on, it waits for the stream's acknowledgement and returns the stream and sequence.

 */

export async function main(

  auth: RT.Nats,

  subject: string,

  payload: any,

  headers: { [key: string]: string } | undefined,

  jetstream: boolean = false

) {

  if (payload === undefined) throw new Error("payload is required")

  // nats.js falls back to 127.0.0.1:4222 when given no server.

  const servers = (auth.servers ?? []).map((s) => s.trim()).filter((s) => s)

  if (!servers.length) throw new Error("The nats resource lists no servers")

  const a = auth.auth ?? { label: "NO_AUTH" }

  // Seeds and JWTs are often pasted from a .creds file with a trailing newline.

  const seed = (s: string) => new TextEncoder().encode(s.trim())

  const nc = await connect({

    servers,

    tls: auth.require_tls ? {} : undefined,

    token: a.label === "TOKEN" ? a.token : undefined,

    user: a.label === "USER_PASSWORD" ? a.user : undefined,

    pass: a.label === "USER_PASSWORD" ? a.password : undefined,

    authenticator:

      a.label === "NKEY"

        ? nkeyAuthenticator(seed(a.seed))

        : a.label === "JWT"

          ? jwtAuthenticator(a.jwt.trim(), seed(a.seed))

          : undefined,

    maxReconnectAttempts: 0,

  })

  try {

    const data = new TextEncoder().encode(

      typeof payload === "string" ? payload : JSON.stringify(payload)

    )

    const entries = Object.entries(headers ?? {})

    const h = entries.length ? natsHeaders() : undefined

    for (const [k, v] of entries) h!.append(k, String(v))

    if (jetstream) {

      const ack = await nc.jetstream().publish(subject, data, { headers: h })

      return {

        subject,

        stream: ack.stream,

        seq: ack.seq,

        duplicate: ack.duplicate,

      }

    }

    nc.publish(subject, data, { headers: h })

    // A round trip to the server: the message has left and was accepted.

    await nc.flush()

    return { subject }

  } finally {

    await nc.close()

  }

}