Rename time Message and onInterval timestamp

Signed-off-by: Didier Wenzek <[email protected]>

Author: didier-wenzek

crates/extensions/tedge_flows/src/js_script.rs

@@ -319,7 +319,7 @@ export async function onMessage(message, config) {
     async fn using_date() {
         let js = r#"
 export function onMessage(message, config) {
-    let time = message.timestamp;
+    let time = message.time;
     return {
         "topic": message.topic,
         "payload": JSON.stringify({

crates/extensions/tedge_flows/src/js_value.rs

@@ -61,7 +61,7 @@ impl From<Message> for JsonValue {
             ("topic", JsonValue::string(value.topic)),
             ("payload", payload),
             ("raw_payload", raw_payload),
-            ("timestamp", JsonValue::option(value.timestamp)),
+            ("time", JsonValue::option(value.timestamp)),
         ])
     }
 }

docs/src/references/mappers/flows.md

@@ -64,22 +64,22 @@ interface FlowStep {
     onMessage(message: Message, config: object): null | Message | Message[],
 
     // called at regular intervals to produce aggregated messages
-    onInterval(timestamp: Timestamp, config: object): null | Message | Message[],
+    onInterval(time: Date, config: object): null | Message | Message[],
 
     // update the step config given a config update message
     onConfigUpdate(message: Message, config: object): object
 }
 ```
 
-A message contains the message topic and payload as well as an ingestion timestamp.
+A message contains the message topic and payload as well as a processing timestamp.
 The bytes of the raw message payload are also accessible as an array of unsigned bytes: 
 
 ```ts
 type Message = {
   topic: string,
   payload: string,
   raw_payload: Uint8Array,
-  timestamp: Date
+  time: Date
 }
 ```
 
@@ -88,7 +88,7 @@ These values are configured by the flow and can be dynamically updated on recept
 
 The `onMessage` function is called for each message to be transformed
   - The arguments passed to the function are:
-    - The message `{ topic: string, payload: string, raw_payload: Uint8Array, timestamp: Date }`
+    - The message `{ topic: string, payload: string, raw_payload: Uint8Array, time: Date }`
     - The config as read from the flow config or updated by the script
   - The function is expected to return zero, one or many transformed messages `[{ topic: string, payload: string }]`
   - An exception can be thrown if the input message cannot be transformed.

tests/RobotFramework/tests/tedge_flows/flows/add_timestamp.js

@@ -1,8 +1,8 @@
 export function onMessage (message) {
   let payload = JSON.parse(message.payload)
   if (!payload.time) {
-    let timestamp = message.timestamp
-    payload.time = timestamp.getTime() / 1000;
+    let time = message.time
+    payload.time = time.getTime() / 1000;
   }
 
   return [{

tests/RobotFramework/tests/tedge_flows/flows/circuit-breaker.js

@@ -14,7 +14,6 @@ class State {
     static batch = [0]
 }
 
-
 export function onMessage (message, config) {
     State.total += 1
     State.batch[0] += 1
@@ -45,8 +44,7 @@ export function onMessage (message, config) {
     }
 }
 
-
-export function onInterval(timestamp, config) {
+export function onInterval(time, config) {
     let max_batch_count = config.tick_count || 10
     let new_batch_count = State.batch.unshift(0)
     if (new_batch_count > max_batch_count) {
@@ -61,5 +59,4 @@ export function onInterval(timestamp, config) {
     } else {
         return []
     }
-
 }

tests/RobotFramework/tests/tedge_flows/flows/count-messages.js

@@ -11,7 +11,7 @@ export function onMessage (message) {
     return []
 }
 
-export function onInterval(timestamp, config) {
+export function onInterval(time, config) {
     let message = {
         topic: config.topic || "te/error",
         payload: JSON.stringify(State.count_per_topic)

tests/RobotFramework/tests/tedge_flows/flows/drop_stragglers.js

@@ -1,23 +1,22 @@
 // Reject any message that is too old, too new or with no timestamp
 export function onMessage (message, config) {
     let payload = JSON.parse(message.payload)
-    let msg_time = payload.time
-    if (!msg_time) {
+    if (!payload.time) {
         return []
     }
 
-    let msg_timestamp = msg_time
+    let msg_time = payload.time
     if (typeof(msg_time) === "string") {
-        msg_timestamp = Date.parse(msg_time) / 1e3
+        msg_time = Date.parse(msg_time) / 1e3
     }
 
-    let timestamp = message.timestamp
+    let time = message.time
     let max = time + (config.max_advance || 1);
     let min = time - (config.max_delay || 10);
 
-    if (min <= msg_timestamp && msg_timestamp <= max) {
+    if (min <= msg_time && msg_time <= max) {
         return [message]
     } else {
-        return [{"topic":" te/error", "payload":`straggler rejected on ${message.topic} with time=${msg_timestamp} at ${time}`}]
+        return [{"topic":" te/error", "payload":`straggler rejected on ${message.topic} with time=${msg_time} at ${time}`}]
     }
 }

tests/RobotFramework/tests/tedge_flows/flows/get-localtime.js

@@ -1,11 +1,11 @@
 export function onMessage (message) {
   let payload = JSON.parse(message.payload)
-  let timestamp = message.timestamp
-  let tzOffset = timestamp.getTimezoneOffset() * 60000;
+  let time = message.time
+  let tzOffset = time.getTimezoneOffset() * 60000;
 
-  payload.time = timestamp.toString();
-  payload.utc = timestamp.toISOString();
-  payload.local = (new Date(timestamp.getTime() - tzOffset)).toISOString().slice(0, -1);
+  payload.time = time.toString();
+  payload.utc = time.toISOString();
+  payload.local = (new Date(time.getTime() - tzOffset)).toISOString().slice(0, -1);
 
   return {
     topic: message.topic,