added events

src/events/base-listener.ts

@@ -0,0 +1,43 @@
+import { Stan, Message } from 'node-nats-streaming';
+import { Subjects } from './subjects';
+interface Event {
+  subject: Subjects;
+  data: any;
+}
+abstract class Listener<T extends Event> {
+  abstract subject: T['subject'];
+  abstract queueGroupName: string;
+  abstract onMessage(data: T['data'], msg: Message): void;
+  private client: Stan;
+  protected ackWait = 5 * 1000;
+
+  constructor(client: Stan) {
+    this.client = client;
+  }
+
+  subscriptionOptions() {
+    return this.client.subscriptionOptions()
+        .setDeliverAllAvailable()
+        .setManualAckMode(true)
+        .setDurableName(this.queueGroupName);
+  }
+
+  listen() {
+    const subscription = this.client.subscribe(this.subject, this.queueGroupName,
+        this.subscriptionOptions());
+
+    subscription.on('message', (msg: Message) => {
+      console.log(`Message received: ${this.subject} / ${this.queueGroupName}`);
+
+      const parsedData = this.parseMessage(msg);
+      this.onMessage(parsedData, msg);
+    });
+  }
+
+  parseMessage(msg: Message) {
+    const data = msg.getData();
+    return typeof data === 'string' ? JSON.parse(data) : JSON.parse(data.toString('utf8'));
+  }
+}
+
+export {Listener};

src/events/base-publisher.ts

@@ -0,0 +1,28 @@
+import { Stan } from 'node-nats-streaming';
+import { Subjects } from './subjects';
+
+interface Event {
+  subject: Subjects;
+  data: any;  
+} 
+
+export abstract class Publisher<T extends Event> {
+  abstract subject: T['subject'];
+  private client: Stan;
+
+  constructor(client: Stan) {
+    this.client = client;
+  }
+
+  publish(data: T['data']): Promise<void> {
+    return new Promise((resolve, reject) => {
+      this.client.publish(this.subject, JSON.stringify(data), (err) => {
+        if (err) {
+          return reject(err);
+        }
+        console.log('Event published to subject', this.subject);
+        resolve();
+      });
+    })
+  }
+}

src/events/subjects.ts

@@ -0,0 +1,4 @@
+export enum Subjects {
+  TicketCreated = 'ticket:created',
+  OrderUpdated = 'order:updated'
+}

src/events/ticket-created-event.ts

@@ -0,0 +1,10 @@
+import { Subjects } from './subjects';
+
+export interface TicketCreatedEvent {
+  subject: Subjects.TicketCreated;
+  data: {
+    id: string;
+    title: string;
+    price: number;
+  };
+}