feat: use version from SmartModuleInstance

infinyon · Jul 25, 2023 · e0041f3 · e0041f3
1 parent ae33a22
commit e0041f3
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 9 deletions.
diff --git a/crates/fluvio-smartengine/src/engine/wasmtime/engine.rs b/crates/fluvio-smartengine/src/engine/wasmtime/engine.rs
@@ -78,7 +78,7 @@ impl SmartModuleChainBuilder {
             let init = SmartModuleInit::try_instantiate(&ctx, &mut state)?;
             let look_back = SmartModuleLookBack::try_instantiate(&ctx, &mut state)?;
             let transform = create_transform(&ctx, config.initial_data, &mut state)?;
-            let mut instance = SmartModuleInstance::new(ctx, init, look_back, transform);
+            let mut instance = SmartModuleInstance::new(ctx, init, look_back, transform, version);
 
             instance.call_init(&mut state)?;
             instances.push(instance);
@@ -160,10 +160,8 @@ impl SmartModuleChainInstance {
                     // encountered error, we stop processing and return partial output
                     return Ok(output);
                 } else {
-                    next_input = SmartModuleInput::try_from_records(
-                        output.successes,
-                        DEFAULT_SMARTENGINE_VERSION,
-                    )?;
+                    next_input =
+                        SmartModuleInput::try_from_records(output.successes, instance.version())?;
                     next_input.set_base_offset(base_offset);
                     next_input.set_base_timestamp(base_timestamp);
                 }
@@ -202,7 +200,7 @@ impl SmartModuleChainInstance {
                 debug!("look_back on instance");
                 let records: Vec<Record> = read_fn(lookback).await?;
                 let input: SmartModuleInput =
-                    SmartModuleInput::try_from_records(records, DEFAULT_SMARTENGINE_VERSION)?;
+                    SmartModuleInput::try_from_records(records, instance.version())?;
 
                 metrics.add_bytes_in(input.raw_bytes().len() as u64);
                 self.store.top_up_fuel();

diff --git a/crates/fluvio-smartengine/src/engine/wasmtime/instance.rs b/crates/fluvio-smartengine/src/engine/wasmtime/instance.rs
@@ -6,7 +6,7 @@ use tracing::debug;
 use anyhow::{Error, Result};
 use wasmtime::{Memory, Module, Caller, Extern, Instance, Func, AsContextMut, AsContext};
 
-use fluvio_protocol::{Encoder, Decoder};
+use fluvio_protocol::{Encoder, Decoder, Version};
 
 use fluvio_smartmodule::dataplane::smartmodule::{
     SmartModuleExtraParams, SmartModuleInput, SmartModuleOutput, SmartModuleInitInput,
@@ -25,6 +25,7 @@ pub(crate) struct SmartModuleInstance {
     init: Option<SmartModuleInit>,
     look_back: Option<SmartModuleLookBack>,
     transform: Box<dyn DowncastableTransform>,
+    version: Version,
 }
 
 impl SmartModuleInstance {
@@ -44,12 +45,14 @@ impl SmartModuleInstance {
         init: Option<SmartModuleInit>,
         look_back: Option<SmartModuleLookBack>,
         transform: Box<dyn DowncastableTransform>,
+        version: Version,
     ) -> Self {
         Self {
             ctx,
             init,
             look_back,
             transform,
+            version,
         }
     }
 
@@ -90,13 +93,18 @@ impl SmartModuleInstance {
         self.look_back.as_ref()?; // return None if there is no function
         self.ctx.lookback
     }
+
+    /// Retrieves SmartModule Version
+    pub fn version(&self) -> Version {
+        self.version
+    }
 }
 
 pub(crate) struct SmartModuleInstanceContext {
     instance: Instance,
     records_cb: Arc<RecordsCallBack>,
     params: SmartModuleExtraParams,
-    version: i16,
+    version: Version,
     lookback: Option<Lookback>,
 }
 
@@ -113,7 +121,7 @@ impl SmartModuleInstanceContext {
         state: &mut WasmState,
         module: Module,
         params: SmartModuleExtraParams,
-        version: i16,
+        version: Version,
         lookback: Option<Lookback>,
     ) -> Result<Self, EngineError> {
         debug!("creating WasmModuleInstance");