@egomobile/nats

Classes, functions and tools that help to connect and communicate to and with a NATS streaming server, written in TypeScript.

Install

Execute the following command from your project folder, where your package.json file is stored:

1npm install --save @egomobile/nats

Usage

1import {
2  INatsMessageConsumerContext,
3  NatsClient,
4  NatsMessageError
5} from "@egomobile/nats"
6
7interface IFooMessage {
8  bar: number;
9}
10
11// creates and opens an instance to a NATS
12// server using `NATS_URL`, `NATS_USER` and `NATS_PASSWORD`
13// environment variables by default
14const client = NatsClient.open({
15  "name": process.env.POD_NAME!.trim()
16})
17
18// optional:
19// https://developer.mozilla.org/en-US/docs/Web/API/AbortController
20const ac = new AbortController()
21
22const consumer = client.createConsumer<IFooMessage>({ "streamName": "foo-stream" })
23consumer.on("message", (context: INatsMessageConsumerContext<IFooMessage>) => {
24  const {
25    ack,
26    message
27  } = context
28
29  // process `message` ...
30
31  // ack() is usually executed automatically
32})
33consumer.on("error", (error: any) => {
34  if (error instanceof NatsMessageError) {
35    // is happends if handling a message failed
36    //
37    // error.msg contains `JsMsg`
38    // error.cause contains inner exception
39    console.error('Consumer message error:', error)
40  } else {
41    console.error('Consumer error:', error)
42  }
43})
44
45const disposeSubscription = consumer.subscribe({
46  signal: ac.signal
47})
48
49const publisher = client.createPublisher<IFooMessage>({ "streamName": "foo-stream" })
50await publisher.publish({
51  "bar": 42
52})
53
54setTimeout(() => {
55  ac.abort()
56
57  // alternative, if there is no AbortController:
58  //
59  // disposeSubscription()
60}, 10000)

Documentation

The API documentation can be found here.