NATS.js - A NATS client for Node.Js

[!IMPORTANT]

Nats.node is now part of a new nats.js repo.
The new repository houses all nats clients for JavaScript.
Changes are well documented and should be easy to locate and implement, and are all described in migration.md

A Node.js client for the NATS messaging system.

Installation

npm install nats@latest

The nats.js@2.0.0 is not API compatible with previous versions of nats.js. For a migration guide, please see the migration guide.

Documentation and the NATS Base Client

This repository implements the Node.js transport of the client. This transport depends on a common module called the NATS Base Client which lives along the nats.deno. The NATS Base Client provides the same API and functionality across all JavaScript NATS clients supported by nats.io (nats.deno, nats.js and nats.ws).

While the best documentation is looking at code examples, you may want to browse the JSDoc documentation. The best entry point into the JS Doc is the NatsConnection all functionality starts with a connection.

Basics

Connecting to a nats-server

To connect to a server you use the connect() function. It returns a connection that you can use to interact with the server. You can customize the behavior of the client by specifying many ConnectionOptions.

By default, a connection will attempt a connection on127.0.0.1:4222. If the connection is dropped, the client will attempt to reconnect. You can customize the server you want to connect to by specifying port (for local connections), or full host port on the servers option. Note that the servers option can be a single hostport (a string) or an array of hostports.

The example below will attempt to connect to different servers by specifying different ConnectionOptions. At least two of them should work if your internet is working.

import { connect } from "nats";
const servers = [
    {},
    { servers: ["demo.nats.io:4442", "demo.nats.io:4222"] },
    { servers: "demo.nats.io:4443" },
    { port: 4222 },
    { servers: "localhost" },
];

for (const v of servers) {
    try {
        const nc = await connect(v);
        console.log(`connected to ${nc.getServer()}`);
        // this promise indicates the client closed
        const done = nc.closed();
        // do something with the connection

        // close the connection
        await nc.close();
        // check if the close was OK
        const err = await done;
        if (err) {
            console.log(`error closing:`, err);
        }
    } catch (err) {
        console.log(`error connecting to ${JSON.stringify(v)}`);
    }
};

To disconnect from the nats-server, call close() on the connection. A connection can also be terminated when an unexpected error happens. For example, the server returns a run-time error. In those cases, the client will re-initiate a connection.

By default, the client will always attempt to reconnect if the connection is closed for a reason other than calling close(). To get notified when the connection is closed for some reason, await the resolution of the Promise returned by closed(). If closed resolves to a value, the value is a NatsError indicating why the connection closed.

Publish and Subscribe

The basic client operations are publish to send messages and subscribe to receive messages.

Messages are published to a subject. A subject is like a URL with the exception that it doesn't specify an actual endpoint. All recipients that have expressed interest in a subject will receive messages addressed to that subject (provided they have access and permissions to get it). To express interest in a subject, you create a subscription.

In JavaScript clients (websocket, Deno, or Node) subscriptions work as an async iterator - clients simply loop to process messages as they become available.

NATS messages are payload agnostic. Payloads are Uint8Arrays. You can easily convert to and from JSON or strings by using JSONCodec or StringCodec, or a custom Codec.

To cancel a subscription and terminate your interest, you call unsubscribe() or drain() on a subscription. Unsubscribe will typically terminate regardless of whether there are messages in flight for the client. Drain ensures that all messages that are inflight are processed before canceling the subscription. Connections can also be drained as well. Draining a connection closes it, after all subscriptions have been drained and all outbound messages have been sent to the server.

import { connect, StringCodec } from "nats";

// to create a connection to a nats-server:
const nc = await connect({ servers: "demo.nats.io:4222" });

// create a codec
const sc = StringCodec();
// create a simple subscriber and iterate over messages
// matching the subscription
const sub = nc.subscribe("hello");
(async () => {
  for await (const m of sub) {
    console.log(`[${sub.getProcessed()}]: ${sc.decode(m.data)}`);
  }
  console.log("subscription closed");
})();

nc.publish("hello", sc.encode("world"));
nc.publish("hello", sc.encode("again"));

// we want to ensure that messages that are in flight
// get processed, so we are going to drain the
// connection. Drain is the same as close, but makes
// sure that all messages in flight get seen
// by the iterator. After calling drain on the connection
// the connection closes.
await nc.drain();

Wildcard Subscriptions

Subjects can be used to organize messages into hierarchies. For example, a subject may contain additional information that can be useful in providing a context to the message, such as the ID of the client that sent the message, or the region where a message originated.

Instead of subscribing to each specific subject, you can create subscriptions that have subjects with wildcards. Wildcards match one or more tokens in a subject. A token is a string following a period.

All subscriptions are independent. If two different subscriptions match a subject, both will get to process the message:

import { connect, StringCodec } from "nats";

const nc = await connect({ servers: "demo.nats.io:4222" });
const sc = StringCodec();

// subscriptions can have wildcard subjects
// the '*' matches any string in the specified token position
const s1 = nc.subscribe("help.*.system");
const s2 = nc.subscribe("help.me.*");
// the '>' matches any tokens in that position or following
// '>' can only be specified at the end of the subject
const s3 = nc.subscribe("help.>");

async function printMsgs(s) {
  let subj = s.getSubject();
  console.log(`listening for ${subj}`);
  const c = 13 - subj.length;
  const pad = "".padEnd(c);
  for await (const m of s) {
    console.log(
      `[${subj}]${pad} #${s.getProcessed()} - ${m.subject} ${
        m.data ? " " + sc.decode(m.data) : ""
      }`,
    );
  }
}

printMsgs(s1);
printMsgs(s2);
printMsgs(s3);

// don't exit until the client closes
await nc.closed();

Services: Request/Reply

Request/Reply is NATS equivalent to an HTTP request. To make requests you publish messages as you did before, but also specify a reply subject. The reply subject is where a service will publish your response.

NATS provides syntactic sugar, for publishing requests. The request() API will generate a reply subject and manage the creation of a subscription under the covers. It will also start a timer to ensure that if a response is not received within your allotted time, the request fails. The example also illustrates a graceful shutdown.

Services

Here's an example of a service. It is a bit more complicated than expected simply to illustrate not only how to create responses, but how the subject itself is used to dispatch different behaviors.

import { connect, StringCodec } from "nats";

// create a connection
const nc = await connect({ servers: "demo.nats.io" });

// create a codec
const sc = StringCodec();

// this subscription listens for `time` requests and returns the current time
const subscription = nc.subscribe("time");
(async (sub) => {
  console.log(`listening for ${sub.getSubject()} requests...`);
  for await (const m of sub) {
    if (m.respond(sc.encode(new Date().toISOString()))) {
      console.info(`[time] handled #${sub.getProcessed()}`);
    } else {
      console.log(`[time] #${sub.getProcessed()} ignored - no reply subject`);
    }
  }
  console.log(`subscription ${sub.getSubject()} drained.`);
})(subscription);

// this subscription listens for admin.uptime and admin.stop
// requests to admin.uptime returns how long the service has been running
// requests to admin.stop gracefully stop the client by draining
// the connection
const started = Date.now();
const msub = nc.subscribe("admin.*");
(async (sub) => {
  console.log(`listening for ${sub.getSubject()} requests [uptime | stop]`);
  // it would be very good to verify the origin of the request
  // before implementing something that allows your service to be managed.
  // NATS can limit which client can send or receive on what subjects.
  for await (const m of sub) {
    const chunks = m.subject.split(".");
    console.info(`[admin]