Edits history of script submission #24006 for ' Publish message (nats)'

• bun

  Ap­pro­ved version

  Edit in Windmill

  Embed

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

  Submitted by hugo989 8 hours ago