Feat/comms behaviour protocol

.github/workflows/clippy.yml

@@ -19,7 +19,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        project: [engine, client, products/commandline, derive, utils, communication]
+        project: [engine, client, products/commandline, derive, utils, communication-refactored]
 
     steps:
       - uses: actions/checkout@v2

.github/workflows/format.yml

@@ -19,7 +19,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        project: [engine, client, products/commandline, derive, utils, communication]
+        project: [engine, client, products/commandline, derive, utils, communication-refactored]
 
     steps:
       - uses: actions/checkout@v2

.github/workflows/test.yml

@@ -19,7 +19,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        project: [engine, client, products/commandline, derive, utils, communication]
+        project: [engine, client, products/commandline, derive, utils, communication-refactored]
         os: [ubuntu-latest, macos-latest, windows-latest]
 
     steps:

Cargo.toml

@@ -3,7 +3,7 @@ members = [
   "engine",
   "engine/runtime",
   "client",
-  "communication",
+  "communication-refactored",
   "utils",
   "derive"
 ]

communication-refactored/.license_template

@@ -0,0 +1 @@
+../.license_template

communication-refactored/Cargo.toml

@@ -0,0 +1,39 @@
+[package]
+name = "stronghold-communication-refactored"
+version = "0.3.0"
+authors = [
+  "IOTA Stiftung",
+  "Elena Frank <[email protected]"
+]
+edition = "2018"
+license = "Apache-2.0"
+readme = "README.md"
+repository = "https://github.com/iotaledger/stronghold.rs"
+homepage = "https://stronghold.docs.iota.org"
+
+[lib]
+name = "communication_refactored"
+
+[dependencies]
+actix = { version = "0.12", optional = true }
+futures = "0.3"
+libp2p = { version = "0.39", default-features = false, features = ["noise", "yamux"] }
+serde = { version = "1.0", default-features = false, features = [ "alloc", "derive" ] }
+serde_json = { version = "1.0", default-features = false, features = [ "alloc" ] }
+smallvec = "1.6.1"
+stronghold-derive = { path = "../derive", version = "0.2.0" }
+tokio = { version = "1.7", default-features = false, features = ["rt", "sync"] }
+wasm-timer = "0.2.5"
+
+[features]
+default = [ "mdns", "relay", "tcp-transport"]
+mdns = ["libp2p/mdns"]
+relay = ["libp2p/relay"]
+tcp-transport = ["libp2p/tcp-tokio", "libp2p/dns-tokio", "libp2p/websocket"]
+actor = ["actix"]
+
+[dev-dependencies]
+rand = "0.8.3"
+tokio = {version = "1.7", features = ["time", "macros"]}
+libp2p = { version = "0.39", default-features = false, features = ["tcp-tokio"] }
+actix-rt = { version = "2.2"}

communication-refactored/README.md

@@ -0,0 +1 @@
+# Stronghold Communication

communication-refactored/src/actor.rs

@@ -0,0 +1,166 @@
+// Copyright 2020-2021 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+#![cfg(feature = "actor")]
+
+pub mod messages;
+use crate::{ListenErr, Multiaddr, OutboundFailure, PeerId, ReceiveRequest, RqRsMessage, ShCommunication};
+use actix::{dev::ToEnvelope, prelude::*};
+use futures::{channel::mpsc, FutureExt, TryFutureExt};
+use messages::*;
+use std::{borrow::Borrow, io, marker::PhantomData};
+
+#[macro_use]
+macro_rules! impl_handler {
+    ($mty:ty => $rty:ty, |$cid:ident, $mid:ident| $($body:stmt)+ ) => {
+        impl<ARegistry, C, Rq, Rs, TRq> Handler<$mty> for CommunicationActor<ARegistry, C, Rq, Rs, TRq>
+        where
+            ARegistry: ArbiterService + Handler<GetClient<Rq, C>>,
+            ARegistry::Context: ToEnvelope<ARegistry, GetClient<Rq, C>>,
+            C: Actor + Handler<Rq> + Send,
+            C::Context: ToEnvelope<C, Rq>,
+            Rq: Message<Result = Rs> + RqRsMessage + Borrow<TRq> + Clone,
+            Rs: RqRsMessage + Clone,
+            TRq: Clone + Send + 'static,
+        {
+            type Result = ResponseActFuture<Self, $rty>;
+            fn handle(&mut self, $mid: $mty, _: &mut Self::Context) -> Self::Result {
+                let mut $cid = self.comms.clone();
+                async move { $($body)+ }.into_actor(self).boxed_local()
+            }
+        }
+    };
+}
+
+pub struct GetClient<Rq: Message, C: Actor + Handler<Rq>> {
+    pub remote: PeerId,
+    _marker: (PhantomData<C>, PhantomData<Rq>),
+}
+
+impl<Rq: Message, C: Actor + Handler<Rq>> Message for GetClient<Rq, C> {
+    type Result = Addr<C>;
+}
+
+pub struct CommunicationActor<ARegistry, C, Rq, Rs, TRq = Rq>
+where
+    ARegistry: Actor,
+    Rq: Message + RqRsMessage + Borrow<TRq>,
+    Rs: RqRsMessage,
+    TRq: Clone + Send + 'static,
+{
+    comms: ShCommunication<Rq, Rs, TRq>,
+    inbound_request_rx: Option<mpsc::Receiver<ReceiveRequest<Rq, Rs>>>,
+    _marker: (PhantomData<ARegistry>, PhantomData<C>),
+}
+
+impl<ARegistry, C, Rq, Rs, TRq> CommunicationActor<ARegistry, C, Rq, Rs, TRq>
+where
+    ARegistry: Actor,
+    Rq: Message + RqRsMessage + Borrow<TRq>,
+    Rs: RqRsMessage,
+    TRq: Clone + Send + 'static,
+{
+    #[cfg(feature = "tcp-transport")]
+    pub async fn new() -> Result<Self, io::Error> {
+        let (firewall_tx, _) = mpsc::channel(0);
+        let (inbound_request_tx, inbound_request_rx) = mpsc::channel(1);
+        let comms = ShCommunication::new(firewall_tx, inbound_request_tx, None).await?;
+        let actor = Self {
+            comms,
+            inbound_request_rx: Some(inbound_request_rx),
+            _marker: (PhantomData, PhantomData),
+        };
+        Ok(actor)
+    }
+}
+
+impl<ARegistry, C, Rq, Rs, TRq> Actor for CommunicationActor<ARegistry, C, Rq, Rs, TRq>
+where
+    ARegistry: ArbiterService + Handler<GetClient<Rq, C>>,
+    ARegistry::Context: ToEnvelope<ARegistry, GetClient<Rq, C>>,
+    C: Actor + Handler<Rq> + Send,
+    C::Context: ToEnvelope<C, Rq>,
+    Rq: Message<Result = Rs> + RqRsMessage + Borrow<TRq>,
+    Rs: RqRsMessage,
+    TRq: Clone + Send + 'static,
+{
+    type Context = Context<Self>;
+
+    fn started(&mut self, ctx: &mut Self::Context) {
+        let inbound_request_rx = self.inbound_request_rx.take().unwrap();
+        Self::add_stream(inbound_request_rx, ctx);
+    }
+}
+
+impl<ARegistry, C, Rq, Rs, TRq> StreamHandler<ReceiveRequest<Rq, Rs>> for CommunicationActor<ARegistry, C, Rq, Rs, TRq>
+where
+    ARegistry: ArbiterService + Handler<GetClient<Rq, C>>,
+    ARegistry::Context: ToEnvelope<ARegistry, GetClient<Rq, C>>,
+    C: Actor + Handler<Rq> + Send,
+    C::Context: ToEnvelope<C, Rq>,
+    Rq: Message<Result = Rs> + RqRsMessage + Borrow<TRq>,
+    Rs: RqRsMessage,
+    TRq: Clone + Send + 'static,
+{
+    fn handle(&mut self, item: ReceiveRequest<Rq, Rs>, ctx: &mut Self::Context) {
+        let ReceiveRequest {
+            request,
+            response_tx,
+            peer,
+            ..
+        } = item;
+
+        let registry = ARegistry::from_registry();
+        let fut = registry
+            .send(GetClient {
+                remote: peer,
+                _marker: (PhantomData, PhantomData),
+            })
+            .and_then(|client| client.send(request))
+            .map_ok(|response| response_tx.send(response))
+            .map(|_| ())
+            .into_actor(self);
+        ctx.wait(fut);
+    }
+}
+
+impl<ARegistry, C, Rq, Rs, TRq> From<(ShCommunication<Rq, Rs, TRq>, mpsc::Receiver<ReceiveRequest<Rq, Rs>>)>
+    for CommunicationActor<ARegistry, C, Rq, Rs, TRq>
+where
+    ARegistry: Actor,
+    Rq: Message + RqRsMessage + Borrow<TRq>,
+    Rs: RqRsMessage,
+    TRq: Clone + Send + 'static,
+{
+    fn from((comms, request_rx): (ShCommunication<Rq, Rs, TRq>, mpsc::Receiver<ReceiveRequest<Rq, Rs>>)) -> Self {
+        Self {
+            comms,
+            inbound_request_rx: Some(request_rx),
+            _marker: (PhantomData, PhantomData),
+        }
+    }
+}
+
+impl_handler!(SendRequest<Rq, Rs> => Result<Rs, OutboundFailure>, |comms, msg| {
+    comms.send_request(msg.peer, msg.request).await
+});
+
+impl_handler!(StartListening => Result<Multiaddr, ListenErr>, |comms, msg| {
+     comms.start_listening(msg.address).await
+});
+
+impl_handler!(GetLocalPeerId => PeerId, |_comms, _msg| {
+    _comms.get_peer_id()
+});
+
+impl_handler!(AddPeerAddr => (), |comms, msg| {
+    comms.add_address(msg.peer, msg.address).await
+});
+
+impl_handler!(SetFirewallRule<TRq> => (), |comms, msg| {
+    comms.set_peer_rule(msg.peer, msg.direction, msg.rule).await
+});
+
+impl_handler!(SetFirewallDefault<TRq> => (), |comms, msg| {
+    comms.set_firewall_default(msg.direction, msg.rule).await
+});

communication-refactored/src/actor/messages.rs

@@ -0,0 +1,61 @@
+// Copyright 2020-2021 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+use std::marker::PhantomData;
+
+use crate::{
+    firewall::{Rule, RuleDirection},
+    ListenErr, OutboundFailure,
+};
+use actix::Message;
+use libp2p::{Multiaddr, PeerId};
+
+#[derive(Message)]
+#[rtype(result = "Result<Rs, OutboundFailure>")]
+pub struct SendRequest<Rq, Rs: 'static> {
+    pub peer: PeerId,
+    pub request: Rq,
+    _marker: PhantomData<Rs>,
+}
+
+impl<Rq, Rs: 'static> SendRequest<Rq, Rs> {
+    pub fn new(peer: PeerId, request: Rq) -> Self {
+        SendRequest {
+            peer,
+            request,
+            _marker: PhantomData,
+        }
+    }
+}
+
+#[derive(Message)]
+#[rtype(result = "Result<Multiaddr, ListenErr>")]
+pub struct StartListening {
+    pub address: Option<Multiaddr>,
+}
+
+#[derive(Message)]
+#[rtype(result = "PeerId")]
+pub struct GetLocalPeerId;
+
+#[derive(Message)]
+#[rtype(result = "()")]
+pub struct AddPeerAddr {
+    pub peer: PeerId,
+    pub address: Multiaddr,
+}
+
+#[derive(Message)]
+#[rtype(result = "()")]
+pub struct SetFirewallRule<TRq> {
+    pub peer: PeerId,
+    pub direction: RuleDirection,
+    pub rule: Rule<TRq>,
+}
+
+#[derive(Message)]
+#[rtype(result = "()")]
+pub struct SetFirewallDefault<TRq> {
+    pub direction: RuleDirection,
+    pub rule: Rule<TRq>,
+}