feat(cli):support input formatted messages to publish

emqx · Jun 28, 2023 · 3aa1758 · 3aa1758
1 parent 99e4ce8
commit 3aa1758
Show file tree

Hide file tree

Showing 6 changed files with 31 additions and 23 deletions.
diff --git a/cli/src/index.ts b/cli/src/index.ts
@@ -197,11 +197,7 @@ export class Commander {
       )
       .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')
-      .option(
-        '-Pft, --protobuf-format-type <TYPE>',
-        'the format type of message body, support base64, json, hex',
-        parseFormat,
-      )
+      .option('-f, --format <TYPE>', 'the input message format type, support base64, json, hex', parseFormat)
       .allowUnknownOption(false)
       .action(pub)
 

diff --git a/cli/src/lib/pub.ts b/cli/src/lib/pub.ts
@@ -18,17 +18,17 @@ const send = (
     message: string | Buffer
     protobufPath: string | undefined
     protobufMessageName: string | undefined
-    protobufFormatType: FormatType | undefined
+    format: FormatType | 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, protobufPath, protobufMessageName, protobufFormatType } = pubOpts
+    const { topic, message, protobufPath, protobufMessageName, format } = pubOpts
     basicLog.publishing()
-    let bufferMessage = serializeProtobufToBuffer(message, protobufPath, protobufMessageName, protobufFormatType)
+    let bufferMessage = serializeProtobufToBuffer(message, protobufPath, protobufMessageName, format)
     client.publish(topic, bufferMessage, pubOpts.opts, (err) => {
       if (err) {
         signale.warn(err)
@@ -59,7 +59,7 @@ const multisend = (
     message: string | Buffer
     protobufPath: string | undefined
     protobufMessageName: string | undefined
-    protobufFormatType: FormatType | undefined
+    format: FormatType | undefined
     opts: IClientPublishOptions
   },
   maximumReconnectTimes: number,
@@ -72,9 +72,9 @@ const multisend = (
     objectMode: true,
   })
   sender._write = (line, _enc, cb) => {
-    const { topic, opts, protobufPath, protobufMessageName, protobufFormatType } = pubOpts
+    const { topic, opts, protobufPath, protobufMessageName, format } = pubOpts
 
-    let bufferMessage = serializeProtobufToBuffer(line.trim(), protobufPath, protobufMessageName, protobufFormatType)
+    let bufferMessage = serializeProtobufToBuffer(line.trim(), protobufPath, protobufMessageName, format)
     client.publish(topic, bufferMessage, opts, cb)
   }
 

diff --git a/cli/src/types/global.d.ts b/cli/src/types/global.d.ts
@@ -74,7 +74,7 @@ declare global {
     connUserProperties?: Record<string, string | string[]>
     protobufPath?: string
     protobufMessageName?: string
-    protobufFormatType?: FormatType
+    format?: FormatType
   }
 
   interface SubscribeOptions extends ConnectOptions {
@@ -100,7 +100,7 @@ declare global {
 
   type OmitPublishOptions = Omit<
     PublishOptions,
-    'stdin' | 'multiline' | 'protobufPath' | 'protobufMessageName' | 'protobufFormatType'
+    'stdin' | 'multiline' | 'protobufPath' | 'protobufMessageName' | 'format'
   >
 
   interface BenchPublishOptions extends OmitPublishOptions {

diff --git a/cli/src/utils/parse.ts b/cli/src/utils/parse.ts
@@ -291,7 +291,7 @@ const parsePublishOptions = (options: PublishOptions) => {
     contentType,
     protobufPath,
     protobufMessageName,
-    protobufFormatType,
+    format,
   } = options
 
   const publishOptions: IClientPublishOptions = {
@@ -317,7 +317,7 @@ const parsePublishOptions = (options: PublishOptions) => {
     )
   }
 
-  return { topic, message, protobufPath, protobufMessageName, protobufFormatType, opts: publishOptions }
+  return { topic, message, protobufPath, protobufMessageName, format, opts: publishOptions }
 }
 
 const parseSubscribeOptions = (options: SubscribeOptions) => {

diff --git a/cli/src/utils/protobuf.ts b/cli/src/utils/protobuf.ts
@@ -5,11 +5,13 @@ import { transformPBJSError } from './protobufErrors'
 const convertObject = (raw: string | Buffer, format?: FormatType | undefined) => {
   switch (format) {
     case 'base64':
-      return JSON.parse(Buffer.from(raw.toString('utf-8'), 'base64').toString('utf-8'))
+      return Buffer.from(raw.toString('utf-8'), 'base64').toString('utf-8')
     case 'hex':
-      return JSON.parse(Buffer.from(raw.toString('utf-8').replaceAll(' ', ''), 'hex').toString('utf-8'))
+      return Buffer.from(raw.toString('utf-8').replaceAll(' ', ''), 'hex').toString('utf-8')
+    case 'json':
+      return JSON.stringify(JSON.parse(raw.toString('utf-8')), null, 2)
     default:
-      return JSON.parse(raw.toString('utf-8'))
+      return raw.toString('utf-8')
   }
 }
 
@@ -19,22 +21,29 @@ export const serializeProtobufToBuffer = (
   protobufMessageName: string | undefined,
   format?: FormatType | undefined,
 ): Buffer => {
-  let bufferMessage = Buffer.from(raw)
+  let rawData
+  try {
+    rawData = convertObject(raw, format)
+  } catch (error: unknown) {
+    signale.error(`Message format type error : ${(error as Error).message.split('\n')[0]}`)
+    process.exit(1)
+  }
+
+  let bufferMessage = Buffer.from(rawData)
   if (protobufPath && protobufMessageName) {
     try {
       const root = protobuf.loadSync(protobufPath)
       const Message = root.lookupType(protobufMessageName)
-      const rawData = convertObject(raw, format)
-      const err = Message.verify(rawData)
+      const err = Message.verify(JSON.parse(rawData))
       if (err) {
         signale.error(`Message serialization error: ${err}`)
         process.exit(1)
       }
-      const data = Message.create(rawData)
+      const data = Message.create(JSON.parse(rawData))
       const serializedMessage = Message.encode(data).finish()
       bufferMessage = Buffer.from(serializedMessage)
     } catch (error: unknown) {
-      signale.error(`Message format type error : ${(error as Error).message.split('\n')[0]}`)
+      signale.error(`Message serialization error: ${(error as Error).message.split('\n')[0]}`)
       process.exit(1)
     }
   }

diff --git a/cli/src/utils/protobufErrors.ts b/cli/src/utils/protobufErrors.ts
@@ -1,3 +1,6 @@
+/**
+ * @reference https://github.com/spluxx/Protoman/blob/master/src/core/protobuf/pbjsErrors.ts
+ */
 function prepend(msg: string): string {
   return `Message deserialization error: ${msg}`
 }