feat: adds try/catch block to increase resilience (#666)

asyncapi · Mar 18, 2024 · 485068c · 485068c
1 parent 5c5d4f4
commit 485068c
Show file tree

Hide file tree

Showing 4 changed files with 41 additions and 30 deletions.
diff --git a/src/adapters/http/client.ts b/src/adapters/http/client.ts
@@ -15,13 +15,13 @@ class HttpClientAdapter extends Adapter {
     return 'HTTP client'
   }
   async connect(): Promise<this> {
-    this.emit('connect', {
-      name: this.name(),
-      adapter: this,
-      connection: http,
-      channel: this.channelNames,
-    })
-    return this
+      this.emit('connect', {
+        name: this.name(),
+        adapter: this,
+        connection: http,
+        channel: this.channelNames,
+      })
+      return this
   }
 
   async send(message: GleeMessage): Promise<void> {

diff --git a/src/adapters/kafka/index.ts b/src/adapters/kafka/index.ts
@@ -10,26 +10,30 @@ class KafkaAdapter extends Adapter {
     return 'Kafka adapter'
   }
 
-  async connect() {
-    const kafkaOptions: KafkaAdapterConfig = await this.resolveProtocolConfig(
-      'kafka'
-    )
-    const auth: KafkaAuthConfig = await this.getAuthConfig(kafkaOptions?.auth)
-    const securityRequirements = this.AsyncAPIServer.security().map(
-      (sec) => {
-        const secName = Object.keys(sec.values())[0]
-        return this.parsedAsyncAPI.components().securitySchemes().get(secName)
-      }
-    )
-    const userAndPasswordSecurityReq = securityRequirements.find(
-      (sec) => sec.type() === 'userPassword'
-    )
-    const scramSha256SecurityReq = securityRequirements.find(
-      (sec) => sec.type() === 'scramSha256'
-    )
-    const scramSha512SecurityReq = securityRequirements.find(
-      (sec) => sec.type() === 'scramSha512'
-    )
+  async connect(): Promise<void> {
+    await this._connect()
+  }
+
+  async _connect() {
+      const kafkaOptions: KafkaAdapterConfig = await this.resolveProtocolConfig(
+        'kafka'
+      )
+      const auth: KafkaAuthConfig = await this.getAuthConfig(kafkaOptions?.auth)
+      const securityRequirements = this.AsyncAPIServer.security().map(
+        (sec) => {
+          const secName = Object.keys(sec.values())[0]
+          return this.parsedAsyncAPI.components().securitySchemes().get(secName)
+        }
+      )
+      const userAndPasswordSecurityReq = securityRequirements.find(
+        (sec) => sec.type() === 'userPassword'
+      )
+      const scramSha256SecurityReq = securityRequirements.find(
+        (sec) => sec.type() === 'scramSha256'
+      )
+      const scramSha512SecurityReq = securityRequirements.find(
+        (sec) => sec.type() === 'scramSha512'
+      )
 
     const brokerUrl = new URL(this.AsyncAPIServer.url())
     this.kafka = new Kafka({

diff --git a/src/adapters/mqtt/index.ts b/src/adapters/mqtt/index.ts
@@ -244,6 +244,7 @@ class MqttAdapter extends Adapter {
     }
 
     return connectClient()
+
   }
 
   _send(message: GleeMessage): Promise<void> {

diff --git a/src/lib/glee.ts b/src/lib/glee.ts
@@ -155,9 +155,15 @@ export default class Glee extends EventEmitter {
       promises.push(a.instance.connect())
     })
 
-    if (this._clusterAdapter) {
-      this._clusterAdapter.instance = new this._clusterAdapter.Adapter(this)
-      promises.push(this._clusterAdapter.instance.connect())
+    try {
+      if (this._clusterAdapter) {
+        this._clusterAdapter.instance = new this._clusterAdapter.Adapter(this)
+        promises.push(this._clusterAdapter.instance.connect().catch((error) => {
+          console.error('Error connecting to cluster:', error)
+        }))
+      }
+    } catch (error) {
+      console.error('Error connecting:', error)
     }
 
     return Promise.all(promises)