move sparkplug decoding to backend
This commit is contained in:
@@ -1,15 +1,18 @@
|
||||
const { Base64 } = require('js-base64')
|
||||
|
||||
import { Base64 } from 'js-base64'
|
||||
import { Decoder } from './Decoder'
|
||||
|
||||
export class Base64Message {
|
||||
private base64Message: string
|
||||
private unicodeValue: string
|
||||
|
||||
public decoder: Decoder
|
||||
public length: number
|
||||
|
||||
private constructor(base64Str: string) {
|
||||
this.base64Message = base64Str
|
||||
this.unicodeValue = Base64.decode(base64Str)
|
||||
this.length = base64Str.length
|
||||
this.decoder = Decoder.NONE
|
||||
}
|
||||
|
||||
public static toUnicodeString(message: Base64Message) {
|
||||
@@ -27,15 +30,4 @@ export class Base64Message {
|
||||
public static toDataUri(message: Base64Message, mimeType: string) {
|
||||
return `data:${mimeType};base64,${message.base64Message}`
|
||||
}
|
||||
|
||||
public static toUint8Array(message: Base64Message) {
|
||||
const binaryString = window.atob(message.base64Message)
|
||||
const len = binaryString.length
|
||||
const bytes = new Uint8Array(len)
|
||||
for (const i of Array.from(Array(len).keys())) {
|
||||
bytes[i] = binaryString.charCodeAt(i)
|
||||
}
|
||||
|
||||
return bytes
|
||||
}
|
||||
}
|
||||
|
||||
4
backend/src/Model/Decoder.ts
Normal file
4
backend/src/Model/Decoder.ts
Normal file
@@ -0,0 +1,4 @@
|
||||
export enum Decoder {
|
||||
NONE,
|
||||
SPARKPLUG
|
||||
}
|
||||
@@ -1,10 +1,24 @@
|
||||
const protobuf = require('protobufjs')
|
||||
const sparkplugBProto = require('../../../res/sparkplug_b.proto')
|
||||
import { readFileSync } from 'fs'
|
||||
import * as protobuf from 'protobufjs'
|
||||
import { Base64Message } from './Base64Message';
|
||||
import { Decoder } from './Decoder';
|
||||
|
||||
export let Payload = undefined
|
||||
const buffer = readFileSync(require.resolve('../../../../res/sparkplug_b.proto'));
|
||||
const root = protobuf.parse(buffer.toString()).root
|
||||
export let SparkplugPayload = root.lookupType('com.cirruslink.sparkplug.protobuf.Payload')
|
||||
|
||||
export function loadSparkplugBPayload() {
|
||||
protobuf.load(sparkplugBProto).then((root: any) => {
|
||||
Payload = root.lookupType('com.cirruslink.sparkplug.protobuf.Payload')
|
||||
})
|
||||
export const SparkplugDecoder = {
|
||||
decode(input: Buffer): Base64Message | undefined {
|
||||
try {
|
||||
let message = Base64Message.fromString(
|
||||
JSON.stringify(
|
||||
SparkplugPayload.toObject(SparkplugPayload.decode(new Uint8Array(input)))
|
||||
)
|
||||
)
|
||||
message.decoder = Decoder.SPARKPLUG
|
||||
return message
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,11 +2,42 @@ import { TreeNodeFactory } from '../'
|
||||
import { Base64Message } from '../Base64Message'
|
||||
import { TreeNode } from '../TreeNode'
|
||||
import { MqttMessage } from '../../../../events'
|
||||
import { SparkplugPayload } from '../sparkplugb'
|
||||
|
||||
interface Decoder {
|
||||
decode(input: string): string | null
|
||||
}
|
||||
|
||||
const SparkplugDecoder = {
|
||||
decoderTime: 0,
|
||||
encoder: new TextEncoder(),
|
||||
decode(input: string): string | null {
|
||||
if (!SparkplugPayload) {
|
||||
return null
|
||||
}
|
||||
|
||||
const start = performance.now()
|
||||
|
||||
let result
|
||||
try {
|
||||
result = JSON.stringify(SparkplugPayload.toObject(SparkplugPayload.decode(this.encoder.encode(input))))
|
||||
} catch { }
|
||||
|
||||
this.decoderTime += performance.now() - start;
|
||||
return result ?? null
|
||||
}
|
||||
}
|
||||
|
||||
let i = 1
|
||||
setInterval(() => {
|
||||
console.log(`decoder time after ${i++ * 10} seconds: ${SparkplugDecoder.decoderTime}ms`)
|
||||
}, 10000)
|
||||
|
||||
export function makeTreeNode(topic: string, message?: string): TreeNode<any> {
|
||||
let sparkplugMessage = message && SparkplugDecoder.decode(message)
|
||||
const mqttMessage: MqttMessage = {
|
||||
topic,
|
||||
payload: message ? Base64Message.fromString(message) : null,
|
||||
payload: message ? Base64Message.fromString(sparkplugMessage ?? message) : null,
|
||||
qos: 0,
|
||||
retain: false,
|
||||
messageId: undefined,
|
||||
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
makePublishEvent,
|
||||
removeConnection,
|
||||
} from '../../events'
|
||||
import { SparkplugDecoder } from './Model/sparkplugb'
|
||||
|
||||
export class ConnectionManager {
|
||||
private connections: { [s: string]: DataSource<any> } = {}
|
||||
@@ -48,7 +49,7 @@ export class ConnectionManager {
|
||||
|
||||
backendEvents.emit(messageEvent, {
|
||||
topic,
|
||||
payload: Base64Message.fromBuffer(buffer),
|
||||
payload: SparkplugDecoder.decode(buffer) ?? Base64Message.fromBuffer(buffer),
|
||||
qos: packet.qos,
|
||||
retain: packet.retain,
|
||||
messageId: packet.messageId,
|
||||
|
||||
Reference in New Issue
Block a user