Better internal message switching (#213)

* Run cargo update + fix compiler warnings

* Minor improvments:

- Add todos that need to be processed
- Rename some structs
- Handle callback errors (drop the connection from worker connections subset)

* Add TwinID type

* Remove some un-necessary async_trait usage

* Worker rework

- Separate worker to its own strcture
- Use cancellation token to track if either of reader or writer routines exited
  and make sure connection is removed from global sessions set

* Increase connection channel size

* Use semaphore to track capacity

Also run cargo clippy to clean up all warnings

* WIP: Better worker loop

* Implement back pressure for slow clients

* Add some tests

* Fix formatting

Author: muhamadazmy

Cargo.lock

@@ -46,6 +46,7 @@ codec = { package = "parity-scale-codec", version = "3.0.0", default-features =
 ] }
 itertools = "0.11"
 ttl_cache = "0.5"
+tokio-util = "0.7"
 
 # for static build
 openssl = { version = "0.10", features = ["vendored"] }
@@ -73,6 +74,7 @@ lru = "0.9.0"
 rand = "0.8.5"
 tokio-stream = "0.1.14"
 async-stream = "0.3.5"
+derive_more = { version = "2.0.1", features = ["full"] }
 
 [build-dependencies]
 git-version = "0.3.5"

Cargo.toml

@@ -172,7 +172,7 @@ async fn app(args: Params) -> Result<()> {
 
         // we need to make sure our twin is up to date
         let twin = db
-            .get_twin(id)
+            .get_twin(id.into())
             .await
             .context("failed to get twin details")?
             .ok_or_else(|| anyhow::anyhow!("self twin not found!"))?;

src/bins/rmb-peer.rs

@@ -15,7 +15,6 @@ use tokio::sync::oneshot;
 
 /// A peer requires only which rely to connect to, and
 /// which identity (mnemonics)
-
 /// the reliable message bus
 #[derive(Parser, Debug)]
 #[clap(name ="rmb-rely", author, version = env!("GIT_VERSION"), about, long_about = None)]
@@ -175,8 +174,7 @@ async fn app(args: Args, tx: oneshot::Sender<()>) -> Result<()> {
         limiter,
         ranker,
     )
-    .await
-    .unwrap();
+    .context("failed to initialize relay")?;
 
     let mut l = events::Listener::new(args.substrate, redis_cache).await?;
     tokio::spawn(async move {

src/bins/rmb-relay.rs

@@ -5,7 +5,6 @@ use tokio::sync::RwLock;
 
 use super::Cache;
 use anyhow::Result;
-use async_trait::async_trait;
 use ttl_cache::TtlCache;
 
 static IN_MEMORY_CAP: usize = 500;
@@ -33,7 +32,6 @@ impl<V> MemCache<V> {
     }
 }
 
-#[async_trait]
 impl<T> Cache<T> for MemCache<T>
 where
     T: Clone + Send + Sync + 'static,

src/cache/memory.rs

@@ -4,17 +4,24 @@ pub use memory::MemCache;
 pub use redis::RedisCache;
 
 use anyhow::Result;
-use async_trait::async_trait;
-use std::marker::{Send, Sync};
+use std::{
+    future::Future,
+    marker::{Send, Sync},
+};
 
-#[async_trait]
 pub trait Cache<T>: Send + Sync + 'static {
-    async fn set<S: ToString + Send + Sync>(&self, id: S, obj: T) -> Result<()>;
-    async fn get<S: ToString + Send + Sync>(&self, id: S) -> Result<Option<T>>;
-    async fn flush(&self) -> Result<()>;
+    fn set<S: ToString + Send + Sync>(
+        &self,
+        id: S,
+        obj: T,
+    ) -> impl Future<Output = Result<()>> + Send;
+    fn get<S: ToString + Send + Sync>(
+        &self,
+        id: S,
+    ) -> impl Future<Output = Result<Option<T>>> + Send;
+    fn flush(&self) -> impl Future<Output = Result<()>> + Send;
 }
 
-#[async_trait]
 impl<T, C> Cache<T> for Option<C>
 where
     C: Cache<T>,
@@ -43,7 +50,6 @@ where
 #[derive(Clone, Copy)]
 pub struct NoCache;
 
-#[async_trait]
 impl<T> Cache<T> for NoCache
 where
     T: Send + Sync + 'static,

src/cache/mod.rs

@@ -1,7 +1,6 @@
 use super::Cache;
 
 use anyhow::{Context, Result};
-use async_trait::async_trait;
 use bb8_redis::{
     bb8::{Pool, PooledConnection},
     redis::cmd,
@@ -41,7 +40,6 @@ impl RedisCache {
     }
 }
 
-#[async_trait]
 impl<T> Cache<T> for RedisCache
 where
     T: Serialize + DeserializeOwned + Send + Sync + 'static,
@@ -53,7 +51,7 @@ where
             .arg(&self.prefix)
             .arg(key.to_string())
             .arg(obj)
-            .query_async(&mut *conn)
+            .query_async::<_, ()>(&mut *conn)
             .await?;
 
         Ok(())

src/cache/redis.rs

@@ -79,7 +79,7 @@ where
     }
 }
 
-impl<'a, S> SigningAlgorithm for JwtSigner<'a, S>
+impl<S> SigningAlgorithm for JwtSigner<'_, S>
 where
     S: Signer,
 {
@@ -108,7 +108,7 @@ where
     }
 }
 
-impl<'a, I> VerifyingAlgorithm for JwtVerifier<'a, I>
+impl<I> VerifyingAlgorithm for JwtVerifier<'_, I>
 where
     I: Identity,
 {

src/identity/mod.rs

@@ -164,7 +164,7 @@ where
         }
         let twin = self
             .twins
-            .get_twin(envelope.source.twin)
+            .get_twin(envelope.source.twin.into())
             .await
             .context("failed to get twin information")?
             .ok_or(ProtocolError::UnknownTwin(envelope.source.twin))?;
@@ -253,7 +253,7 @@ where
     pub async fn write(&mut self, mut envelope: Envelope) -> Result<(), ProtocolError> {
         let twin = self
             .twins
-            .get_twin(envelope.destination.twin)
+            .get_twin(envelope.destination.twin.into())
             .await?
             .ok_or_else(|| ProtocolError::UnknownTwin(envelope.destination.twin))?;
 

src/peer/protocol.rs

@@ -18,7 +18,7 @@ struct BacklogKey<'a>(&'a str);
 
 const MAX_COMMANDS: isize = 10000;
 
-impl<'a> ToRedisArgs for BacklogKey<'a> {
+impl ToRedisArgs for BacklogKey<'_> {
     fn write_redis_args<W>(&self, out: &mut W)
     where
         W: ?Sized + bb8_redis::redis::RedisWrite,
@@ -27,15 +27,15 @@ impl<'a> ToRedisArgs for BacklogKey<'a> {
     }
 }
 
-impl<'a> Display for BacklogKey<'a> {
+impl Display for BacklogKey<'_> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "msgbus.backlog.{}", self.0)
     }
 }
 
 struct RunKey<'a>(&'a str);
 
-impl<'a> ToRedisArgs for RunKey<'a> {
+impl ToRedisArgs for RunKey<'_> {
     fn write_redis_args<W>(&self, out: &mut W)
     where
         W: ?Sized + bb8_redis::redis::RedisWrite,
@@ -44,7 +44,7 @@ impl<'a> ToRedisArgs for RunKey<'a> {
     }
 }
 
-impl<'a> Display for RunKey<'a> {
+impl Display for RunKey<'_> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "msgbus.{}", self.0)
     }