AuI NATS Wrapper

The purpose of this package is to wrap the NATS module to serve micro-services pattern.

Install

npm i --save @aui.io/aui-nats

API Documentation

Connect to NATS

// src/index.ts
import { natsWrapper } from '@aui.io/aui-nats'

...
...

try {
    await natsWrapper.connect(
      process.env.NATS_CLUSTER_ID,
      process.env.NATS_CLIENT_ID,
      process.env.NATS_URL
    );
    natsWrapper.client.on('close', () => {
      console.log('NATS connection closed!');
      process.exit();
    });
    process.on('SIGINT', () => natsWrapper.client.close());
    process.on('SIGTERM', () => natsWrapper.client.close());

    ...
    ...

  } catch (err) {
    console.error(err);
  }

Create an event and add a subject (in the cluster common directory)

// subjects.ts
export enum Subjects {
  YourCustomMethod = "service-name:model-name:created",
}

// your-custom-event.ts
import { Subjects } from "./subjects";

export interface YourCustomEvent {
  subject: Subjects.YourCustomMethod;
  data: {
    id: string;
    version: number;
    ...
    ...
    createdAt: Date;
    updatedAt?: Date;
  };
}

Create a publisher in the publishing service (/src/events/publishers)

// your-custom-publisher.ts
import { Publisher } from '@aui.io/aui-nats';
import { Subjects, YourCustomEvent } from 'YOUR_CLUSTER_COMMON_LIBRARY';

export class YourCustomPublisher extends Publisher<YourCustomEvent> {
  // readonly subject = Subjects.YourCustomMethod;
}

When the item added/updated/etc.. on the DB, publish an event

import { natsWrapper } from '@aui.io/aui-nats';
import { Subjects } from 'YOUR_CLUSTER_COMMON_LIBRARY';
import { YourCustomPublisher } from '../events/publishers/your-custom-publisher';

new YourCustomPublisher(natsWrapper.client, Subjects.YourCustomMethod).publish({
      id: ...,
      ...
      ...
      createdAt: ...,
      updatedAt: ...,
      version: ...
    }); // type = YourCustomEvent["data"]

Create a listener in the listening service (/src/events/listeners)

import { Listener, Message } from '@aui.io/aui-nats';
import { Subjects, YourCustomEvent } from 'YOUR_CLUSTER_COMMON_LIBRARY';
import { queueGroupName } from './queue-group-name';
import { YourSchema } from '../../models/your-schema';

export class YourCustomListener extends Listener<YourCustomEvent> {
  readonly subject = Subjects.YourCustom;
  queueGroupName = queueGroupName;

  async onMessage(data: YourCustomEvent['data'], msg: Message) {
    const {
      id,
      version,
      ...
      ...
      createdAt,
      updatedAt
    } = data;

    // save in the db
    const yourSchema = YourSchema.build({
      id,
      version,
      ...
      ...
      createdAt,
      updatedAt
    });
    await yourSchema.save();
    // publish on Kinesis
    console.log('publish to kinesis', yourSchema);

    msg.ack();
  }
}

Import the listener in the index.ts

import { natsWrapper } from '@aui.io/aui-nats';
import { Subjects } from 'YOUR_CLUSTER_COMMON_LIBRARY';
import { YourCustomListener } from './events/listeners/your-custom-listener';

...
...
// under the following code
await natsWrapper.connect(
  process.env.NATS_CLUSTER_ID,
  process.env.NATS_CLIENT_ID,
  process.env.NATS_URL
);
natsWrapper.client.on('close', () => {
  console.log('NATS connection closed!');
  process.exit();
});
process.on('SIGINT', () => natsWrapper.client.close());
process.on('SIGTERM', () => natsWrapper.client.close());

// add the listener
new YourCustomListener(natsWrapper.client, Subjects.YourCustomMethod).listen();

Dependencies

"node-nats-streaming": "^0.3.2"

Contributors

• Esmat Nawahda

Contributing

Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change. Please make sure to update tests as appropriate.

License

Please see License File for more information.