Support multiple relay domains (#217)

* fix dependency_on_unit_never_type_fallback compiler warnings

* feat: support_multiple_relay_domains

* Update readme

* Update readme

Author: sameh-farouk

docs/readme.md

@@ -272,3 +272,19 @@ Example:
 ```bash
   rmb-peer -m "{MNEMONIC}" --substrate wss://tfchain.dev.grid.tf:443 --relay wss://r1.dev.grid.tf --relay wss://r2.dev.grid.tf
 ```
+
+## Relay Domains
+
+Starting from v1.3.1, the relay supports multiple domains (when needed), which are used to determine whether a destination twin is directly connected to this relay. these domains should be points to the public IP of this relay.
+
+```bash
+rmb-relay --domain relay.example.com --domain relay1.example.com
+```
+
+or short name
+
+```bash
+rmb-relay -m relay.example.com -m relay1.example.com
+```
+
+The relay will internally route messages over existing WebSocket connections to twins that use one of the provided domains as their relay domain. If the destination twin isn't associated with any of these domains, the message will be forwarded to one of the twin's relays via federation (https protocol).

src/bins/rmb-relay.rs

@@ -20,10 +20,10 @@ use tokio::sync::oneshot;
 #[derive(Parser, Debug)]
 #[clap(name ="rmb-rely", author, version = env!("GIT_VERSION"), about, long_about = None)]
 struct Args {
-    /// domain of this relay or it's public IP. used to identify
+    /// domains of this relay or it's public IPs. used to identify
     /// if a twin is on this relay or not.
-    #[clap(short = 'm', long)]
-    domain: String,
+    #[clap(short = 'm', long = "domain", num_args = 1..)]
+    domains: Vec<String>,
 
     /// redis address
     #[clap(short, long, default_value_t = String::from("redis://localhost:6379"))]
@@ -167,9 +167,16 @@ async fn app(args: Args, tx: oneshot::Sender<()>) -> Result<()> {
         )
     };
     let ranker = relay::ranker::RelayRanker::new(Duration::from_secs(args.ranker_period));
-    let r = relay::Relay::new(&args.domain, twins, opt, federation, limiter, ranker)
-        .await
-        .unwrap();
+    let r = relay::Relay::new(
+        args.domains.iter().cloned().collect(),
+        twins,
+        opt,
+        federation,
+        limiter,
+        ranker,
+    )
+    .await
+    .unwrap();
 
     let mut l = events::Listener::new(args.substrate, redis_cache).await?;
     tokio::spawn(async move {

src/cache/redis.rs

@@ -49,7 +49,7 @@ where
     async fn set<S: ToString + Send + Sync>(&self, key: S, obj: T) -> Result<()> {
         let mut conn = self.get_connection().await?;
         let obj = serde_json::to_vec(&obj).context("unable to serialize twin object for redis")?;
-        cmd("HSET")
+        let _: () = cmd("HSET")
             .arg(&self.prefix)
             .arg(key.to_string())
             .arg(obj)
@@ -79,7 +79,7 @@ where
     }
     async fn flush(&self) -> Result<()> {
         let mut conn = self.get_connection().await?;
-        cmd("DEL").arg(&self.prefix).query_async(&mut *conn).await?;
+        let _: () = cmd("DEL").arg(&self.prefix).query_async(&mut *conn).await?;
 
         Ok(())
     }

src/peer/storage/redis_storage.rs

@@ -152,7 +152,7 @@ impl Storage for RedisStorage {
         }
         let mut conn = self.get_connection().await?;
         let key = BacklogKey(&backlog.uid);
-        conn.set_ex(&key, backlog, backlog.ttl as usize).await?;
+        let _: () = conn.set_ex(&key, backlog, backlog.ttl as usize).await?;
 
         Ok(())
     }
@@ -167,8 +167,8 @@ impl Storage for RedisStorage {
         request.reply_to = Queue::Response.to_string();
 
         let key = RunKey(&request.command);
-        conn.lpush(&key, &request).await?;
-        conn.ltrim(&key, 0, self.max_commands - 1).await?;
+        let _: () = conn.lpush(&key, &request).await?;
+        let _: () = conn.ltrim(&key, 0, self.max_commands - 1).await?;
 
         Ok(())
     }
@@ -177,8 +177,8 @@ impl Storage for RedisStorage {
         let mut conn = self.get_connection().await?;
         // set reply queue
 
-        conn.lpush(queue, &response).await?;
-        conn.ltrim(queue, 0, self.max_commands - 1).await?;
+        let _: () = conn.lpush(queue, &response).await?;
+        let _: () = conn.ltrim(queue, 0, self.max_commands - 1).await?;
 
         Ok(())
     }

src/relay/api.rs

@@ -15,6 +15,7 @@ use hyper_tungstenite::{HyperWebsocket, WebSocketStream};
 use prometheus::Encoder;
 use prometheus::TextEncoder;
 use protobuf::Message as ProtoMessage;
+use std::collections::HashSet;
 use std::fmt::Display;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -28,7 +29,7 @@ use super::HttpError;
 pub(crate) struct AppData<D: TwinDB, R: RateLimiter> {
     switch: Arc<Switch<RelayHook>>,
     twins: D,
-    domain: String,
+    domains: HashSet<String>,
     federator: Arc<Federator>,
     limiter: R,
 }
@@ -38,15 +39,15 @@ where
     D: TwinDB,
     R: RateLimiter,
 {
-    pub(crate) fn new<S: Into<String>>(
-        domain: S,
+    pub(crate) fn new(
+        domains: HashSet<String>,
         switch: Arc<Switch<RelayHook>>,
         twins: D,
         federator: Federator,
         limiter: R,
     ) -> Self {
         Self {
-            domain: domain.into(),
+            domains: domains,
             switch,
             twins,
             federator: Arc::new(federator),
@@ -134,7 +135,7 @@ async fn entry<D: TwinDB, R: RateLimiter>(
         let stream = Stream::new(
             // todo: improve the domain clone
             claims,
-            data.domain.clone(),
+            data.domains.clone(),
             Arc::clone(&data.switch),
             Arc::clone(&data.federator),
             metrics,
@@ -275,7 +276,7 @@ impl Hook for RelayHook {
 
 struct Stream<M: Metrics, D: TwinDB> {
     id: StreamID,
-    domain: String,
+    domains: HashSet<String>,
     switch: Arc<Switch<RelayHook>>,
     federator: Arc<Federator>,
     metrics: M,
@@ -284,7 +285,7 @@ struct Stream<M: Metrics, D: TwinDB> {
 impl<M: Metrics, D: TwinDB> Stream<M, D> {
     fn new(
         claims: Claims,
-        domain: String,
+        domains: HashSet<String>,
         switch: Arc<Switch<RelayHook>>,
         federator: Arc<Federator>,
         metrics: M,
@@ -293,7 +294,7 @@ impl<M: Metrics, D: TwinDB> Stream<M, D> {
         let id: StreamID = (claims.id, claims.sid).into();
         Self {
             id,
-            domain,
+            domains,
             switch,
             federator,
             metrics,
@@ -338,7 +339,7 @@ impl<M: Metrics, D: TwinDB> Stream<M, D> {
         if !twin
             .relay
             .ok_or_else(|| anyhow::Error::msg("relay info is not set for this twin"))?
-            .contains(&self.domain)
+            .has_common(&self.domains)
         {
             log::debug!("got an foreign message");
             // push message to the (relay.federation) queue

src/relay/mod.rs

@@ -15,6 +15,7 @@ use self::ranker::RelayRanker;
 use api::RelayHook;
 use federation::Federation;
 pub use federation::FederationOptions;
+use std::collections::HashSet;
 use std::sync::Arc;
 use switch::Switch;
 pub use switch::SwitchOptions;
@@ -23,7 +24,7 @@ pub mod ranker;
 pub struct Relay<D: TwinDB, R: RateLimiter> {
     switch: Arc<Switch<RelayHook>>,
     twins: D,
-    domain: String,
+    domains: HashSet<String>,
     federation: Federation<D>,
     limiter: R,
 }
@@ -33,8 +34,8 @@ where
     D: TwinDB + Clone,
     R: RateLimiter,
 {
-    pub async fn new<S: Into<String>>(
-        domain: S,
+    pub async fn new(
+        domains: HashSet<String>,
         twins: D,
         opt: SwitchOptions,
         federation: FederationOptions<D>,
@@ -46,7 +47,7 @@ where
         Ok(Self {
             switch: Arc::new(switch),
             twins,
-            domain: domain.into(),
+            domains: domains,
             federation,
             limiter,
         })
@@ -56,7 +57,7 @@ where
         let tcp_listener = TcpListener::bind(address).await?;
         let federator = self.federation.start();
         let http = api::HttpService::new(api::AppData::new(
-            self.domain,
+            self.domains,
             self.switch,
             self.twins,
             federator,

src/relay/switch/mod.rs

@@ -436,7 +436,7 @@ where
             c = c.arg(id);
         }
 
-        c.query_async(&mut *con).await?;
+        let _: () = c.query_async(&mut *con).await?;
 
         Ok(())
     }
@@ -493,7 +493,7 @@ async fn send<'a>(
         .query_async(&mut *con)
         .await?;
 
-    cmd("EXPIRE")
+    let _: () = cmd("EXPIRE")
         .arg(stream_id)
         .arg(QUEUE_EXPIRE)
         .query_async(&mut *con)

src/twin/mod.rs

@@ -97,4 +97,8 @@ impl RelayDomains {
     pub fn iter(&self) -> std::collections::hash_set::Iter<String> {
         self.0.iter()
     }
+
+    pub fn has_common(&self, other: &HashSet<String>) -> bool {
+        !self.0.is_disjoint(other)
+    }
 }