Add Redis broker

brokers/Cargo.toml

@@ -11,10 +11,15 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-lapin = "1.1"
+deadpool-redis = { version = "0.9", optional = true }
+lapin = { version = "1.1", optional = true }
 env_logger = "0.7"
+futures = "0.3"
 log = "0.4"
 nanoid = "0.3"
+redis = { version = "0.21", optional = true, default-features = false, features = ["streams"] }
+rmp-serde = "0.15"
+serde = "1.0"
 thiserror = "1.0"
 tokio-stream = "0.1"
 
@@ -25,3 +30,23 @@ features = ["rt-multi-thread"]
 [dev-dependencies.tokio]
 version = "1.0"
 features = ["rt-multi-thread", "macros"]
+
+[features]
+amqp-broker = ["lapin"]
+redis-broker = ["deadpool-redis", "redis"]
+
+[[example]]
+name = "amqp_consumer"
+required-features = ["amqp-broker"]
+
+[[example]]
+name = "amqp_producer"
+required-features = ["amqp-broker"]
+
+[[example]]
+name = "amqp_rpc_consumer"
+required-features = ["amqp-broker"]
+
+[[example]]
+name = "amqp_rpc_producer"
+required-features = ["amqp-broker"]

brokers/src/error.rs

@@ -1,13 +1,16 @@
+#[cfg(feature = "redis-broker")]
+use deadpool_redis::{redis::RedisError, PoolError};
+#[cfg(feature = "amqp-broker")]
 use lapin::Error as LapinError;
-
 use std::{io::Error as IoError, result::Result as StdResult};
 use thiserror::Error;
-use tokio::sync::oneshot::error::RecvError;
+use tokio::{sync::oneshot::error::RecvError, task::JoinError};
 
-pub type Result<T> = StdResult<T, Error>;
+pub type Result<T, E = Error> = StdResult<T, E>;
 
 #[derive(Error, Debug)]
 pub enum Error {
+    #[cfg(feature = "amqp-broker")]
     #[error("Lapin error")]
     Lapin(#[from] LapinError),
     #[error("IO error")]
@@ -16,4 +19,16 @@ pub enum Error {
     Recv(#[from] RecvError),
     #[error("Reply error")]
     Reply(String),
+    #[cfg(feature = "redis-broker")]
+    #[error("Redis error")]
+    Redis(#[from] RedisError),
+    #[cfg(feature = "redis-broker")]
+    #[error("Pool error")]
+    Deadpool(#[from] PoolError),
+    #[error("Join error")]
+    Join(#[from] JoinError),
+    #[error("MessagePack encode error")]
+    MsgpackEncode(#[from] rmp_serde::encode::Error),
+    #[error("MessagePack decode error")]
+    MsgpackDecode(#[from] rmp_serde::decode::Error),
 }

brokers/src/lib.rs

@@ -1,2 +1,6 @@
+#[cfg(feature = "amqp-broker")]
 pub mod amqp;
 pub mod error;
+#[cfg(feature = "redis-broker")]
+pub mod redis;
+mod util;