Merge pull request #35 from didier-wenzek/feat-protobuf-es

Improve protobuf example to read protobuf encoded msg

Author: reubenmiller

flows/protobuf-xform/README.md

@@ -31,3 +31,22 @@ Topic: **te/device/main///m/location**
   "longitude": 153.02223634041275
 }
 ```
+
+### Decoding Protobuf messages
+
+The encoded measurements produced by the protobuf example flow ...
+
+```
+$ tedge flows test --base64-output te/device/main///m/environment '{ "temperature": 29, "humidity": 50 }'
+
+[c8y-mqtt/proto/sensor_data] ChIJAAAAAAAAPUARAAAAAAAASUA=
+```
+
+... can be decoded back by the same protobuf example flow:
+
+```
+$ tedge flows test --base64-input c8y-mqtt/proto/setpoint ChIJAAAAAAAAPUARAAAAAAAASUA=                                           
+
+[te/device/main///sig/setpoint] {"$typeName":"sensorpackage.SensorMessage","sensor":{"case":"environment","value":{"$typeName":"sensorpackage.EnvironmentSensor","metaInfo":{},"temperature":29,"humidity":50}}}
+
+```

flows/protobuf-xform/flow.toml

@@ -2,9 +2,12 @@
 mqtt.topics = [
     "te/device/main/+/+/m/environment",
     "te/device/main/+/+/m/location",
+    "c8y-mqtt/proto/setpoint",
 ]
 
 [[steps]]
 script = "dist/main.mjs"
 # config.topic: Use the template variable, '{{type}}' if you wish to include the sensor data type within the topic
 config.topic = "c8y-mqtt/proto/sensor_data"
+config.cmdtopic = "te/device/main///sig/setpoint"
+config.base64 = false

flows/protobuf-xform/src/main.ts

@@ -3,8 +3,8 @@
 */
 // import * as proto from 'protobufjs';
 import { Message } from "../../common/tedge";
-import { create, toBinary } from "@bufbuild/protobuf";
-import { base64Encode } from "@bufbuild/protobuf/wire";
+import { create, toBinary, fromBinary } from "@bufbuild/protobuf";
+import { base64Decode, base64Encode } from "@bufbuild/protobuf/wire";
 import {
   SensorMessageSchema,
   EnvironmentSensorSchema,
@@ -15,16 +15,36 @@ export interface Config {
   topic: string;
 }
 
-export function onMessage(
+function onSetpoint(
   message: Message,
-  { topic = "out/proto/sensor" },
+  { topic = "out/proto/actuator", base64 = false },
 ): Message[] {
-  const payload = JSON.parse(message.payload);
+  let binPayload;
+  if (base64) {
+    binPayload = base64Decode(message.payload);
+  } else {
+    binPayload = message.raw_payload
+  }
 
+  let setPoint = fromBinary(SensorMessageSchema, binPayload)
+  return [
+    {
+      timestamp: message.timestamp,
+      topic: topic,
+      payload: JSON.stringify(setPoint),
+    },
+  ];
+}
+
+export function onMessage(
+  message: Message,
+  { topic = "out/proto/sensor", cmdtopic = "out/proto/actuator", base64 = false },
+): Message[] {
   const payloadType = message.topic.split("/").slice(-1)[0];
 
   let data;
   if (payloadType == "environment") {
+    const payload = JSON.parse(message.payload);
     data = {
       case: "environment",
       value: create(EnvironmentSensorSchema, {
@@ -34,6 +54,7 @@ export function onMessage(
       }),
     };
   } else if (payloadType == "location") {
+    const payload = JSON.parse(message.payload);
     data = {
       case: "location",
       value: create(LocationSensorSchema, {
@@ -43,7 +64,10 @@ export function onMessage(
         },
       }),
     };
+  } else if (payloadType == "setpoint") {
+    return onSetpoint(message, {topic: cmdtopic, base64: base64});
   }
+
   if (!data) {
     return [];
   }
@@ -54,11 +78,16 @@ export function onMessage(
 
   const outputTopic = topic.replaceAll("{{type}}", payloadType);
 
+  let binPayload = toBinary(SensorMessageSchema, sensor);
+  if (base64) {
+    binPayload = base64Encode(binPayload);
+  }
+
   return [
     {
       timestamp: message.timestamp,
       topic: outputTopic,
-      payload: base64Encode(toBinary(SensorMessageSchema, sensor)),
+      payload: binPayload,
     },
   ];
 }