fix(cli): fix split option bugs & set default value for split option

cli/src/index.ts

@@ -501,7 +501,7 @@ export class Commander {
         'load the parameters from the local configuration file, which supports json and yaml format, default path is ./mqttx-cli-config.json',
       )
       .option('--file-read <PATH>', 'read the message body from the file', parseFileRead)
-      .option('--split <CHARACTER>', 'split the input message in a single file by a specified character.')
+      .option('--split [CHARACTER]', 'split the input message in a single file by a specified character, default is \n')
       .allowUnknownOption(false)
       .action(benchPub)
 

cli/src/lib/pub.ts

@@ -10,7 +10,7 @@ import delay from '../utils/delay'
 import { saveConfig, loadConfig } from '../utils/config'
 import { loadSimulator } from '../utils/simulate'
 import { serializeProtobufToBuffer } from '../utils/protobuf'
-import { readFile, processPath, fileDataSplitter, getPublishMessageFromFile } from '../utils/fileUtils'
+import { readFile, processPath, fileDataSplitter } from '../utils/fileUtils'
 import convertPayload from '../utils/convertPayload'
 import * as Debug from 'debug'
 import _ from 'lodash'
@@ -256,6 +256,8 @@ const multiPub = async (commandType: CommandType, options: BenchPublishOptions |
 
   let inFlightMessageCount = 0
 
+  const splitLimit = splitedMessageArr.length * count
+
   const isNewConnArray = Array(count).fill(true)
 
   const retryTimesArray = Array(count).fill(0)
@@ -288,8 +290,7 @@ const multiPub = async (commandType: CommandType, options: BenchPublishOptions |
 
   for (let i = 1; i <= count; i++) {
     // Duplicate splited messages array for each connection
-    const dupSplitedMessageArr: string[] =
-      split && fileRead && splitedMessageArr.length !== 0 ? _.cloneDeep(splitedMessageArr) : []
+    const dupSplitedMessageArr = splitedMessageArr.length !== 0 ? _.cloneDeep(splitedMessageArr) : []
 
     ;((i: number, connOpts: mqtt.IClientOptions) => {
       const opts = { ...connOpts }
@@ -318,8 +319,20 @@ const multiPub = async (commandType: CommandType, options: BenchPublishOptions |
               signale.success(`All ${total} messages have been sent, reaching the limit of ${limit}.`)
               process.exit(0)
             }
+            // If the segmented message has been completely sent, exit the process.
+            if (splitLimit > 0 && total >= splitLimit) {
+              // Wait for the total number of sent messages to be printed, then exit the process.
+              await delay(1000)
+              signale.success(`All ${total} messages from the ${fileRead} have been successfully sent.`)
+              process.exit(0)
+            }
             // If not initialized or client is not connected or message count exceeds the limit, do not send messages.
-            if (!initialized || !client.connected || (limit > 0 && total + inFlightMessageCount >= limit)) {
+            if (
+              !initialized ||
+              !client.connected ||
+              (limit > 0 && total + inFlightMessageCount >= limit) ||
+              (splitLimit > 0 && total + inFlightMessageCount >= splitLimit)
+            ) {
               return
             }
             inFlightMessageCount += 1
@@ -334,10 +347,14 @@ const multiPub = async (commandType: CommandType, options: BenchPublishOptions |
               publishMessage = simulationResult.message
             }
             if (fileRead) {
-              publishMessage = await getPublishMessageFromFile(split, dupSplitedMessageArr, fileData, {
-                total,
-                fileRead,
-              })
+              if (!split) {
+                publishMessage = fileData
+              } else {
+                if (dupSplitedMessageArr.length === 0) {
+                  return
+                }
+                publishMessage = Buffer.from(dupSplitedMessageArr.shift()!)
+              }
             }
             client.publish(publishTopic, publishMessage, pubOpts.opts, (err) => {
               inFlightMessageCount -= 1

cli/src/types/global.d.ts

@@ -3,7 +3,7 @@ declare global {
 
   type MQTTVersion = 3 | 4 | 5
 
-  type Protocol = 'mqtt' | 'mqtts'
+  type Protocol = 'mqtt' | 'mqtts' | 'ws' | 'wss'
 
   type QoS = 0 | 1 | 2
 
@@ -114,7 +114,7 @@ declare global {
     messageInterval: number
     limit: number
     verbose: boolean
-    split: string
+    split?: boolean | string
   }
 
   type OmitSubscribeOptions = Omit<

cli/src/utils/fileUtils.ts

@@ -92,31 +92,16 @@ const createNextNumberedFileName = (filePath: string): string => {
   }
 }
 
-const fileDataSplitter = (data: string | Buffer, split: string): string[] => {
+const fileDataSplitter = (data: string | Buffer, split: true | string): string[] => {
+  let defaultSplit = '\n'
+  if (split === true) {
+    split = defaultSplit
+  }
   const stringData = data.toString('utf-8')
   const splitRegex = new RegExp(split, 'g')
   return stringData.split(splitRegex)
 }
 
-const getPublishMessageFromFile = async (
-  split: string,
-  dupSplitedMessageArr: string[],
-  fileData: string | Buffer,
-  meta: { total: number; fileRead: string },
-): Promise<Buffer | string> => {
-  if (!split) {
-    return fileData
-  }
-  if (dupSplitedMessageArr.length === 0) {
-    await delay(1000)
-    signale.success(`All ${meta.total} messages from the ${meta.fileRead} have been successfully sent.`)
-    process.exit(0)
-  }
-
-  const unshiftedMessage = dupSplitedMessageArr.shift()
-  return Buffer.from(unshiftedMessage!)
-}
-
 export {
   processPath,
   getPathExtname,
@@ -129,5 +114,4 @@ export {
   appendFile,
   createNextNumberedFileName,
   fileDataSplitter,
-  getPublishMessageFromFile,
 }

cli/src/utils/parse.ts

@@ -244,10 +244,6 @@ const parseConnectOptions = (
     connectOptions.ca = fs.readFileSync(ca)
   }
 
-  if (key && cert && protocol !== 'mqtts') {
-    connectOptions.protocol = 'mqtts'
-  }
-
   if (insecure) {
     connectOptions.rejectUnauthorized = false
   }