feat(cli): support auto reconnect

cli/src/index.ts

@@ -53,6 +53,12 @@ export class Commander {
       .option('--cert <PATH>', 'path to the cert file')
       .option('--ca <PATH>', 'path to the ca certificate')
       .option('--insecure', 'do not verify the server certificate')
+      .option(
+        '-rp, --reconnect-period <MILLISECONDS>',
+        'Interval between two reconnections, disable auto reconnect by setting to 0 (default: 1000ms)',
+        parseNumber,
+        1000,
+      )
       // properties options of MQTT 5.0
       .option('-se, --session-expiry-interval <SECONDS>', 'the session expiry interval in seconds', parseNumber)
       .option('--rcv-max, --receive-maximum <NUMBER>', 'the receive maximum value', parseNumber)
@@ -128,6 +134,12 @@ export class Commander {
       .option('--cert <PATH>', 'path to the cert file')
       .option('--ca <PATH>', 'path to the ca certificate')
       .option('--insecure', 'do not verify the server certificate')
+      .option(
+        '-rp, --reconnect-period <MILLISECONDS>',
+        'Interval between two reconnections, disable auto reconnect by setting to 0 (default: 1000ms)',
+        parseNumber,
+        1000,
+      )
       // connect properties options of MQTT 5.0
       .option('-se, --session-expiry-interval <SECONDS>', 'the session expiry interval in seconds', parseNumber)
       .option('--rcv-max, --receive-maximum <NUMBER>', 'the receive maximum value', parseNumber)
@@ -193,6 +205,12 @@ export class Commander {
       .option('--cert <PATH>', 'path to the cert file')
       .option('--ca <PATH>', 'path to the ca certificate')
       .option('--insecure', 'do not verify the server certificate')
+      .option(
+        '-rp, --reconnect-period <MILLISECONDS>',
+        'Interval between two reconnections, disable auto reconnect by setting to 0 (default: 1000ms)',
+        parseNumber,
+        1000,
+      )
       // connect properties options of MQTT 5.0
       .option('-se, --session-expiry-interval <SECONDS>', 'the session expiry interval in seconds', parseNumber)
       .option('--rcv-max, --receive-maximum <NUMBER>', 'the receive maximum value', parseNumber)

cli/src/lib/conn.ts

@@ -18,6 +18,14 @@ const conn = (options: ConnectOptions) => {
     signale.error(err)
     client.end()
   })
+
+  client.on('reconnect', () => {
+    signale.await('Reconnecting...')
+  })
+
+  client.on('close', () => {
+    signale.error('Connection closed')
+  })
 }
 
 const benchConn = async (options: BenchConnectOptions) => {

cli/src/lib/pub.ts

@@ -37,6 +37,7 @@ const multisend = (
   connOpts: IClientOptions,
   pubOpts: { topic: string; message: string | Buffer; opts: IClientPublishOptions },
 ) => {
+  let isNewConnection = true
   const client = mqtt.connect(connOpts)
   signale.await('Connecting...')
   const sender = new Writable({
@@ -49,12 +50,30 @@ const multisend = (
 
   client.on('connect', () => {
     signale.success('Connected, press Enter to publish, press Ctrl+C to exit')
-    pump(process.stdin, split2(), sender, (err) => {
-      client.end()
-      if (err) {
-        throw err
-      }
-    })
+    isNewConnection &&
+      pump(process.stdin, split2(), sender, (err) => {
+        client.end()
+        if (err) {
+          throw err
+        }
+      })
+  })
+
+  client.on('error', (err) => {
+    signale.error(err)
+    client.end()
+  })
+
+  client.on('reconnect', () => {
+    signale.await('Reconnecting...')
+    isNewConnection = false
+    sender.uncork()
+  })
+
+  client.on('close', () => {
+    signale.error('Connection closed')
+    const { reconnectPeriod } = connOpts
+    reconnectPeriod ? sender.cork() : process.exit(1)
   })
 }
 

cli/src/lib/sub.ts

@@ -59,6 +59,14 @@ const sub = (options: SubscribeOptions) => {
     signale.error(err)
     client.end()
   })
+
+  client.on('reconnect', () => {
+    signale.await('Reconnecting...')
+  })
+
+  client.on('close', () => {
+    signale.error('Connection closed')
+  })
 }
 
 const benchSub = async (options: BenchSubscribeOptions) => {

cli/src/types/global.d.ts

@@ -21,6 +21,7 @@ declare global {
     cert?: string
     ca?: string
     insecure?: boolean
+    reconnectPeriod: number
     // properties of MQTT 5.0
     sessionExpiryInterval?: number
     receiveMaximum?: number

cli/src/utils/parse.ts

@@ -90,6 +90,7 @@ const parseConnectOptions = (
     cert,
     ca,
     insecure,
+    reconnectPeriod,
     sessionExpiryInterval,
     receiveMaximum,
     maximumPacketSize,
@@ -119,6 +120,7 @@ const parseConnectOptions = (
     username,
     password,
     protocol,
+    reconnectPeriod,
   }
 
   if (key) {