feat(cli):support protobuf format

cli/package.json

@@ -26,6 +26,7 @@
     "concat-stream": "^2.0.0",
     "core-js": "^3.26.0",
     "mqtt": "^4.3.7",
+    "protobufjs": "^7.2.3",
     "pump": "^3.0.0",
     "readable-stream": "^3.6.0",
     "signale": "^1.4.0",

cli/src/index.ts

@@ -192,6 +192,8 @@ export class Commander {
         '--config [PATH]',
         'load the parameters from the local configuration file, which supports json and yaml format, default path is ./mqttx-cli-config.json',
       )
+      .option('-Pp, --protobuf-path <PATH>', 'the .proto file that defines the message format of protobuf')
+      .option('-Pmn, --protobuf-message-name <NAME>', 'the name of the protobuf message type')
       .action(pub)
 
     this.program
@@ -280,6 +282,8 @@ export class Commander {
         '--config [PATH]',
         'load the parameters from the local configuration file, which supports json and yaml format, default path is ./mqttx-cli-config.json',
       )
+      .option('-Pp, --protobuf-path <PATH>', 'the .proto file that defines the message format of protobuf')
+      .option('-Pmn, --protobuf-message-name <NAME>', 'the name of the protobuf message type')
       .action(sub)
 
     const benchCmd = this.program.command('bench').description('MQTT Benchmark in performance testing.')

cli/src/lib/pub.ts

@@ -9,19 +9,27 @@ import { parseConnectOptions, parsePublishOptions, checkTopicExists, checkScenar
 import delay from '../utils/delay'
 import { saveConfig, loadConfig } from '../utils/config'
 import { loadSimulator } from '../utils/simulate'
-
+import { serializeProtobufToBuffer } from '../utils/protobuf'
 const send = (
   config: boolean | string | undefined,
   connOpts: IClientOptions,
-  pubOpts: { topic: string; message: string | Buffer; opts: IClientPublishOptions },
+  pubOpts: {
+    topic: string
+    message: string | Buffer
+    protobufPath: string | undefined
+    protobufMessageName: string | undefined
+    opts: IClientPublishOptions
+  },
 ) => {
   const client = mqtt.connect(connOpts)
   basicLog.connecting(config, connOpts.hostname!, connOpts.port, pubOpts.topic, pubOpts.message.toString())
   client.on('connect', () => {
     basicLog.connected()
-    const { topic, message } = pubOpts
+    const { topic, message, protobufPath, protobufMessageName } = pubOpts
     basicLog.publishing()
-    client.publish(topic, message, pubOpts.opts, (err) => {
+
+    let bufferMessage = serializeProtobufToBuffer(message, protobufPath, protobufMessageName)
+    client.publish(topic, bufferMessage, pubOpts.opts, (err) => {
       if (err) {
         signale.warn(err)
       } else {
@@ -46,7 +54,13 @@ const send = (
 const multisend = (
   config: boolean | string | undefined,
   connOpts: IClientOptions,
-  pubOpts: { topic: string; message: string | Buffer; opts: IClientPublishOptions },
+  pubOpts: {
+    topic: string
+    message: string | Buffer
+    protobufPath: string | undefined
+    protobufMessageName: string | undefined
+    opts: IClientPublishOptions
+  },
   maximumReconnectTimes: number,
 ) => {
   let isNewConnection = true
@@ -57,8 +71,10 @@ const multisend = (
     objectMode: true,
   })
   sender._write = (line, _enc, cb) => {
-    const { topic, opts } = pubOpts
-    client.publish(topic, line.trim(), opts, cb)
+    const { topic, opts, protobufPath, protobufMessageName } = pubOpts
+
+    let bufferMessage = serializeProtobufToBuffer(line.trim(), protobufPath, protobufMessageName)
+    client.publish(topic, bufferMessage, opts, cb)
   }
 
   client.on('connect', () => {

cli/src/lib/sub.ts

@@ -4,6 +4,7 @@ import { parseConnectOptions, parseSubscribeOptions, checkTopicExists } from '..
 import delay from '../utils/delay'
 import convertPayload from '../utils/convertPayload'
 import { saveConfig, loadConfig } from '../utils/config'
+import { deserializeBufferToProtobuf } from '../utils/protobuf'
 
 const sub = (options: SubscribeOptions) => {
   const { save, config } = options
@@ -59,13 +60,14 @@ const sub = (options: SubscribeOptions) => {
   })
 
   client.on('message', (topic, payload, packet) => {
-    const { format } = options
+    const { format, protobufPath, protobufMessageName } = options
 
     const msgData: Record<string, unknown>[] = []
 
     options.verbose && msgData.push({ label: 'topic', value: topic })
 
-    msgData.push({ label: 'payload', value: convertPayload(payload, format) })
+    let payloadMessage = deserializeBufferToProtobuf(payload, protobufPath, protobufMessageName)
+    msgData.push({ label: 'payload', value: payloadMessage ? payloadMessage : convertPayload(payload, format) })
 
     packet.retain && msgData.push({ label: 'retain', value: packet.retain })
 

cli/src/types/global.d.ts

@@ -72,6 +72,8 @@ declare global {
     subscriptionIdentifier?: number
     contentType?: string
     connUserProperties?: Record<string, string | string[]>
+    protobufPath?: string
+    protobufMessageName?: string
   }
 
   interface SubscribeOptions extends ConnectOptions {
@@ -86,14 +88,16 @@ declare global {
     outputMode?: OutputMode
     verbose: boolean
     connUserProperties?: Record<string, string | string[]>
+    protobufPath?: string
+    protobufMessageName?: string
   }
 
   interface BenchConnectOptions extends ConnectOptions {
     count: number
     interval: number
   }
 
-  type OmitPublishOptions = Omit<PublishOptions, 'stdin' | 'multiline'>
+  type OmitPublishOptions = Omit<PublishOptions, 'stdin' | 'multiline' | 'protobufPath' | 'protobufMessageName'>
 
   interface BenchPublishOptions extends OmitPublishOptions {
     count: number
@@ -102,7 +106,7 @@ declare global {
     verbose: boolean
   }
 
-  type OmitSubscribeOptions = Omit<SubscribeOptions, 'format' | 'outputMode'>
+  type OmitSubscribeOptions = Omit<SubscribeOptions, 'format' | 'outputMode' | 'protobufPath' | 'protobufMessageName'>
 
   interface BenchSubscribeOptions extends OmitSubscribeOptions {
     count: number

cli/src/utils/parse.ts

@@ -289,6 +289,8 @@ const parsePublishOptions = (options: PublishOptions) => {
     userProperties,
     subscriptionIdentifier,
     contentType,
+    protobufPath,
+    protobufMessageName,
   } = options
 
   const publishOptions: IClientPublishOptions = {
@@ -314,7 +316,7 @@ const parsePublishOptions = (options: PublishOptions) => {
     )
   }
 
-  return { topic, message, opts: publishOptions }
+  return { topic, message, protobufPath, protobufMessageName, opts: publishOptions }
 }
 
 const parseSubscribeOptions = (options: SubscribeOptions) => {

cli/src/utils/protobuf.ts

@@ -0,0 +1,29 @@
+import protobuf from 'protobufjs'
+
+export const serializeProtobufToBuffer = (
+  raw: string | Buffer,
+  protobufPath: string | undefined,
+  protobufMessageName: string | undefined,
+): Buffer => {
+  let bufferMessage = Buffer.from(raw)
+  if (protobufPath && protobufMessageName) {
+    const root = protobuf.loadSync(protobufPath)
+    const MyMessage = root.lookupType(protobufMessageName)
+    const data = MyMessage.create(JSON.parse(raw.toString()))
+    const serializedMessage = MyMessage.encode(data).finish()
+    bufferMessage = Buffer.from(serializedMessage)
+  }
+  return bufferMessage
+}
+
+export const deserializeBufferToProtobuf = (
+  payload: Buffer,
+  protobufPath: string | undefined,
+  protobufMessageName: string | undefined,
+): any => {
+  if (protobufPath && protobufMessageName) {
+    const root = protobuf.loadSync(protobufPath)
+    const MyMessage = root.lookupType(protobufMessageName)
+    return MyMessage.decode(payload)
+  }
+}