feat: add multiconnection

tinybudgie · Sep 28, 2023 · 019ee30 · 019ee30
1 parent 0b74307
commit 019ee30
Show file tree

Hide file tree

Showing 10 changed files with 365 additions and 163 deletions.
diff --git a/libs/external/nats/src/lib/nats-configs/nats-module.config.ts b/libs/external/nats/src/lib/nats-configs/nats-module.config.ts
@@ -2,16 +2,25 @@ import { ConfigurableModuleBuilder } from '@nestjs/common'
 import { merge } from 'lodash'
 import { ConnectionOptions } from 'nats'
 
-export interface NatsConfig extends ConnectionOptions {
+export interface NatsConnectionConfig extends ConnectionOptions {
+    /**
+     * Connection name, provide it if you want to support different NATS connection (NOT clusters)
+     */
+    connectionName?: string
+
     /**
      * @default: true
      */
     enableJetstream?: boolean
+}
+
+export interface NatsConfig {
+    connections: NatsConnectionConfig[]
 
     /**
      * When set to `true` the client will print protocol messages that it receives
      * or sends to the server using NestJS logger. If you want to debug logs from `nats.io` directly
-     * use `debug: true`
+     * use `debug: true` in `NatsConnectionConfig`
      */
     debugLog?: {
         enable:
@@ -26,11 +35,7 @@ export interface NatsConfig extends ConnectionOptions {
     }
 }
 
-export const DEFAULT_NATS_CONFIG: Pick<
-    NatsConfig,
-    'enableJetstream' | 'debugLog'
-> = {
-    enableJetstream: true,
+export const DEFAULT_NATS_CONFIG: Pick<NatsConfig, 'debugLog'> = {
     debugLog: {
         enable: false,
     },

diff --git a/libs/external/nats/src/lib/nats-constants/nats.constants.ts b/libs/external/nats/src/lib/nats-constants/nats.constants.ts
@@ -8,3 +8,5 @@ export const CONSUME_ARGS_METADATA = 'nats.consume.args.metadata'
 export const CONSUME_PAYLOAD_TYPE = 3
 export const CONSUME_HEADER_TYPE = 4
 export const CONSUME_ACK_TYPE = 5
+
+export const DEFAULT_CONNECTION_NAME = 'nats-default-connection'
diff --git a/libs/external/nats/src/lib/nats-decorators/reply.decorator.ts b/libs/external/nats/src/lib/nats-decorators/reply.decorator.ts
@@ -3,6 +3,12 @@ import { applyDecorators, SetMetadata } from '@nestjs/common'
 import { REPLY_METADATA } from '../nats-constants/nats.constants'
 import { SubscriptionOptions } from '../nats-interfaces/nats.interfaces'
 
-export function Reply(subject: string, options?: SubscriptionOptions) {
-    return applyDecorators(SetMetadata(REPLY_METADATA, { subject, options }))
+export function Reply(
+    subject: string,
+    options?: SubscriptionOptions,
+    connectionName?: string,
+) {
+    return applyDecorators(
+        SetMetadata(REPLY_METADATA, { subject, options, connectionName }),
+    )
 }
diff --git a/libs/external/nats/src/lib/nats-errors/nats-errors.enum.ts b/libs/external/nats/src/lib/nats-errors/nats-errors.enum.ts
@@ -4,6 +4,7 @@ export enum NatsErrorsEnum {
     NoConnection = 'NATS-002',
     JetStreamNotEnabled = 'NATS-003',
     JetStreamNotEnabledConfig = 'NATS-004',
+    UnknownConnectionName = 'NATS-005',
 }
 
 export const NATS_ERROR_TITLES = {
@@ -12,4 +13,5 @@ export const NATS_ERROR_TITLES = {
     [NatsErrorsEnum.NoConnection]: `No connection established to NATS`,
     [NatsErrorsEnum.JetStreamNotEnabled]: `JetStream not enabled. Please provide '--jetstream' option to your nats server`,
     [NatsErrorsEnum.JetStreamNotEnabledConfig]: `JetStream not enabled. Please provide 'enableJetStream: true' option to your nats module config`,
+    [NatsErrorsEnum.UnknownConnectionName]: `Provide connection name if you use multiple NATS connections`,
 }
diff --git a/libs/external/nats/src/lib/nats-indicators/nats-connection-health.indicator.ts b/libs/external/nats/src/lib/nats-indicators/nats-connection-health.indicator.ts
@@ -13,17 +13,38 @@ export class NatsConnectionHealthIndicator {
     @HealthIndicator('nats')
     async isHealthy(): Promise<HealthIndicatorResult> {
         try {
-            const status = this.natsConnection.status()
+            const allConnections = this.natsConnection.getAllConnections()
+            let allConnected = true
+            const details: Record<string, any> = []
 
-            if (status.connected) {
+            for (const connection of allConnections) {
+                const status = this.natsConnection.status(connection.name)
+
+                if (!status.connected) {
+                    allConnected = false
+                    details.push({
+                        connectionName: connection.name,
+                        connected: false,
+                        error: status.error,
+                    })
+                } else {
+                    details.push({
+                        connectionName: connection.name,
+                        connected: true,
+                    })
+                }
+            }
+
+            if (allConnected) {
                 return {
                     status: 'up',
+                    details,
                 }
             }
 
             return {
                 status: 'down',
-                error: status.error,
+                details,
             }
         } catch (error) {
             return {

diff --git a/libs/external/nats/src/lib/nats-interfaces/nats.interfaces.ts b/libs/external/nats/src/lib/nats-interfaces/nats.interfaces.ts
@@ -1,13 +1,17 @@
 import {
     ConsumeOptions as NatsConsumeOptions,
     ConsumerConfig,
+    JetStreamOptions,
     JetStreamPublishOptions,
     MsgHdrs,
+    NatsConnection,
     RequestOptions as NatsRequestOptions,
     StreamConfig,
     SubscriptionOptions as NatsSubscriptionOptions,
 } from 'nats'
 
+import { NatsConnectionConfig } from '../nats-configs/nats-module.config'
+
 export interface NatsResponse<T> {
     subject: string
     data: T
@@ -23,6 +27,17 @@ export interface RequestOptions
     headers: Record<string, string>
 }
 
+export interface NatsConnectionObject {
+    name: string
+    options: NatsConnectionConfig
+    connection: NatsConnection
+}
+
+export interface GetJetStreamManagerOptions {
+    connectionName?: string
+    options?: JetStreamOptions
+}
+
 export interface PublishOptions
     extends Omit<Partial<JetStreamPublishOptions>, 'headers' | 'timeout'> {
     timeout?: number
@@ -84,6 +99,7 @@ export interface ConsumerAcks {
 }
 
 export interface ConsumeOptions {
+    connectionName?: string
     stream: string
     consumer: Partial<ConsumerConfig>
     options?: Omit<NatsConsumeOptions, 'callback'>

diff --git a/libs/external/nats/src/lib/nats-services/nats-client.service.ts b/libs/external/nats/src/lib/nats-services/nats-client.service.ts
@@ -29,8 +29,10 @@ export class NatsClientService {
         subject: string,
         payload?: T,
         options?: RequestOptions,
+        connectionName?: string,
     ): Promise<NatsResponse<K>> {
-        const nc = this.natsConnection.getNatsConnection()
+        const nc =
+            this.natsConnection.getNatsConnection(connectionName).connection
 
         const encodedPayload: Payload = encodeMessage(payload)
 
@@ -103,8 +105,10 @@ export class NatsClientService {
     async reply(
         subject: string,
         options?: SubscriptionOptions,
+        connectionName?: string,
     ): Promise<Subscription> {
-        const nc = this.natsConnection.getNatsConnection()
+        const nc =
+            this.natsConnection.getNatsConnection(connectionName).connection
 
         const subscription = nc.subscribe(subject, options)
         this.logger.log(`Mapped {${subject}, NATS} route`)