redis-async-transaction.diff

diff --git a/autopush-common/src/redis.rs b/autopush-common/src/redis_util.rs
similarity index 52%
rename from autopush-common/src/redis.rs
rename to autopush-common/src/redis_util.rs
index 800a37d1..4add622d 100644
--- a/autopush-common/src/redis.rs
+++ b/autopush-common/src/redis_util.rs
@@ -1,11 +1,11 @@
-use redis::{cmd, pipe};
-use redis::{ConnectionLike, Pipeline, RedisResult, ToRedisArgs};
+use redis::{aio::ConnectionLike, Pipeline, RedisFuture, RedisResult, ToRedisArgs};
+use redis::{cmd, pipe, AsyncCommands};
 
-pub fn transaction<
-    C: ConnectionLike,
+pub async fn transaction<
+    C: ConnectionLike + AsyncCommands,
     K: ToRedisArgs,
     T,
-    F: FnMut(&mut C, &mut Pipeline) -> RedisResult<Option<T>>,
+    F: AsyncFnMut(&mut C, &mut Pipeline) -> RedisResult<Option<T>>,
 >(
     con: &mut C,
     keys: &[K],
@@ -13,9 +13,9 @@ pub fn transaction<
 ) -> RedisResult<T> {
     let mut func = func;
     loop {
-        cmd("WATCH").arg(keys).exec(con)?;
+        cmd("WATCH").arg(keys).exec_async(con).await?;
         let mut p = pipe();
-        let response: Option<T> = func(con, p.atomic())?;
+        let response: Option<T> = func(con, p.atomic()).await?;
         match response {
             None => {
                 continue;
@@ -23,7 +23,7 @@ pub fn transaction<
             Some(response) => {
                 // make sure no watch is left in the connection, even if
                 // someone forgot to use the pipeline.
-                cmd("UNWATCH").exec(con)?;
+                cmd("UNWATCH").exec_async(con).await?;
                 return Ok(response);
             }
         }