[Merged by Bors] - docs: example of aggregation sm with initial value

smartmodule/examples/Cargo.toml

@@ -4,6 +4,7 @@ members = [
     "aggregate-sum",
     "aggregate-json",
     "aggregate-average",
+    "aggregate-init",
     "filter",
     "filter_init",
     "filter_with_param",
@@ -29,3 +30,9 @@ fluvio-smartmodule = { path = "../../crates/fluvio-smartmodule" }
 [profile.release]
 lto = true
 strip = "symbols"
+
+[profile.release-lto]
+inherits = "release"
+lto = true
+strip = "symbols"
+

smartmodule/examples/aggregate-init/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "sm-integer-sum-aggegrate"
+version = "0.0.0"
+authors = ["Fluvio Contributors <team@fluvio.io>"]
+edition = "2021"
+publish = false
+
+[lib]
+crate-type = ['cdylib']
+
+[dependencies]
+fluvio-smartmodule = { workspace = true }
+once_cell = { version = "1.17" }
+

smartmodule/examples/aggregate-init/README.md

@@ -0,0 +1,20 @@
+In order to build and use this SmartModule in your cluster, follow these steps:
+
+1. Install `smdk` tool:
+```bash
+fluvio install smdk
+```
+2. Build:
+```bash
+smdk build
+```
+
+3. Load built SmartModule into the cluster:
+```bash
+smdk load
+```
+
+After that, you can consume from your topic and apply the aggregation as trasnformation:
+```bash
+fluvio consume test-aggr --transforms-file transforms.yaml
+```

smartmodule/examples/aggregate-init/SmartModule.toml

@@ -0,0 +1,14 @@
+[package]
+name = "integer-sum-aggegrate"
+group = "infinyon"
+version = "0.1.0"
+apiVersion = "0.1.0"
+description = ""
+license = "Apache-2.0"
+visibility = "private"
+
+[[params]]
+name = "initial_value"
+description = "initial value for aggregation"
+optional = false
+

smartmodule/examples/aggregate-init/src/lib.rs

@@ -0,0 +1,65 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use fluvio_smartmodule::{
+    dataplane::smartmodule::SmartModuleExtraParams, smartmodule, Record, RecordData, Result,
+};
+
+use once_cell::sync::OnceCell;
+
+static INITIAL_VALUE: OnceCell<UseOnce<RecordData>> = OnceCell::new();
+
+const PARAM_NAME: &str = "initial_value";
+
+#[smartmodule(aggregate)]
+pub fn aggregate(accumulator: RecordData, current: &Record) -> Result<RecordData> {
+    let accumulator = if let Some(initial_value) = INITIAL_VALUE.get() {
+        initial_value.get_or(&accumulator)
+    } else {
+        &accumulator
+    };
+    // Parse the accumulator and current record as strings
+    let accumulator_string = std::str::from_utf8(accumulator.as_ref())?;
+    let current_string = std::str::from_utf8(current.value.as_ref())?;
+
+    // Parse the strings into integers
+    let accumulator_int = accumulator_string.trim().parse::<i32>().unwrap_or(0);
+    let current_int = current_string.trim().parse::<i32>()?;
+
+    // Take the sum of the two integers and return it as a string
+    let sum = accumulator_int + current_int;
+    Ok(sum.to_string().into())
+}
+
+#[smartmodule(init)]
+fn init(params: SmartModuleExtraParams) -> Result<()> {
+    if let Some(raw_spec) = params.get(PARAM_NAME) {
+        INITIAL_VALUE
+            .set(UseOnce::new(raw_spec.as_str().into()))
+            .expect("initial value is already initialized");
+    };
+    Ok(())
+}
+
+#[derive(Debug)]
+struct UseOnce<T> {
+    value: T,
+    used: AtomicBool,
+}
+
+impl<T> UseOnce<T> {
+    fn new(value: T) -> Self {
+        Self {
+            value,
+            used: AtomicBool::new(false),
+        }
+    }
+
+    fn get_or<'a: 'b, 'b>(&'a self, default: &'b T) -> &'b T {
+        if self.used.load(Ordering::SeqCst) {
+            default
+        } else {
+            self.used.store(true, Ordering::SeqCst);
+            &self.value
+        }
+    }
+}

smartmodule/examples/aggregate-init/transforms.yaml

@@ -0,0 +1,4 @@
+transforms:
+  - uses: infinyon/integer-sum-aggegrate@0.1.0
+    with:
+      initial_value: "10"