diff --git a/crates/trigger-redis/src/lib.rs b/crates/trigger-redis/src/lib.rs index 2101ae59b2..08b2a092df 100644 --- a/crates/trigger-redis/src/lib.rs +++ b/crates/trigger-redis/src/lib.rs @@ -7,7 +7,8 @@ use serde::Deserialize; use spin_factor_variables::VariablesFactor; use spin_factors::RuntimeFactors; use spin_trigger::{cli::NoCliArgs, App, Trigger, TriggerApp}; -use spin_world::exports::fermyon::spin::inbound_redis; +use spin_world::exports::fermyon::spin::inbound_redis as v1; +use spin_world::exports::spin::redis::inbound_redis as v3; use tracing::{instrument, Level}; pub struct RedisTrigger; @@ -211,14 +212,60 @@ impl Subscriber { .await?; let pre = instance.instance_pre(&store); - let guest_indices = inbound_redis::GuestIndices::new(&pre)?; - let guest = guest_indices.load(&mut store, &instance)?; - let payload = msg.get_payload_bytes().to_vec(); + match HandlerType::from_instance_pre(&pre)? { + HandlerType::V1(guest_indices) => { + let guest = guest_indices.load(&mut store, &instance)?; - guest - .call_handle_message(&mut store, &payload) - .await? - .context("Redis handler returned an error") + let payload = msg.get_payload_bytes().to_vec(); + + guest + .call_handle_message(&mut store, &payload) + .await? + .context("Redis handler returned an error") + } + HandlerType::V3(guest_indices) => { + let guest = guest_indices.load(&mut store, &instance)?; + + let payload = msg.get_payload_bytes().to_vec(); + let res = std::pin::pin!(store.as_mut().run_concurrent(async |accessor| { + guest.call_handle_message(accessor, payload).await + })) + .await; + + res.map_err(|e| anyhow::anyhow!("{e}")) + .context("Redis handler returned an error (run_concurrent)")? + .map_err(|e| anyhow::anyhow!("{e}")) + .context("Redis handler returned an error")? + .context("Redis handler returned an error") + } + } + } +} + +/// The type of Redis handler export used by a component. +pub enum HandlerType /**/ { + V1(v1::GuestIndices), + V3(v3::GuestIndices), +} + +impl HandlerType { + /// Determine the handler type from the exports of a component. + pub fn from_instance_pre( + pre: &spin_factors::wasmtime::component::InstancePre, /*, handler_state: S*/ + ) -> anyhow::Result { + let mut candidates = Vec::new(); + if let Ok(indices) = v1::GuestIndices::new(pre) { + candidates.push(HandlerType::V1(indices)); + } + if let Ok(indices) = v3::GuestIndices::new(pre) { + candidates.push(HandlerType::V3(indices)); + } + + match candidates.len() { + 0 => anyhow::bail!("component does not export a Redis interface"), + 1 => Ok(candidates.pop().unwrap()), + _ => anyhow::bail!("component exports multiple Redis interfaces"), + } } } diff --git a/crates/world/src/lib.rs b/crates/world/src/lib.rs index 284b7601f4..7f1ad31473 100644 --- a/crates/world/src/lib.rs +++ b/crates/world/src/lib.rs @@ -14,6 +14,7 @@ wasmtime::component::bindgen!({ include spin:up/platform@3.4.0; include spin:up/platform@3.7.0; include wasi:keyvalue/imports@0.2.0-draft2; + export spin:redis/inbound-redis@3.0.0; } "#, path: "../../wit", diff --git a/tests/integration.rs b/tests/integration.rs index 8780912eba..48655be32d 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -214,6 +214,64 @@ mod integration_tests { Ok(()) } + #[test] + #[cfg(feature = "extern-dependencies-tests")] + #[allow(dependency_on_unit_never_type_fallback)] + /// Test that basic redis trigger support works + fn redis_async_trigger_smoke_test() -> anyhow::Result<()> { + use anyhow::Context; + use redis::Commands; + run_test( + "redis-async-trigger-smoke-test", + SpinConfig { + binary_path: spin_binary(), + spin_up_args: Vec::new(), + app_type: SpinAppType::Redis, + }, + ServicesConfig::new(vec!["redis"])?, + move |env| { + let redis_port = env + .services_mut() + .get_port(6379)? + .context("no redis port was exposed by test services")?; + + let mut redis = redis::Client::open(format!("redis://localhost:{redis_port}")) + .context("could not connect to redis in test")?; + redis + .publish::<_, _, ()>("my-channel", "msg-from-test") + .context("could not publish test message to redis")?; + assert_eventually!( + { + match env.read_file(".spin/logs/hello_stdout.txt") { + Ok(logs) => { + let logs = String::from_utf8_lossy(&logs); + logs.contains("Got message: 'msg-from-test'") + } + Err(e) + if e.downcast_ref() + .map(|e: &std::io::Error| { + e.kind() == std::io::ErrorKind::NotFound + }) + .unwrap_or_default() => + { + false + } + Err(e) => { + return Err( + anyhow::anyhow!("could not read stdout file: {e}").into() + ); + } + } + }, + 2 + ); + Ok(()) + }, + )?; + + Ok(()) + } + #[test] #[cfg(feature = "extern-dependencies-tests")] /// Test that basic otel tracing works diff --git a/tests/test-components/components/Cargo.lock b/tests/test-components/components/Cargo.lock index ddea96a7a7..aabf7cab59 100644 --- a/tests/test-components/components/Cargo.lock +++ b/tests/test-components/components/Cargo.lock @@ -1041,6 +1041,15 @@ dependencies = [ "getrandom", ] +[[package]] +name = "redis-async-trigger-smoke-test" +version = "0.1.0" +dependencies = [ + "anyhow", + "bytes", + "wit-bindgen 0.54.0", +] + [[package]] name = "routefinder" version = "0.5.4" diff --git a/tests/test-components/components/redis-async-trigger-smoke-test/Cargo.toml b/tests/test-components/components/redis-async-trigger-smoke-test/Cargo.toml new file mode 100644 index 0000000000..bd59cb5fbf --- /dev/null +++ b/tests/test-components/components/redis-async-trigger-smoke-test/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "redis-async-trigger-smoke-test" +version = "0.1.0" +edition = "2021" + +[lib] +crate-type = ["cdylib"] + +[dependencies] +anyhow = "1" +bytes = "1" +wit-bindgen = "0.54" diff --git a/tests/test-components/components/redis-async-trigger-smoke-test/src/lib.rs b/tests/test-components/components/redis-async-trigger-smoke-test/src/lib.rs new file mode 100644 index 0000000000..d6e1574c54 --- /dev/null +++ b/tests/test-components/components/redis-async-trigger-smoke-test/src/lib.rs @@ -0,0 +1,25 @@ +use std::str::from_utf8; + +const KV_KEY: &str = "message"; + +wit_bindgen::generate!({ + path: "../../../../wit", + world: "spin:up/redis-trigger@3.7.0", + generate_all, +}); + +struct Guest; + +impl exports::spin::redis::inbound_redis::Guest for Guest { + async fn handle_message(message: Vec) -> Result<(), spin::redis::redis::Error> { + // Do some async stuff to prove it works + let kv = spin::key_value::key_value::Store::open("default".to_string()).await.expect("should have had access to default KV store"); + kv.set(KV_KEY.to_string(), message.clone()).await.expect("should have set KV entry"); + let message = kv.get(KV_KEY.to_string()).await.expect("should have read KV entry").expect("KV entry should have existed"); + + println!("Got message: '{}'", from_utf8(&message).unwrap_or("")); + Ok(()) + } +} + +export!(Guest); diff --git a/tests/testcases/redis-async-trigger-smoke-test/spin.toml b/tests/testcases/redis-async-trigger-smoke-test/spin.toml new file mode 100644 index 0000000000..9abc0867d5 --- /dev/null +++ b/tests/testcases/redis-async-trigger-smoke-test/spin.toml @@ -0,0 +1,20 @@ +spin_manifest_version = 2 + +[application] +authors = ["Spin Framework Contributors"] +description = "A simple redis application that exercises the Rust SDK in the current branch" +name = "redis-async-trigger-smoke-test" +version = "1.0.0" + +[application.trigger.redis] +address = "redis://localhost:%{port=6379}" + +[[trigger.redis]] +channel = "my-channel" +component = "hello" + +[component.hello] +source = "%{source=redis-async-trigger-smoke-test}" +key_value_stores = ["default"] +[component.hello.build] +command = "cargo build --target wasm32-wasip2 --release" diff --git a/wit/deps/spin-redis@3.0.0/redis.wit b/wit/deps/spin-redis@3.0.0/redis.wit index f162535e8c..83f8e92e0c 100644 --- a/wit/deps/spin-redis@3.0.0/redis.wit +++ b/wit/deps/spin-redis@3.0.0/redis.wit @@ -70,3 +70,10 @@ interface redis { binary(payload) } } + +interface inbound-redis { + use redis.{payload, error}; + + // The entrypoint for a Redis handler. + handle-message: async func(message: payload) -> result<_, error>; +} diff --git a/wit/world.wit b/wit/world.wit index 9b67f793ff..520d0392ef 100644 --- a/wit/world.wit +++ b/wit/world.wit @@ -6,6 +6,12 @@ world http-trigger { export wasi:http/handler@0.3.0-rc-2026-03-15; } +/// The full world of a guest targeting a redis-trigger +world redis-trigger { + include platform; + export spin:redis/inbound-redis@3.0.0; +} + /// The imports needed for a guest to run on a Spin host world platform { include wasi:cli/imports@0.2.6;