move transform unit tests to common

xxchan · xxchan · Feb 26, 2023 · Feb 26, 2023 · Apr 9, 2023 · Feb 26, 2023
commit d0bb940084f4376e206edf5abcb9f10c50d239e2
diff --git a/crates/fluvio-smartengine/Cargo.toml b/crates/fluvio-smartengine/Cargo.toml
@@ -12,7 +12,7 @@ description = "The official Fluvio SmartEngine"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
-engine = ["wasmtime-engine", "wasmedge-engine"]
+engine = []
 wasi = ["wasmtime-wasi", "wasmtime-engine", "engine"]
 transformation = ["serde_json", "serde_yaml"]
 default = ["engine"]
@@ -25,7 +25,8 @@ thiserror = { workspace = true }
 anyhow = { workspace = true }
 # Use the Clone trait
 wasmedge-sdk = { git = "https://github.com/WasmEdge/WasmEdge", rev = "fe5fe2c", optional = true }
-serde = { workspace = true,  features = ['derive'] }
+# std is needed for ser/de AtomicU64
+serde = { workspace = true,  features = ['derive', 'std'] }
 serde_json = { workspace = true, optional = true }
 serde_yaml = { workspace = true, default-features = false, optional = true }
 cfg-if = { workspace = true }

diff --git a/crates/fluvio-smartengine/src/engine/common/init.rs b/crates/fluvio-smartengine/src/engine/common/init.rs
@@ -0,0 +1,63 @@
+use super::{WasmFn, WasmInstance};
+use anyhow::Result;
+use fluvio_smartmodule::dataplane::smartmodule::{
+    SmartModuleInitInput, SmartModuleInitErrorStatus, SmartModuleInitOutput,
+};
+
+pub(crate) const INIT_FN_NAME: &str = "init";
+
+pub(crate) struct SmartModuleInit<F: WasmFn>(F);
+
+impl<F: WasmFn> std::fmt::Debug for SmartModuleInit<F> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "InitFn")
+    }
+}
+
+impl<F: WasmFn + Send + Sync> SmartModuleInit<F> {
+    pub(crate) fn try_instantiate<I>(
+        instance: &mut I,
+        ctx: &mut <I as WasmInstance>::Context,
+    ) -> Result<Option<Self>>
+    where
+        I: WasmInstance<Func = F>,
+        F: WasmFn<Context = I::Context>,
+    {
+        match instance.get_fn(INIT_FN_NAME, ctx)? {
+            Some(func) => Ok(Some(Self(func))),
+            None => Ok(None),
+        }
+    }
+}
+
+impl<F: WasmFn + Send + Sync> SmartModuleInit<F> {
+    /// initialize SmartModule
+    pub(crate) fn initialize<I>(
+        &mut self,
+        input: SmartModuleInitInput,
+        instance: &mut I,
+        ctx: &mut I::Context,
+    ) -> Result<()>
+    where
+        I: WasmInstance,
+        F: WasmFn<Context = I::Context>,
+    {
+        let (ptr, len, version) = instance.write_input(&input, ctx)?;
+        let init_output = self.0.call(ptr, len, version, ctx)?;
+
+        if init_output < 0 {
+            let internal_error = SmartModuleInitErrorStatus::try_from(init_output)
+                .unwrap_or(SmartModuleInitErrorStatus::UnknownError);
+
+            match internal_error {
+                SmartModuleInitErrorStatus::InitError => {
+                    let output: SmartModuleInitOutput = instance.read_output(ctx)?;
+                    Err(output.error.into())
+                }
+                _ => Err(internal_error.into()),
+            }
+        } else {
+            Ok(())
+        }
+    }
+}
diff --git a/crates/fluvio-smartengine/src/engine/common/mod.rs b/crates/fluvio-smartengine/src/engine/common/mod.rs
@@ -0,0 +1,87 @@
+mod init;
+mod transforms;
+pub(crate) use init::*;
+pub(crate) use transforms::*;
+
+use anyhow::Result;
+use fluvio_protocol::{Decoder, Encoder};
+use fluvio_smartmodule::dataplane::smartmodule::{
+    SmartModuleExtraParams, SmartModuleInitInput, SmartModuleInput, SmartModuleOutput,
+};
+
+pub(crate) trait WasmInstance {
+    type Context;
+    type Func: WasmFn<Context = Self::Context> + Send + Sync + 'static;
+
+    fn params(&self) -> SmartModuleExtraParams;
+
+    fn get_fn(&self, name: &str, ctx: &mut Self::Context) -> Result<Option<Self::Func>>;
+
+    fn write_input<E: Encoder>(
+        &mut self,
+        input: &E,
+        ctx: &mut Self::Context,
+    ) -> Result<(i32, i32, i32)>;
+    fn read_output<D: Decoder + Default>(&mut self, ctx: &mut Self::Context) -> Result<D>;
+}
+
+/// All smartmodule wasm functions have the same ABI:
+/// `(ptr: *mut u8, len: usize, version: i16) -> i32`, which is `(param i32 i32 i32) (result i32)` in wasm.
+pub(crate) trait WasmFn {
+    type Context;
+    fn call(&self, ptr: i32, len: i32, version: i32, ctx: &mut Self::Context) -> Result<i32>;
+}
+
+pub(crate) struct SmartModuleInstance<I: WasmInstance<Func = F>, F: WasmFn> {
+    pub instance: I,
+    pub transform: Box<dyn DowncastableTransform<I>>,
+    pub init: Option<SmartModuleInit<F>>,
+}
+
+impl<I: WasmInstance<Func = F>, F: WasmFn + Send + Sync> SmartModuleInstance<I, F> {
+    #[cfg(test)]
+    #[allow(clippy::borrowed_box)]
+    pub(crate) fn transform(&self) -> &Box<dyn DowncastableTransform<I>> {
+        &self.transform
+    }
+
+    #[cfg(test)]
+    pub(crate) fn get_init(&self) -> &Option<SmartModuleInit<F>> {
+        &self.init
+    }
+
+    pub(crate) fn new(
+        instance: I,
+        init: Option<SmartModuleInit<F>>,
+        transform: Box<dyn DowncastableTransform<I>>,
+    ) -> Self {
+        Self {
+            instance,
+            init,
+            transform,
+        }
+    }
+
+    pub(crate) fn process(
+        &mut self,
+        input: SmartModuleInput,
+        ctx: &mut I::Context,
+    ) -> Result<SmartModuleOutput> {
+        self.transform.process(input, &mut self.instance, ctx)
+    }
+
+    pub fn init<C>(&mut self, ctx: &mut I::Context) -> Result<()>
+    where
+        I: WasmInstance<Context = C>,
+        F: WasmFn<Context = C>,
+    {
+        if let Some(init) = &mut self.init {
+            let input = SmartModuleInitInput {
+                params: self.instance.params(),
+            };
+            init.initialize(input, &mut self.instance, ctx)
+        } else {
+            Ok(())
+        }
+    }
+}
diff --git a/...c/engine/wasmedge/transforms/aggregate.rs → ...src/engine/common/transforms/aggregate.rs b/...c/engine/wasmedge/transforms/aggregate.rs → ...src/engine/common/transforms/aggregate.rs
@@ -13,8 +13,18 @@ mod test {
 
     const SM_AGGEGRATE: &str = "fluvio_smartmodule_aggregate";
 
-    type AggregateTransform =
-        crate::engine::common::AggregateTransform<crate::engine::wasmedge::instance::WasmedgeFn>;
+    cfg_if::cfg_if! {
+
+        if #[cfg(feature = "wasmtime-engine")] {
+            type AggregateTransform =
+                crate::engine::common::AggregateTransform<crate::engine::wasmtime::instance::WasmtimeFn>;
+        } else if #[cfg(feature = "wasmedge-engine")] {
+            type AggregateTransform =
+                crate::engine::common::AggregateTransform<crate::engine::wasmedge::instance::WasmedgeFn>;
+        } else {
+            compile_error!("no engine specified");
+        }
+    }
 
     #[ignore]
     #[test]

diff --git a/...c/engine/wasmedge/transforms/array_map.rs → ...src/engine/common/transforms/array_map.rs b/...c/engine/wasmedge/transforms/array_map.rs → ...src/engine/common/transforms/array_map.rs
diff --git a/.../src/engine/wasmedge/transforms/filter.rs → ...ne/src/engine/common/transforms/filter.rs b/.../src/engine/wasmedge/transforms/filter.rs → ...ne/src/engine/common/transforms/filter.rs
diff --git a/.../engine/wasmedge/transforms/filter_map.rs → ...rc/engine/common/transforms/filter_map.rs b/.../engine/wasmedge/transforms/filter_map.rs → ...rc/engine/common/transforms/filter_map.rs
@@ -12,7 +12,6 @@ mod test {
 
     const SM_FILTER_MAP: &str = "fluvio_smartmodule_filter_map";
 
-
     #[ignore]
     #[test]
     fn test_filter_map() {

diff --git a/...ine/src/engine/wasmedge/transforms/map.rs → ...ngine/src/engine/common/transforms/map.rs b/...ine/src/engine/wasmedge/transforms/map.rs → ...ngine/src/engine/common/transforms/map.rs
@@ -9,7 +9,7 @@ mod test {
     use crate::engine::{SmartEngine, SmartModuleChainBuilder};
     use crate::engine::config::SmartModuleConfig;
     use crate::engine::fixture::read_wasm_module;
-    
+
     const SM_MAP: &str = "fluvio_smartmodule_map";
 
     #[ignore]

diff --git a/...s/fluvio-smartengine/src/engine/common.rs → ...ngine/src/engine/common/transforms/mod.rs b/...s/fluvio-smartengine/src/engine/common.rs → ...ngine/src/engine/common/transforms/mod.rs
@@ -1,40 +1,20 @@
 use std::any::Any;
-
 use anyhow::Result;
-use fluvio_protocol::{Decoder, Encoder};
 use fluvio_smartmodule::dataplane::smartmodule::{
-    SmartModuleExtraParams, SmartModuleInitErrorStatus, SmartModuleInitInput,
-    SmartModuleInitOutput, SmartModuleInput, SmartModuleOutput, SmartModuleTransformErrorStatus,
+    SmartModuleInput, SmartModuleOutput, SmartModuleTransformErrorStatus,
     SmartModuleAggregateInput, SmartModuleAggregateOutput,
 };
 use tracing::debug;
 
-use crate::SmartModuleInitialData;
-
-use super::error::EngineError;
-
-pub(crate) trait WasmInstance {
-    type Context;
-    type Func: WasmFn<Context = Self::Context> + Send + Sync + 'static;
-
-    fn params(&self) -> SmartModuleExtraParams;
+use crate::{SmartModuleInitialData, engine::error::EngineError};
 
-    fn get_fn(&self, name: &str, ctx: &mut Self::Context) -> Result<Option<Self::Func>>;
+use super::{WasmInstance, WasmFn};
 
-    fn write_input<E: Encoder>(
-        &mut self,
-        input: &E,
-        ctx: &mut Self::Context,
-    ) -> Result<(i32, i32, i32)>;
-    fn read_output<D: Decoder + Default>(&mut self, ctx: &mut Self::Context) -> Result<D>;
-}
-
-/// All smartmodule wasm functions have the same ABI:
-/// `(ptr: *mut u8, len: usize, version: i16) -> i32`, which is `(param i32 i32 i32) (result i32)` in wasm.
-pub(crate) trait WasmFn {
-    type Context;
-    fn call(&self, ptr: i32, len: i32, version: i32, ctx: &mut Self::Context) -> Result<i32>;
-}
+mod filter;
+mod map;
+mod filter_map;
+mod array_map;
+mod aggregate;
 
 pub(crate) trait SmartModuleTransform<I: WasmInstance>: Send + Sync {
     /// transform records
@@ -222,115 +202,3 @@ where
         Err(EngineError::UnknownSmartModule.into())
     }
 }
-
-pub(crate) const INIT_FN_NAME: &str = "init";
-
-pub(crate) struct SmartModuleInit<F: WasmFn>(F);
-
-impl<F: WasmFn> std::fmt::Debug for SmartModuleInit<F> {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "InitFn")
-    }
-}
-
-impl<F: WasmFn + Send + Sync> SmartModuleInit<F> {
-    pub(crate) fn try_instantiate<I>(
-        instance: &mut I,
-        ctx: &mut <I as WasmInstance>::Context,
-    ) -> Result<Option<Self>>
-    where
-        I: WasmInstance<Func = F>,
-        F: WasmFn<Context = I::Context>,
-    {
-        match instance.get_fn(INIT_FN_NAME, ctx)? {
-            Some(func) => Ok(Some(Self(func))),
-            None => Ok(None),
-        }
-    }
-}
-
-impl<F: WasmFn + Send + Sync> SmartModuleInit<F> {
-    /// initialize SmartModule
-    pub(crate) fn initialize<I>(
-        &mut self,
-        input: SmartModuleInitInput,
-        instance: &mut I,
-        ctx: &mut I::Context,
-    ) -> Result<()>
-    where
-        I: WasmInstance,
-        F: WasmFn<Context = I::Context>,
-    {
-        let (ptr, len, version) = instance.write_input(&input, ctx)?;
-        let init_output = self.0.call(ptr, len, version, ctx)?;
-
-        if init_output < 0 {
-            let internal_error = SmartModuleInitErrorStatus::try_from(init_output)
-                .unwrap_or(SmartModuleInitErrorStatus::UnknownError);
-
-            match internal_error {
-                SmartModuleInitErrorStatus::InitError => {
-                    let output: SmartModuleInitOutput = instance.read_output(ctx)?;
-                    Err(output.error.into())
-                }
-                _ => Err(internal_error.into()),
-            }
-        } else {
-            Ok(())
-        }
-    }
-}
-
-pub(crate) struct SmartModuleInstance<I: WasmInstance<Func = F>, F: WasmFn> {
-    pub instance: I,
-    pub transform: Box<dyn DowncastableTransform<I>>,
-    pub init: Option<SmartModuleInit<F>>,
-}
-
-impl<I: WasmInstance<Func = F>, F: WasmFn + Send + Sync> SmartModuleInstance<I, F> {
-    #[cfg(test)]
-    #[allow(clippy::borrowed_box)]
-    pub(crate) fn transform(&self) -> &Box<dyn DowncastableTransform<I>> {
-        &self.transform
-    }
-
-    #[cfg(test)]
-    pub(crate) fn get_init(&self) -> &Option<SmartModuleInit<F>> {
-        &self.init
-    }
-
-    pub(crate) fn new(
-        instance: I,
-        init: Option<SmartModuleInit<F>>,
-        transform: Box<dyn DowncastableTransform<I>>,
-    ) -> Self {
-        Self {
-            instance,
-            init,
-            transform,
-        }
-    }
-
-    pub(crate) fn process(
-        &mut self,
-        input: SmartModuleInput,
-        ctx: &mut I::Context,
-    ) -> Result<SmartModuleOutput> {
-        self.transform.process(input, &mut self.instance, ctx)
-    }
-
-    pub fn init<C>(&mut self, ctx: &mut I::Context) -> Result<()>
-    where
-        I: WasmInstance<Context = C>,
-        F: WasmFn<Context = C>,
-    {
-        if let Some(init) = &mut self.init {
-            let input = SmartModuleInitInput {
-                params: self.instance.params(),
-            };
-            init.initialize(input, &mut self.instance, ctx)
-        } else {
-            Ok(())
-        }
-    }
-}
-Original file line number
+Diff line change
@@ Expand Up / @@ -12,7 +12,6 @@ mod test { @@
         const SM_FILTER_MAP: &str = "fluvio_smartmodule_filter_map";
         #[ignore]
         #[test]
         fn test_filter_map() {
@@ Expand Down @@