feat: better qos support

src/adapters/mqtt/index.ts

@@ -11,6 +11,9 @@ interface IMQTTHeaders {
   length: number,
 }
 
+const MQTT_UNSPECIFIED_ERROR_REASON = 0x80
+const MQTT_SUCCESS_REASON = 0
+
 class MqttAdapter extends Adapter {
   private client: MqttClient
   private firstConnect: boolean
@@ -30,7 +33,8 @@ class MqttAdapter extends Adapter {
   _connect(): Promise<this> {
     return new Promise((resolve) => {
       const subscribedChannels = this.getSubscribedChannels()
-      const serverBinding = this.AsyncAPIServer.binding('mqtt')
+      const mqttServerBinding = this.AsyncAPIServer.binding('mqtt')
+      const mqtt5ServerBinding = this.AsyncAPIServer.binding('mqtt5')
       const securityRequirements = (this.AsyncAPIServer.security() || []).map(sec => {
         const secName = Object.keys(sec.json())[0]
         return this.parsedAsyncAPI.components().securityScheme(secName)
@@ -42,12 +46,18 @@ class MqttAdapter extends Adapter {
       const certsConfig = process.env.GLEE_SERVER_CERTS?.split(',').map(t => t.split(':'))
       const certs = certsConfig?.filter(tuple => tuple[0] === this.serverName)?.map(t => fs.readFileSync(t[1])) // eslint-disable-line security/detect-non-literal-fs-filename
 
+      const serverBinding = mqttServerBinding
+      const protocolVersion = mqtt5ServerBinding ? 5 : 4
+
       this.client = mqtt.connect({
         host: url.host,
         port: url.port || (url.protocol === 'mqtt:' ? 1883 : 8883),
         protocol: url.protocol.substr(0, url.protocol.length - 1),
         clientId: serverBinding && serverBinding.clientId,
         clean: serverBinding && serverBinding.cleanSession,
+        properties: {
+          sessionExpiryInterval: serverBinding && serverBinding.sessionExpiryInterval
+        },
         will: serverBinding && serverBinding.will && {
           topic: serverBinding && serverBinding.lastWill && serverBinding.lastWill.topic ? serverBinding.lastWill.topic : undefined,
           qos: serverBinding && serverBinding.lastWill && serverBinding.lastWill.qos ? serverBinding.lastWill.qos : undefined,
@@ -58,15 +68,19 @@ class MqttAdapter extends Adapter {
         username: userAndPasswordSecurityReq ? process.env.GLEE_USERNAME : undefined,
         password: userAndPasswordSecurityReq ? process.env.GLEE_PASSWORD : undefined,
         ca: X509SecurityReq ? certs : undefined,
+        protocolVersion,
+        customHandleAcks: this._customAckHandler.bind(this),
       })
 
-      this.client.on('connect', () => {
+      this.client.on('connect', connAckPacket => {
+        const isSessionResume = connAckPacket.sessionPresent
+
         if (!this.firstConnect) {
           this.firstConnect = true
           this.emit('connect', { name: this.name(), adapter: this, connection: this.client, channels: this.channelNames })
         }
 
-        if (Array.isArray(subscribedChannels)) {
+        if (!isSessionResume && Array.isArray(subscribedChannels)) {
           subscribedChannels.forEach((channel) => {
             const operation = this.parsedAsyncAPI.channel(channel).publish()
             const binding = operation.binding('mqtt')
@@ -80,6 +94,9 @@ class MqttAdapter extends Adapter {
       })
 
       this.client.on('message', (channel, message, mqttPacket) => {
+        const qos = mqttPacket.qos
+        if ( protocolVersion === 5 && qos > 0 ) return   // ignore higher qos messages. already processed
+
         const msg = this._createMessage(mqttPacket as IPublishPacket)
         this.emit('message', msg, this.client)
       })
@@ -137,6 +154,15 @@ class MqttAdapter extends Adapter {
       channel: packet.topic,
     })
   }
+
+  _customAckHandler(channel, message, mqttPacket, done) {
+    const msg = this._createMessage(mqttPacket as IPublishPacket)
+
+    msg.on('successfullyProcessed', () => done(MQTT_SUCCESS_REASON))
+    msg.on('failedProcessing', () => done(MQTT_UNSPECIFIED_ERROR_REASON))
+
+    this.emit('message', msg, this.client)
+  }
 }
 
 export default MqttAdapter

src/lib/glee.ts

@@ -223,6 +223,8 @@ export default class Glee extends EventEmitter {
 
     async.seq(...mws)(message, (err: Error, msg: GleeMessage) => {
       if (err) {
+        message.notifyFailedProcessing()
+        debug('Error encountered while processing middlewares.')
         this._processError(errorMiddlewares, err, msg)
         return
       }
@@ -238,6 +240,7 @@ export default class Glee extends EventEmitter {
           }
         })
       } else {
+        message.notifySuccessfulProcessing()
         debug('Inbound pipeline finished.')
       }
     })

src/lib/message.ts

@@ -189,6 +189,20 @@ class GleeMessage extends EventEmitter {
   send() {
     this.emit('send', this)
   }
+
+  /**
+   * Indicates successfully processed the message
+   */
+  notifySuccessfulProcessing() {
+    this.emit('successfullyProcessed')
+  }
+
+  /**
+   * Indicates failure in processing the message
+   */
+  notifyFailedProcessing() {
+    this.emit('failedProcessing')
+  }
 }
 
 export default GleeMessage