Migrate spin-redis to new core

Signed-off-by: Lann Martin <lann.martin@fermyon.com>

Author: lann

Cargo.lock

@@ -39,7 +39,6 @@ spin-manifest = { path = "crates/manifest" }
 spin-publish = { path = "crates/publish" }
 spin-redis-engine = { path = "crates/redis" }
 spin-templates = { path = "crates/templates" }
-spin-trigger = { path = "crates/trigger" }
 spin-trigger-new = { path = "crates/trigger-new" }
 tempfile = "3.3.0"
 tokio = { version = "1.11", features = [ "full" ] }

Cargo.toml

@@ -11,12 +11,12 @@ doctest = false
 anyhow = "1.0"
 async-trait = "0.1"
 futures = "0.3"
-spin-engine = { path = "../engine" }
-spin-manifest = { path = "../manifest" }
-spin-trigger = { path = "../trigger" }
+serde = "1"
+spin-app = { path = "../app" }
+spin-core = { path = "../core" }
+spin-trigger-new = { path = "../trigger-new" }
 redis = { version = "0.21", features = [ "tokio-comp" ] }
 tracing = { version = "0.1", features = [ "log" ] }
-wasmtime = "0.39.1"
 
 [dependencies.wit-bindgen-wasmtime]
 git = "https://github.com/bytecodealliance/wit-bindgen"
@@ -25,4 +25,3 @@ features = ["async"]
 
 [dev-dependencies]
 spin-testing = { path = "../testing" }
-tokio = { version = "1", features = [ "full" ] }

crates/redis/Cargo.toml

@@ -2,100 +2,81 @@
 
 mod spin;
 
-use std::{collections::HashMap, sync::Arc};
+use std::collections::HashMap;
 
-use anyhow::Result;
+use anyhow::{Context, Result};
 use async_trait::async_trait;
 use futures::StreamExt;
 use redis::{Client, ConnectionLike};
-use spin_manifest::{ComponentMap, RedisConfig, RedisTriggerConfiguration, TriggerConfig};
-use spin_redis::SpinRedisData;
-use spin_trigger::{cli::NoArgs, TriggerExecutor};
+use serde::{de::IgnoredAny, Deserialize, Serialize};
+use spin_trigger_new::{cli::NoArgs, TriggerAppEngine, TriggerExecutor};
 
 use crate::spin::SpinRedisExecutor;
 
 wit_bindgen_wasmtime::import!({paths: ["../../wit/ephemeral/spin-redis.wit"], async: *});
 
-type ExecutionContext = spin_engine::ExecutionContext<SpinRedisData>;
-type RuntimeContext = spin_engine::RuntimeContext<SpinRedisData>;
+pub(crate) type RuntimeData = spin_redis::SpinRedisData;
+pub(crate) type Store = spin_core::Store<RuntimeData>;
 
 /// The Spin Redis trigger.
-#[derive(Clone)]
 pub struct RedisTrigger {
-    /// Trigger configuration.
-    trigger_config: RedisTriggerConfiguration,
-    /// Component trigger configurations.
-    component_triggers: ComponentMap<RedisConfig>,
-    /// Spin execution context.
-    engine: Arc<ExecutionContext>,
-    /// Map from channel name to tuple of component name & index.
-    subscriptions: HashMap<String, usize>,
+    engine: TriggerAppEngine<Self>,
+    // Redis address to connect to
+    address: String,
+    // Mapping of subscription channels to component IDs
+    channel_components: HashMap<String, String>,
 }
 
-pub struct RedisTriggerConfig(String, RedisConfig);
-
-impl TryFrom<(String, TriggerConfig)> for RedisTriggerConfig {
-    type Error = spin_manifest::Error;
-
-    fn try_from((component, config): (String, TriggerConfig)) -> Result<Self, Self::Error> {
-        Ok(RedisTriggerConfig(component, config.try_into()?))
-    }
+/// Redis trigger configuration.
+#[derive(Clone, Debug, Default, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct RedisTriggerConfig {
+    /// Component ID to invoke
+    pub component: String,
+    /// Channel to subscribe to
+    pub channel: String,
+    /// Trigger executor (currently unused)
+    #[serde(default, skip_serializing)]
+    pub executor: IgnoredAny,
 }
 
 #[async_trait]
 impl TriggerExecutor for RedisTrigger {
-    type GlobalConfig = RedisTriggerConfiguration;
+    const TRIGGER_TYPE: &'static str = "redis";
+    type RuntimeData = RuntimeData;
     type TriggerConfig = RedisTriggerConfig;
     type RunConfig = NoArgs;
-    type RuntimeContext = SpinRedisData;
-
-    fn new(
-        execution_context: ExecutionContext,
-        global_config: Self::GlobalConfig,
-        trigger_configs: impl IntoIterator<Item = Self::TriggerConfig>,
-    ) -> Result<Self> {
-        let component_triggers: ComponentMap<RedisConfig> = trigger_configs
-            .into_iter()
-            .map(|config| (config.0, config.1))
-            .collect();
-        let subscriptions = execution_context
-            .config
-            .components
-            .iter()
-            .enumerate()
-            .filter_map(|(idx, component)| {
-                component_triggers
-                    .get(&component.id)
-                    .map(|redis_config| (redis_config.channel.clone(), idx))
-            })
+
+    fn new(engine: TriggerAppEngine<Self>) -> Result<Self> {
+        let address = engine
+            .app()
+            .require_metadata("redis_address")
+            .context("Failed to configure Redis trigger")?;
+
+        let channel_components = engine
+            .trigger_configs()
+            .map(|(_, config)| (config.channel.clone(), config.component.clone()))
             .collect();
 
         Ok(Self {
-            trigger_config: global_config,
-            component_triggers,
-            engine: Arc::new(execution_context),
-            subscriptions,
+            engine,
+            address,
+            channel_components,
         })
     }
 
     /// Run the Redis trigger indefinitely.
     async fn run(self, _config: Self::RunConfig) -> Result<()> {
-        let address = self.trigger_config.address.as_str();
+        let address = &self.address;
 
         tracing::info!("Connecting to Redis server at {}", address);
         let mut client = Client::open(address.to_string())?;
         let mut pubsub = client.get_async_connection().await?.into_pubsub();
 
         // Subscribe to channels
-        for (subscription, idx) in self.subscriptions.iter() {
-            let name = &self.engine.config.components[*idx].id;
-            tracing::info!(
-                "Subscribed component #{} ({}) to channel: {}",
-                idx,
-                name,
-                subscription
-            );
-            pubsub.subscribe(subscription).await?;
+        for (channel, component) in self.channel_components.iter() {
+            tracing::info!("Subscribing component {component:?} to channel {channel:?}");
+            pubsub.subscribe(channel).await?;
         }
 
         let mut stream = pubsub.on_message();
@@ -120,35 +101,12 @@ impl RedisTrigger {
         let channel = msg.get_channel_name();
         tracing::info!("Received message on channel {:?}", channel);
 
-        if let Some(idx) = self.subscriptions.get(channel).copied() {
-            let component = &self.engine.config.components[idx];
-            let executor = self
-                .component_triggers
-                .get(&component.id)
-                .and_then(|t| t.executor.clone())
-                .unwrap_or_default();
-
-            let follow = self
-                .engine
-                .config
-                .follow_components
-                .should_follow(&component.id);
-
-            match executor {
-                spin_manifest::RedisExecutor::Spin => {
-                    tracing::trace!("Executing Spin Redis component {}", component.id);
-                    let executor = SpinRedisExecutor;
-                    executor
-                        .execute(
-                            &self.engine,
-                            &component.id,
-                            channel,
-                            msg.get_payload_bytes(),
-                            follow,
-                        )
-                        .await?
-                }
-            };
+        if let Some(component_id) = self.channel_components.get(channel) {
+            tracing::trace!("Executing Redis component {component_id:?}");
+            let executor = SpinRedisExecutor;
+            executor
+                .execute(&self.engine, component_id, channel, msg.get_payload_bytes())
+                .await?
         } else {
             tracing::debug!("No subscription found for {:?}", channel);
         }
@@ -163,11 +121,10 @@ impl RedisTrigger {
 pub(crate) trait RedisExecutor: Clone + Send + Sync + 'static {
     async fn execute(
         &self,
-        engine: &ExecutionContext,
-        component: &str,
+        engine: &TriggerAppEngine<RedisTrigger>,
+        component_id: &str,
         channel: &str,
         payload: &[u8],
-        follow: bool,
     ) -> Result<()>;
 }
 

crates/redis/src/lib.rs

@@ -1,8 +1,9 @@
-use crate::{spin_redis::SpinRedis, ExecutionContext, RedisExecutor, RuntimeContext};
-use anyhow::Result;
+use anyhow::{anyhow, Result};
 use async_trait::async_trait;
-use spin_engine::io::ModuleIoRedirects;
-use wasmtime::{Instance, Store};
+use spin_core::Instance;
+use spin_trigger_new::TriggerAppEngine;
+
+use crate::{spin_redis::SpinRedis, RedisExecutor, RedisTrigger, Store};
 
 #[derive(Clone)]
 pub struct SpinRedisExecutor;
@@ -11,52 +12,40 @@ pub struct SpinRedisExecutor;
 impl RedisExecutor for SpinRedisExecutor {
     async fn execute(
         &self,
-        engine: &ExecutionContext,
-        component: &str,
+        engine: &TriggerAppEngine<RedisTrigger>,
+        component_id: &str,
         channel: &str,
         payload: &[u8],
-        follow: bool,
     ) -> Result<()> {
-        tracing::trace!(
-            "Executing request using the Spin executor for component {}",
-            component
-        );
-
-        let mior = ModuleIoRedirects::new(follow);
+        tracing::trace!("Executing request using the Spin executor for component {component_id}");
 
-        let (store, instance) = engine
-            .prepare_component(component, None, Some(mior.pipes), None, None)
-            .await?;
+        let (instance, store) = engine.prepare_instance(component_id).await?;
 
-        let result = match Self::execute_impl(store, instance, channel, payload.to_vec()).await {
+        match Self::execute_impl(store, instance, channel, payload.to_vec()).await {
             Ok(()) => {
                 tracing::trace!("Request finished OK");
                 Ok(())
             }
             Err(e) => {
-                tracing::trace!("Request finished with error {}", e);
+                tracing::trace!("Request finished with error {e}");
                 Err(e)
             }
-        };
-
-        let log_result =
-            engine.save_output_to_logs(mior.read_handles.read(), component, true, true);
-
-        result.and(log_result)
+        }
     }
 }
 
 impl SpinRedisExecutor {
     pub async fn execute_impl(
-        mut store: Store<RuntimeContext>,
+        mut store: Store,
         instance: Instance,
         _channel: &str,
         payload: Vec<u8>,
     ) -> Result<()> {
-        let engine = SpinRedis::new(&mut store, &instance, |host| host.data.as_mut().unwrap())?;
-
-        let _res = engine.handle_redis_message(&mut store, &payload).await;
+        let engine = SpinRedis::new(&mut store, &instance, |data| data.as_mut())?;
 
-        Ok(())
+        engine
+            .handle_redis_message(&mut store, &payload)
+            .await?
+            .map_err(|err| anyhow!("{err:?}"))
     }
 }

crates/redis/src/spin.rs

@@ -1,9 +1,7 @@
 use super::*;
 use anyhow::Result;
 use redis::{Msg, Value};
-use spin_manifest::{RedisConfig, RedisExecutor};
-use spin_testing::TestConfig;
-use spin_trigger::TriggerExecutorBuilder;
+use spin_testing::{tokio, TestConfig};
 
 fn create_trigger_event(channel: &str, payload: &str) -> redis::Msg {
     Msg::from_value(&redis::Value::Bulk(vec![
@@ -17,15 +15,11 @@ fn create_trigger_event(channel: &str, payload: &str) -> redis::Msg {
 #[ignore]
 #[tokio::test]
 async fn test_pubsub() -> Result<()> {
-    let mut cfg = TestConfig::default();
-    cfg.test_program("redis-rust.wasm")
-        .redis_trigger(RedisConfig {
-            channel: "messages".to_string(),
-            executor: Some(RedisExecutor::Spin),
-        });
-    let app = cfg.build_application();
-
-    let trigger: RedisTrigger = TriggerExecutorBuilder::new(app).build().await?;
+    let trigger: RedisTrigger = TestConfig::default()
+        .test_program("redis-rust.wasm")
+        .redis_trigger("messages")
+        .build_trigger()
+        .await;
 
     let msg = create_trigger_event("messages", "hello");
     trigger.handle(msg).await?;

crates/redis/src/tests.rs

@@ -12,9 +12,7 @@ serde = "1"
 serde_json = "1"
 spin-app = { path = "../app" }
 spin-core = { path = "../core" }
-spin-engine = { path = "../engine" }
-spin-manifest = { path = "../manifest" }
 spin-http = { path = "../http" }
-spin-trigger = { path = "../trigger" }
 spin-trigger-new = { path = "../trigger-new" }
+tokio = { version = "1", features = ["macros", "rt"] }
 tracing-subscriber = "0.3"