Develop (v0.4.1) (#54)

* feat: enhance error message remote receiver is not found * Create LICENSE * Create LICENSE * docs: add README badges * fix: repair dependencies * test: allow remote messages with generics * feat: include Message derive in RemoteMessage * feat: allow generic remote message through alias type + test * feat: add change_id feature to AnyAddr ! changes id for RemoteAddr and ignores for Addr ? solves #49 * feat(WIP): add protocols features - helps quickly implement several protocols * feat(WIP): add protocols features - add f32 support for reduce * feat: allow generic remote message and actor - use alias types in remote_messages macro * tests * fix: remote actor id in macro ! transform name to string ? used literally "#name" as actor id * test: reduce array to main node * release: telepathy-0.4.0 and telepathy-derive-0.3.0 * feat: wait_send method (#53) ? RemoteAddr::do_send happens later then Addr::do_send if they are directly followed ! Wait until RemoteAddr::do_send has written to FramedWrite * release: bump to version 0.4.1 Co-authored-by: wenig <[email protected]>
wenig · Jan 24, 2022 · b972ebe · b972ebe
1 parent dba9801
commit b972ebe
Show file tree

Hide file tree

Showing 7 changed files with 33 additions and 11 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "actix-telepathy"
-version = "0.4.0"
+version = "0.4.1"
 authors = ["wenig <[email protected]>"]
 edition = "2018"
 license = "Apache-2.0"

diff --git a/README.md b/README.md
@@ -1,7 +1,7 @@
 [![crates.io](https://img.shields.io/crates/v/actix-telepathy?label=latest)](https://crates.io/crates/actix-telepathy)
 ![Tests on main](https://github.com/wenig/actix-telepathy/workflows/Rust/badge.svg)
 [![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
-[![Dependency Status](https://deps.rs/crate/actix-telepathy/0.4.0/status.svg)](https://deps.rs/crate/actix-telepathy/0.4.0)
+[![Dependency Status](https://deps.rs/crate/actix-telepathy/0.4.1/status.svg)](https://deps.rs/crate/actix-telepathy/0.4.1)
 ![Downloads](https://img.shields.io/crates/d/actix-telepathy.svg)
 
 # Actix Telepathy
@@ -28,7 +28,7 @@ So far, we only support single seed nodes. Connecting to different seed nodes ca
 ```toml
 [dependencies]
 actix = "0.12"
-actix-telepathy = "0.4.0"
+actix-telepathy = "0.4.1"
 ```
 
 ### main.rs

diff --git a/src/codec.rs b/src/codec.rs
@@ -14,7 +14,8 @@ const ENDIAN_LENGTH: usize = 4;
 #[derive(Message, Deserialize, Serialize)]
 #[rtype(result = "()")]
 pub enum ClusterMessage {
-    Request(u16, bool), // bool = is_seed?
+    /// bool = is_seed?
+    Request(u16, bool),
     Response,
     Message(RemoteWrapper),
     Decline

diff --git a/src/network/mod.rs b/src/network/mod.rs
@@ -131,7 +131,7 @@ impl NetworkInterface {
     }
 
     fn transmit_message(&mut self, msg: ClusterMessage) {
-        &self.writer.as_ref().unwrap().do_send(msg);
+        self.writer.as_ref().unwrap().do_send(msg);
     }
 
     fn received_message(&mut self, mut msg: RemoteWrapper) {
@@ -189,6 +189,18 @@ impl Handler<ClusterMessage> for NetworkInterface {
     }
 }
 
+#[derive(Message)]
+#[rtype(result = "Result<(), MailboxError>")]
+pub struct WrappedClusterMessage(pub(crate) ClusterMessage);
+
+impl Handler<WrappedClusterMessage> for NetworkInterface {
+    type Result = ResponseFuture<Result<(), MailboxError>>;
+
+    fn handle(&mut self, msg: WrappedClusterMessage, _ctx: &mut Self::Context) -> Self::Result {
+        Box::pin(self.writer.as_ref().unwrap().send(msg.0))
+    }
+}
+
 impl WriteHandler<Error> for NetworkInterface {}
 impl Supervised for NetworkInterface {}
 

diff --git a/src/network/writer.rs b/src/network/writer.rs
@@ -18,7 +18,7 @@ impl Writer {
     }
 
     fn transmit_message(&mut self, msg: ClusterMessage) {
-        &self.framed[0].write(msg);
+        self.framed[0].write(msg);
     }
 }
 

diff --git a/src/remote/addr/mod.rs b/src/remote/addr/mod.rs
@@ -8,7 +8,7 @@ use serde::{Deserialize, Serialize};
 use crate::codec::ClusterMessage;
 use crate::remote::{AddrRepresentation, RemoteMessage, RemoteWrapper};
 use actix::dev::ToEnvelope;
-use crate::{NetworkInterface};
+use crate::{NetworkInterface, WrappedClusterMessage};
 
 pub mod resolver;
 #[cfg(test)]
@@ -56,9 +56,20 @@ impl RemoteAddr {
         ));
     }
 
-    pub fn send<T: RemoteMessage + Serialize>(&self, _msg: Box<T>) -> RecipientRequest<ClusterMessage> {
+    pub fn try_send<T: RemoteMessage + Serialize>(&self, _msg: Box<T>) -> RecipientRequest<ClusterMessage> {
+        unimplemented!("So far, it is not possible to use this method!")
+    }
+
+    pub fn send<T: RemoteMessage + Serialize>(&self, _msg: Box<T>) -> () {
         unimplemented!("So far, it is not possible to receive responses from remote destinations as futures!")
     }
+
+    pub fn wait_send<T: RemoteMessage + Serialize>(&self, msg: T) -> Request<NetworkInterface, WrappedClusterMessage> {
+        self.network_interface.as_ref().expect("Network interface must be set!")
+            .send(WrappedClusterMessage(ClusterMessage::Message(
+            RemoteWrapper::new(self.clone(), msg, None)
+        )))
+    }
 }
 
 impl Clone for RemoteAddr {

diff --git a/src/utils/custom_system_service.rs b/src/utils/custom_system_service.rs
@@ -11,6 +11,7 @@ static SREG: Lazy<Mutex<HashMap<usize, PatchedSystemRegistry>>> =
 
 #[derive(Debug)]
 struct PatchedSystemRegistry {
+    #[allow(dead_code)]
     system: ArbiterHandle,
     registry: HashMap<TypeId, Box<dyn Any + Send>>,
 }
@@ -70,6 +71,3 @@ pub trait CustomSystemService: Actor<Context = Context<Self>> + SystemService {
         panic!("Please start Actor before asking for it in registry!");
     }
 }
-
-
-
-Original file line number
+Diff line change
@@ Expand Up / @@ -18,7 +18,7 @@ impl Writer { @@
         }
         fn transmit_message(&mut self, msg: ClusterMessage) {
-            &self.framed[0].write(msg);
+            self.framed[0].write(msg);
         }
     }
@@ Expand Down @@