Update the operator source code (#162)

The Operators source code has been updated to the latest.

## Does this introduce a breaking change?
<!-- Mark one with an "x". -->
```
[ ] Yes
[ x] No
```

## Pull Request Type
What kind of change does this Pull Request introduce?
Changes the operator source in compliant with aziot-operations

<!-- Please check the one that applies to this PR using "x". -->
```
[ ] Bugfix
[ ] Feature
[ x] Code style update (formatting, local variables)
[ ] Refactoring (no functional changes, no api changes)
[ ] Documentation content changes
[ ] Other... Please describe:
```

## How to Test
*  Get the code

```
git clone [repo-address]
cd [repo-name]
git checkout [branch-name]
npm install
```

* Test the code
<!-- Add steps to run the tests suite and/or manually test -->
```
```

## What to Check
Verify that the following are valid
* ...

## Other Information
<!-- Add any other helpful information that may be needed here. -->

Author: ggjjj

samples/wasm/rust/examples/collection/Cargo.toml

@@ -7,7 +7,7 @@ edition = "2021"
 
 [dependencies]
 wit-bindgen = "0.22"
-wasm_graph_sdk = { version = "=1.1.1", registry = "aio-sdks" }
+wasm_graph_sdk = { version = "=1.1.3", registry="aio-wg" }
 
 serde = { version = "1", features = [
 	"derive",
@@ -19,4 +19,3 @@ serde_json = { version = "1.0", default-features = false, features = [
 
 [lib]
 crate-type = ["cdylib"]
-path = "src/lib.rs"

samples/wasm/rust/examples/collection/src/lib.rs

@@ -9,7 +9,10 @@ use wasm_graph_sdk::macros::accumulate_operator;
 use wasm_graph_sdk::metrics::{self, CounterValue, Label};
 
 #[accumulate_operator]
-fn accumulate_sensor_data(staged: DataModel, inputs: Vec<DataModel>) -> Result<DataModel, Error> {
+fn accumulate_sensor_data(
+    staged: DataModel,
+    inputs: Vec<DataModel>,
+) -> Result<DataModel, Error> {
     let labels = vec![Label {
         key: "module".to_owned(),
         value: "module-collection/accumulate".to_owned(),
@@ -23,26 +26,22 @@ fn accumulate_sensor_data(staged: DataModel, inputs: Vec<DataModel>) -> Result<D
     );
 
     let DataModel::Message(mut result) = staged else {
-        panic!("Unexpected input type");
+        return Err(Error {message: "Unexpected input type.".to_string()});
     };
 
     // Extract payload from message to process
     let staged_payload = result.payload.read();
 
     // Initialize sensor data
     let mut sensor_data = if staged_payload.is_empty() {
-        logger::log(
-            Level::Info,
-            "module-collection/accumulate",
-            "first initialization",
-        );
+        logger::log(Level::Info, "module-collection/accumulate", "first initialization");
         Measurement::SensorData(MeasurementSensorData::new())
     } else {
         logger::log(Level::Info, "module-collection/accumulate", "adding ...");
         match serde_json::from_slice(&staged_payload).unwrap() {
             Measurement::SensorData(sensor_data) => Measurement::SensorData(sensor_data),
             _ => {
-                panic!("Unexpected type for result.");
+                return Err(Error {message: "Unexpected type for result.".to_string()});
             }
         }
     };
@@ -57,7 +56,9 @@ fn accumulate_sensor_data(staged: DataModel, inputs: Vec<DataModel>) -> Result<D
                 // Extract payload from message to process
                 rtsp.frame.read()
             }
-            DataModel::BufferOrBytes(_) => panic!("Unexpected input type"),
+            DataModel::BufferOrBytes(_) => {
+                return Err(Error {message: "Unexpected input type.".to_string()});
+            }
         };
 
         let measurement: Measurement = serde_json::from_slice(&payload).unwrap();
@@ -70,7 +71,7 @@ fn accumulate_sensor_data(staged: DataModel, inputs: Vec<DataModel>) -> Result<D
                         sensor_data.temperature.push(measurement);
                         Measurement::SensorData(sensor_data)
                     }
-                    _ => panic!("Unexpected type for result."),
+                    _ => return Err(Error {message: "Unexpected type for result.".to_string()}),
                 }
             }
             Measurement::Humidity(measurement) => {
@@ -81,7 +82,7 @@ fn accumulate_sensor_data(staged: DataModel, inputs: Vec<DataModel>) -> Result<D
                         sensor_data.humidity.push(measurement);
                         Measurement::SensorData(sensor_data)
                     }
-                    _ => panic!("Unexpected type for result."),
+                    _ => return Err(Error {message: "Unexpected type for result.".to_string()}),
                 }
             }
             Measurement::Object(measurement) => {
@@ -92,10 +93,10 @@ fn accumulate_sensor_data(staged: DataModel, inputs: Vec<DataModel>) -> Result<D
                         sensor_data.object.push(measurement);
                         Measurement::SensorData(sensor_data)
                     }
-                    _ => panic!("Unexpected type for result."),
+                    _ => return Err(Error {message: "Unexpected type for result.".to_string()}),
                 }
             }
-            Measurement::SensorData(_) => panic!("Unexpected measurement type."),
+            Measurement::SensorData(_) => return Err(Error {message: "Unexpected measurement type.".to_string()}),
         }
     }
 

samples/wasm/rust/examples/enrichment/Cargo.toml

@@ -7,7 +7,7 @@ edition = "2021"
 
 [dependencies]
 wit-bindgen = "0.22"
-wasm_graph_sdk = { version = "=1.1.1", registry = "aio-sdks" }
+wasm_graph_sdk = { version = "=1.1.3", registry="aio-wg" }
 serde = { version = "1", default-features = false, features = [
 	"derive",
 ] }
@@ -17,4 +17,3 @@ serde_json = { version = "1", default-features = false, features = [
 
 [lib]
 crate-type = ["cdylib"]
-path = "src/lib.rs"

samples/wasm/rust/examples/enrichment/src/lib.rs

@@ -25,7 +25,7 @@ fn overtemp_check(input: DataModel) -> Result<DataModel, Error> {
 
     // Extract message from input
     let DataModel::Message(message) = input else {
-        panic!("Unexpected input type");
+        return Err(Error {message: "Unexpected input type.".to_string()});
     };
 
     // Extract payload from message to process
@@ -56,7 +56,7 @@ fn overtemp_check(input: DataModel) -> Result<DataModel, Error> {
     let sensor_data = match payload_json {
         Measurement::SensorData(sensor_data) => Measurement::SensorData(sensor_data),
         _ => {
-            panic!("Unexpected type for result.");
+            return Err(Error {message: "Unexpected type for result.".to_string()});
         }
     };
 
@@ -193,7 +193,7 @@ fn overtemp_check(input: DataModel) -> Result<DataModel, Error> {
             sensor_data
         }
         _ => {
-            panic!("Unexpected type for result.");
+            return Err(Error {message: "Unexpected type for result.".to_string()});
         }
     };
 

samples/wasm/rust/examples/format/Cargo.toml

@@ -9,7 +9,7 @@ edition = "2021"
 [dependencies]
 image = { version = "0.24" }
 wit-bindgen = "0.22"
-wasm_graph_sdk = { version = "=1.1.1", registry = "aio-sdks" }
+wasm_graph_sdk = { version = "=1.1.3", registry="aio-wg" }
 serde = { version = "1", default-features = false, features = [
 	"derive",
 ] }
@@ -19,4 +19,3 @@ serde_json = { version = "1", default-features = false, features = [
 
 [lib]
 crate-type = ["cdylib"]
-path = "src/lib.rs"

samples/wasm/rust/examples/format/src/lib.rs

@@ -5,7 +5,6 @@
 #![allow(clippy::missing_safety_doc)]
 
 mod map_format {
-    use core::panic;
 
     use wasm_graph_sdk::logger::{self, Level};
     use wasm_graph_sdk::macros::map_operator;
@@ -44,19 +43,21 @@ mod map_format {
         let _ = metrics::add_to_counter("requests", CounterValue::U64(1), Some(&labels));
 
         let (payload, timestamp) = match input {
-            DataModel::Message(message) => (message.payload.read(), message.timestamp),
+            DataModel::Message(message) => {
+                (message.payload.read(), message.timestamp)
+            },
             DataModel::Snapshot(snapshot) => {
                 let format = match snapshot.format {
                     BufferOrString::Buffer(ref s) => String::from_utf8_lossy(&s.read()).to_string(),
-                    BufferOrString::String(_) => panic!("Unexpected format type"),
+                    BufferOrString::String(ref s) => s.clone(),
                 };
                 if format == FORMAT && snapshot.width == WIDTH && snapshot.height == HEIGHT {
                     // If the snapshot is already in the desired format and size, return it directly
                     return Ok(DataModel::Snapshot(snapshot));
                 }
                 (snapshot.frame.read(), snapshot.timestamp)
-            }
-            DataModel::BufferOrBytes(_) => panic!("Unexpected input type"),
+            },
+            DataModel::BufferOrBytes(_) => return Err(Error {message: "Unexpected input type.".to_string()}),
         };
 
         // Extract payload from message to process
@@ -71,13 +72,11 @@ mod map_format {
                     logger::log(
                         Level::Info,
                         "module-format/map",
-                        &format!(
-                            "Unexpected image format or size: expected {} bytes, got {} bytes",
-                            WIDTH * HEIGHT * CELL_LENGTH,
-                            payload.len()
-                        ),
+                        &format!("Unexpected image format or size: expected {} bytes, got {} bytes",
+                                 WIDTH * HEIGHT * CELL_LENGTH,
+                                 payload.len()),
                     );
-                    panic!("Unexpected image format or size");
+                    return Err(Error {message: "Unexpected image format or size.".to_string()});
                 }
                 // If the image format cannot be guessed but the payload size is correct, it
                 // is possibly already in the desired format. So we will return it directly.

samples/wasm/rust/examples/humidity/Cargo.toml

@@ -7,7 +7,7 @@ edition = "2021"
 
 [dependencies]
 wit-bindgen = "0.22"
-wasm_graph_sdk = { version = "=1.1.1", registry = "aio-sdks" }
+wasm_graph_sdk = { version = "=1.1.3", registry="aio-wg" }
 serde = { version = "1", default-features = false, features = [
 	"derive",
 ] }

samples/wasm/rust/examples/humidity/src/lib.rs

@@ -4,22 +4,23 @@
 // Generated by `wit_bindgen::generate` expansion.
 #![allow(clippy::missing_safety_doc)]
 
-use core::panic;
-
 use wasm_graph_sdk::logger::{self, Level};
 use wasm_graph_sdk::macros::accumulate_operator;
 use wasm_graph_sdk::metrics::{self, CounterValue, Label};
 
 #[accumulate_operator]
-fn accumulate_humidity(staged: DataModel, inputs: Vec<DataModel>) -> Result<DataModel, Error> {
+fn accumulate_humidity(
+    staged: DataModel,
+    inputs: Vec<DataModel>,
+) -> Result<DataModel, Error> {
     let labels = vec![Label {
         key: "module".to_owned(),
         value: "module-humidity/accumulate".to_owned(),
     }];
     let _ = metrics::add_to_counter("requests", CounterValue::U64(1), Some(&labels));
 
     let DataModel::Message(mut result) = staged else {
-        panic!("Unexpected input type");
+        return Err(Error {message: "Unexpected input type.".to_string()});
     };
 
     // Extract payload from message to process
@@ -56,16 +57,12 @@ fn accumulate_humidity(staged: DataModel, inputs: Vec<DataModel>) -> Result<Data
         let (ts, topic, payload) = match input {
             DataModel::Message(temp) => {
                 // Extract payload from message to process
-                (
-                    temp.timestamp.timestamp,
-                    temp.topic,
-                    match &temp.payload {
-                        BufferOrBytes::Buffer(buffer) => buffer.read(),
-                        BufferOrBytes::Bytes(bytes) => bytes.clone(),
-                    },
-                )
+                (temp.timestamp.timestamp, temp.topic, match &temp.payload {
+                    BufferOrBytes::Buffer(buffer) => buffer.read(),
+                    BufferOrBytes::Bytes(bytes) => bytes.clone(),
+                })
             }
-            _ => panic!("Unexpected input type"),
+            _ => return Err(Error {message: "Unexpected input type.".to_string()}),
         };
 
         let measurement: Measurement = serde_json::from_slice(&payload).unwrap();
@@ -94,7 +91,7 @@ fn accumulate_humidity(staged: DataModel, inputs: Vec<DataModel>) -> Result<Data
                 // Set the topic
                 result_topic = topic;
             }
-            _ => panic!("Unexpected measurement type."),
+            _ => return Err(Error { message: "Unexpected measurement type.".to_string() }),
         }
     }
 
@@ -106,7 +103,11 @@ fn accumulate_humidity(staged: DataModel, inputs: Vec<DataModel>) -> Result<Data
         r#"{{"humidity":{{"count":{count},"max":{max},"min":{min},"average":{avg},"last":{last}}}}}"#,
     );
 
-    logger::log(Level::Info, "module-humidity/accumulate", &payload_str);
+    logger::log(
+        Level::Info,
+        "module-humidity/accumulate",
+        &payload_str,
+    );
 
     let payload = payload_str.as_bytes().to_vec();
     result.payload = BufferOrBytes::Bytes(payload);
@@ -116,6 +117,7 @@ fn accumulate_humidity(staged: DataModel, inputs: Vec<DataModel>) -> Result<Data
     Ok(DataModel::Message(result))
 }
 
+
 #[derive(Debug, PartialEq, serde::Deserialize, serde::Serialize)]
 pub enum Measurement {
     #[serde(rename = "temperature")]

samples/wasm/rust/examples/otel-enrich/Cargo.toml

@@ -1,13 +1,15 @@
 [package]
 name = "otel-enrich"
-version = "0.1.0"
+version = "0.0.1"
 authors = ["Wasm Graph Devs"]
 license = "MIT"
 edition = "2021"
 
 [dependencies]
 wit-bindgen = "0.22"
-wasm_graph_sdk = { version = "=1.1.1", registry = "aio-sdks" }
+wasm_graph_sdk = { version = "=1.1.3", registry="aio-wg" }
+serde = { version = "1.0", features = ["derive"] } 
+serde_json = "1.0" 
 
 [lib]
 crate-type = ["cdylib"]

samples/wasm/rust/examples/otel-enrich/src/lib.rs

@@ -1,7 +1,6 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
-// Generated by `wit_bindgen::generate` expansion.
 #![allow(clippy::missing_safety_doc)]
 mod otel_enrich {
 
@@ -60,12 +59,12 @@ mod otel_enrich {
         }];
         let _ = metrics::add_to_counter("requests", CounterValue::U64(1), Some(&labels));
 
+        // Extract message from input
         let DataModel::Message(mut result) = input else {
-            return Err(Error {
-                message: "Unexpected input type".to_string(),
-            });
+            return Err(Error { message: "Unexpected input type".to_string() });
         };
 
+        // Get list of keys to enrich from ATTR_KEYS
         let Some(attr_keys) = ATTR_KEYS.get() else {
             logger::log(
                 Level::Error,
@@ -81,9 +80,10 @@ mod otel_enrich {
         for key in keys {
             let key = key.trim();
             if key.is_empty() {
-                continue;
+                continue; // Skip empty keys
             }
 
+            // Get value from DSS
             let state_store_get = state_store::get(key.as_bytes(), None);
             match state_store_get
                 .as_ref()
@@ -96,8 +96,12 @@ mod otel_enrich {
                         &format!("Retrieved from state store: key='{key}', value='{value:?}'"),
                     );
                     let stringify = String::from_utf8_lossy(value).to_string();
-
-                    // Add to user properties with "otel/" prefix so we can add as attributes later for OTel
+                    logger::log(
+                        Level::Info,
+                        "module-otel-enrich/map",
+                        &format!("'{key}' found in state store with value '{stringify}'"),
+                    );
+                    // Add the key-value pair to user_properties_vec
                     user_properties_vec.push((
                         BufferOrString::String(format!("otel/{key}")),
                         BufferOrString::String(stringify),
@@ -109,21 +113,28 @@ mod otel_enrich {
                         "module-otel-enrich/map",
                         &format!("'{key}' not found in state store, skipping"),
                     );
+                    continue; // Skip if key not found
                 }
                 Err(err) => {
                     logger::log(
                         Level::Error,
                         "module-otel-enrich/map",
                         &format!("Failed to get value for key '{key}': {err}"),
                     );
+                    continue; // Skip if error occurs
                 }
             }
         }
 
-        result.properties = MessageProperties {
-            user_properties: user_properties_vec,
-        };
+        result.topic = BufferOrBytes::Bytes("sensors".as_bytes().to_vec());
+        result.properties.user_properties = user_properties_vec;
+
+        logger::log(
+            Level::Info,
+            "module-otel-enrich/map",
+            &format!("result: {result:?}"),
+        );
 
         Ok(DataModel::Message(result))
     }
-}
+}