add general-purpose execute function to outbound-redis

crates/abi-conformance/src/outbound_redis.rs

@@ -1,5 +1,6 @@
 use super::Context;
 use anyhow::{ensure, Result};
+use outbound_redis::{Error, ValueParam, ValueResult};
 use serde::Serialize;
 use std::collections::{HashMap, HashSet};
 use wasmtime::{InstancePre, Store};
@@ -76,6 +77,15 @@ pub struct RedisReport {
     /// "baz"))` as the result.  The host will assert that said function is called exactly once with the specified
     /// arguments.
     pub smembers: Result<(), String>,
+
+    /// Result of the Redis `execute` test
+    ///
+    /// The guest module should expect a call according to [`super::InvocationStyle`] with
+    /// \["outbound-redis-execute", "127.0.0.1", "append", "foo", "baz"\] as arguments. The module should call the
+    /// host-implemented `outbound-redis::execute` function with the arguments \["127.0.0.1", "append", "foo",
+    /// "baz"\] and expect `ok(list(value::int(3)))` as the result.  The host will assert that said function is
+    /// called exactly once with the specified arguments.
+    pub execute: Result<(), String>,
 }
 
 wit_bindgen_wasmtime::export!("../../wit/ephemeral/outbound-redis.wit");
@@ -90,92 +100,93 @@ pub(super) struct OutboundRedis {
     sadd_map: HashMap<(String, String, Vec<String>), i64>,
     srem_map: HashMap<(String, String, Vec<String>), i64>,
     smembers_map: HashMap<(String, String), Vec<String>>,
+    execute_map: HashMap<(String, String, Vec<String>), Vec<ValueResult>>,
 }
 
 impl outbound_redis::OutboundRedis for OutboundRedis {
-    fn publish(
-        &mut self,
-        address: &str,
-        channel: &str,
-        payload: &[u8],
-    ) -> Result<(), outbound_redis::Error> {
+    fn publish(&mut self, address: &str, channel: &str, payload: &[u8]) -> Result<(), Error> {
         if self
             .publish_set
             .remove(&(address.to_owned(), channel.to_owned(), payload.to_vec()))
         {
             Ok(())
         } else {
-            Err(outbound_redis::Error::Error)
+            Err(Error::Error)
         }
     }
 
-    fn get(&mut self, address: &str, key: &str) -> Result<Vec<u8>, outbound_redis::Error> {
+    fn get(&mut self, address: &str, key: &str) -> Result<Vec<u8>, Error> {
         self.get_map
             .remove(&(address.to_owned(), key.to_owned()))
-            .ok_or(outbound_redis::Error::Error)
+            .ok_or(Error::Error)
     }
 
-    fn set(&mut self, address: &str, key: &str, value: &[u8]) -> Result<(), outbound_redis::Error> {
+    fn set(&mut self, address: &str, key: &str, value: &[u8]) -> Result<(), Error> {
         if self
             .set_set
             .remove(&(address.to_owned(), key.to_owned(), value.to_vec()))
         {
             Ok(())
         } else {
-            Err(outbound_redis::Error::Error)
+            Err(Error::Error)
         }
     }
 
-    fn incr(&mut self, address: &str, key: &str) -> Result<i64, outbound_redis::Error> {
+    fn incr(&mut self, address: &str, key: &str) -> Result<i64, Error> {
         self.incr_map
             .remove(&(address.to_owned(), key.to_owned()))
             .map(|value| value + 1)
-            .ok_or(outbound_redis::Error::Error)
+            .ok_or(Error::Error)
     }
 
-    fn del(&mut self, address: &str, keys: Vec<&str>) -> Result<i64, outbound_redis::Error> {
+    fn del(&mut self, address: &str, keys: Vec<&str>) -> Result<i64, Error> {
         self.del_map
             .remove(&(
                 address.into(),
                 keys.into_iter().map(|s| s.to_owned()).collect(),
             ))
-            .ok_or(outbound_redis::Error::Error)
+            .ok_or(Error::Error)
     }
 
-    fn sadd(
-        &mut self,
-        address: &str,
-        key: &str,
-        values: Vec<&str>,
-    ) -> Result<i64, outbound_redis::Error> {
+    fn sadd(&mut self, address: &str, key: &str, values: Vec<&str>) -> Result<i64, Error> {
         self.sadd_map
             .remove(&(
                 address.into(),
                 key.to_owned(),
                 values.into_iter().map(|s| s.to_owned()).collect(),
             ))
-            .ok_or(outbound_redis::Error::Error)
+            .ok_or(Error::Error)
     }
 
-    fn srem(
-        &mut self,
-        address: &str,
-        key: &str,
-        values: Vec<&str>,
-    ) -> Result<i64, outbound_redis::Error> {
+    fn srem(&mut self, address: &str, key: &str, values: Vec<&str>) -> Result<i64, Error> {
         self.srem_map
             .remove(&(
                 address.into(),
                 key.to_owned(),
                 values.into_iter().map(|s| s.to_owned()).collect(),
             ))
-            .ok_or(outbound_redis::Error::Error)
+            .ok_or(Error::Error)
     }
 
-    fn smembers(&mut self, address: &str, key: &str) -> Result<Vec<String>, outbound_redis::Error> {
+    fn smembers(&mut self, address: &str, key: &str) -> Result<Vec<String>, Error> {
         self.smembers_map
             .remove(&(address.into(), key.to_owned()))
-            .ok_or(outbound_redis::Error::Error)
+            .ok_or(Error::Error)
+    }
+
+    fn execute(
+        &mut self,
+        address: &str,
+        command: &str,
+        arguments: Vec<ValueParam<'_>>,
+    ) -> Result<Vec<ValueResult>, Error> {
+        self.execute_map
+            .remove(&(
+                address.into(),
+                command.to_owned(),
+                arguments.iter().map(|v| format!("{v:?}")).collect(),
+            ))
+            .ok_or(Error::Error)
     }
 }
 
@@ -360,5 +371,36 @@ pub(super) fn test(store: &mut Store<Context>, pre: &InstancePre<Context>) -> Re
                 },
             )
         },
+
+        execute: {
+            store.data_mut().outbound_redis.execute_map.insert(
+                (
+                    "127.0.0.1".into(),
+                    "append".to_owned(),
+                    vec!["foo".to_owned(), "baz".to_owned()],
+                ),
+                vec![ValueResult::Int(3)],
+            );
+
+            super::run_command(
+                store,
+                pre,
+                &[
+                    "outbound-redis-execute",
+                    "127.0.0.1",
+                    "append",
+                    "foo",
+                    "baz",
+                ],
+                |store| {
+                    ensure!(
+                        store.data().outbound_redis.execute_map.is_empty(),
+                        "expected module to call `outbound-redis::execute` exactly once"
+                    );
+
+                    Ok(())
+                },
+            )
+        },
     })
 }

crates/outbound-redis/src/lib.rs

@@ -3,13 +3,33 @@ mod host_component;
 use std::collections::{hash_map::Entry, HashMap};
 
 use anyhow::Result;
-use redis::{aio::Connection, AsyncCommands};
+use redis::{aio::Connection, AsyncCommands, FromRedisValue, RedisResult, Value};
 use wit_bindgen_wasmtime::async_trait;
 
 pub use host_component::OutboundRedisComponent;
 
 wit_bindgen_wasmtime::export!({paths: ["../../wit/ephemeral/outbound-redis.wit"], async: *});
-use outbound_redis::Error;
+use outbound_redis::{Error, ValueParam, ValueResult};
+
+struct Values(Vec<ValueResult>);
+
+impl FromRedisValue for Values {
+    fn from_redis_value(value: &Value) -> RedisResult<Self> {
+        fn append(values: &mut Vec<ValueResult>, value: &Value) {
+            match value {
+                Value::Nil | Value::Okay => (),
+                Value::Int(v) => values.push(ValueResult::Int(*v)),
+                Value::Data(bytes) => values.push(ValueResult::Data(bytes.to_owned())),
+                Value::Bulk(bulk) => bulk.iter().for_each(|value| append(values, value)),
+                Value::Status(message) => values.push(ValueResult::String(message.to_owned())),
+            }
+        }
+
+        let mut values = Vec::new();
+        append(&mut values, value);
+        Ok(Values(values))
+    }
+}
 
 #[derive(Default)]
 pub struct OutboundRedis {
@@ -65,6 +85,33 @@ impl outbound_redis::OutboundRedis for OutboundRedis {
         let value = conn.srem(key, values).await.map_err(log_error)?;
         Ok(value)
     }
+
+    async fn execute(
+        &mut self,
+        address: &str,
+        command: &str,
+        arguments: Vec<ValueParam<'_>>,
+    ) -> Result<Vec<ValueResult>, Error> {
+        let conn = self.get_conn(address).await.map_err(log_error)?;
+        let mut cmd = redis::cmd(command);
+        arguments.iter().for_each(|value| match value {
+            ValueParam::Nil => (),
+            ValueParam::String(s) => {
+                cmd.arg(s);
+            }
+            ValueParam::Int(v) => {
+                cmd.arg(v);
+            }
+            ValueParam::Data(v) => {
+                cmd.arg(v);
+            }
+        });
+
+        cmd.query_async::<_, Values>(conn)
+            .await
+            .map(|values| values.0)
+            .map_err(log_error)
+    }
 }
 
 impl OutboundRedis {

sdk/rust/src/lib.rs

@@ -43,8 +43,39 @@ pub mod http {
 /// Implementation of the spin redis interface.
 #[allow(missing_docs)]
 pub mod redis {
+    use std::hash::{Hash, Hasher};
+
     wit_bindgen_rust::import!("../../wit/ephemeral/outbound-redis.wit");
 
+    impl PartialEq for outbound_redis::ValueResult {
+        fn eq(&self, other: &Self) -> bool {
+            use outbound_redis::ValueResult::*;
+
+            match (self, other) {
+                (Nil, Nil) => true,
+                (String(a), String(b)) => a == b,
+                (Int(a), Int(b)) => a == b,
+                (Data(a), Data(b)) => a == b,
+                _ => false,
+            }
+        }
+    }
+
+    impl Eq for outbound_redis::ValueResult {}
+
+    impl Hash for outbound_redis::ValueResult {
+        fn hash<H: Hasher>(&self, state: &mut H) {
+            use outbound_redis::ValueResult::*;
+
+            match self {
+                Nil => (),
+                String(s) => s.hash(state),
+                Int(v) => v.hash(state),
+                Data(v) => v.hash(state),
+            }
+        }
+    }
+
     /// Exports the generated outbound Redis items.
     pub use outbound_redis::*;
 }

tests/integration.rs

@@ -9,7 +9,6 @@ mod integration_tests {
     use std::{
         collections::HashMap,
         ffi::OsStr,
-        fs,
         net::{Ipv4Addr, SocketAddrV4, TcpListener},
         path::Path,
         process::{self, Child, Command, Output},
@@ -1133,7 +1132,7 @@ route = "/..."
             ]
         });
         let manifest_file_path = dir.join("example-plugin-manifest.json");
-        fs::write(
+        std::fs::write(
             &manifest_file_path,
             serde_json::to_string(&plugin_manifest_json).unwrap(),
         )?;
@@ -1161,7 +1160,7 @@ route = "/..."
 
         // Upgrade plugin to newer version
         *plugin_manifest_json.get_mut("version").unwrap() = serde_json::json!("0.2.1");
-        fs::write(
+        std::fs::write(
             dir.join("example-plugin-manifest.json"),
             serde_json::to_string(&plugin_manifest_json).unwrap(),
         )?;
@@ -1184,7 +1183,7 @@ route = "/..."
             .join("plugins")
             .join("manifests")
             .join("example.json");
-        let manifest = fs::read_to_string(installed_manifest)?;
+        let manifest = std::fs::read_to_string(installed_manifest)?;
         assert!(manifest.contains("0.2.1"));
 
         // Uninstall plugin