feat: adds try/catch block to increase resilience

src/adapters/cluster/redis/index.ts

@@ -14,14 +14,25 @@ class RedisClusterAdapter extends ClusterAdapter {
   }
 
   async connect(): Promise<this> {
-    return this._connect()
+    try{
+      return this._connect()
+    } catch (error) { 
+      console.error('Error connecting to Redis:', error)
+      throw error
+    }
   }
 
   async send(message: GleeMessage): Promise<void> {
+    try{
     return this._send(message)
+    } catch (error) {
+      console.error('Error sending message to Redis:', error)
+      throw error
+    }
   }
 
   async _connect(): Promise<this> {
+    try {
     this._channelName = `${this.serverName}-channel`
 
     this._publisher = createClient({
@@ -62,6 +73,10 @@ class RedisClusterAdapter extends ClusterAdapter {
 
     this.emit('connect', { name: this.name(), adapter: this })
     return this
+    } catch (error) {
+      console.error('Error connecting to Redis:', error)
+      throw error
+    }
   }
 
   async _send(message: GleeMessage): Promise<void> {

src/adapters/http/client.ts

@@ -15,13 +15,19 @@ 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
+    try {
+      this.emit('connect', {
+        name: this.name(),
+        adapter: this,
+        connection: http,
+        channel: this.channelNames,
+      })
+      return this
+    } catch (err) {
+      logWarningMessage(`Failed to Connect: An error occurred while connecting to '${this.name()}' on the '${this.channelNames}' channel. Please review the error details below for further information and corrective action.`)
+      this.emit('error', err)
+      throw err
+    }
   }
 
   async send(message: GleeMessage): Promise<void> {

src/adapters/http/server.ts

@@ -17,11 +17,21 @@ class HttpAdapter extends Adapter {
   }
 
   async connect(): Promise<this> {
-    return this._connect()
+    try {
+      return this._connect()
+    } catch (error) {
+      console.error('Error connecting to HTTP:', error)
+      throw error
+    }
   }
 
   async send(message: GleeMessage): Promise<void> {
-    return this._send(message)
+    try{
+      return this._send(message)
+    } catch (error) {
+      console.error('Error sending message to HTTP:', error)
+      throw error
+    }
   }
 
   async _readRequestBody(req: http.IncomingMessage): Promise<string> {
@@ -164,6 +174,7 @@ class HttpAdapter extends Adapter {
   }
 
   async _connect(): Promise<this> {
+    try {
     const config = await this.resolveProtocolConfig('http')
     const httpOptions = config?.server
     const httpServer = httpOptions?.httpServer || http.createServer()
@@ -173,6 +184,10 @@ class HttpAdapter extends Adapter {
     httpServer.listen(port)
     this.emit('server:ready', { name: this.name(), adapter: this })
     return this
+    } catch (error) {
+      console.error(error)
+      throw error
+    }
   }
   _getOperationBindings(channel: ChannelInterface) {
     return channel.operations().filterByReceive().map(operation => operation.bindings().get("http")?.json())

src/adapters/kafka/index.ts

@@ -10,26 +10,27 @@ 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() {
+    try {
+      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({
@@ -74,6 +75,9 @@ class KafkaAdapter extends Adapter {
         this.emit('message', msg, consumer)
       },
     })
+    } catch (error) {
+      console.error('Error connecting to Kafka:', error)
+    }
   }
 
   async send(message: GleeMessage) {

src/adapters/mqtt/index.ts

@@ -37,11 +37,21 @@ class MqttAdapter extends Adapter {
   }
 
   async connect(): Promise<this> {
-    return this._connect()
+    try{
+      return this._connect()
+    } catch(error){
+      console.error('Error connecting to MQTT:', error)
+      throw error
+    }
   }
 
   async send(message: GleeMessage) {
-    return this._send(message)
+    try{
+      return this._send(message)
+    } catch(error){
+      console.error('Error sending message to MQTT:', error)
+      throw error
+    }
   }
 
   private getSecurityReqs() {
@@ -186,6 +196,7 @@ class MqttAdapter extends Adapter {
   }
 
   async _connect(): Promise<this> {
+    try {
     const mqttOptions: MqttAdapterConfig = await this.resolveProtocolConfig(
       'mqtt'
     )
@@ -238,6 +249,10 @@ class MqttAdapter extends Adapter {
     }
 
     return connectClient()
+    } catch (error) {
+      console.error(error)
+      throw error
+    }
   }
 
   _send(message: GleeMessage): Promise<void> {

src/adapters/socket.io/index.ts

@@ -10,14 +10,25 @@ class SocketIOAdapter extends Adapter {
   }
 
   async connect(): Promise<this> {
-    return this._connect()
+    try {
+      return this._connect()
+    } catch(error){
+      console.error('Error connecting to Socket.IO:', error)
+      throw error
+    }
   }
 
   async send(message: GleeMessage): Promise<void> {
-    return this._send(message)
+    try{
+      return this._send(message)
+    } catch(error){
+      console.error('Error sending message to Socket.IO:', error)
+      throw error
+    }
   }
 
   async _connect(): Promise<this> {
+    try {
     const config = await this.resolveProtocolConfig('ws')
     const websocketOptions = config?.server
     const serverUrl: URL = new URL(this.serverUrlExpanded)
@@ -73,6 +84,10 @@ class SocketIOAdapter extends Adapter {
       this.server.listen(port)
     }
     return this
+    } catch (error) {
+      console.error('An error occurred while connecting:', error)
+      throw error
+    }
   }
 
   async _send(message: GleeMessage): Promise<void> {

src/adapters/ws/client.ts

@@ -21,14 +21,25 @@ class WsClientAdapter extends Adapter {
   }
 
   async connect(): Promise<this> {
-    return this._connect()
+    try{
+      return this._connect()
+    } catch(error){
+      console.error('Error connecting to WS:', error)
+      throw error
+    }
   }
 
   async send(message: GleeMessage) {
-    return this._send(message)
+    try{
+      return this._send(message)
+    } catch(error){
+      console.error('Error sending message to WS:', error)
+      throw error
+    }
   }
 
   private async _connect(): Promise<this> {
+    try{
     const channelsOnThisServer = this.getWsChannels()
 
     debug("connecting to ", this.serverName)
@@ -81,6 +92,10 @@ class WsClientAdapter extends Adapter {
       })
     }
     return this
+    }catch (error) {
+      console.error('An error occurred while connecting:', error)
+      throw error
+    }
   }
 
   private getWsChannels() {