Send membership

src/node.rs

@@ -226,9 +226,27 @@ impl Node {
                 .await;
 
             log::info!("Echo broadcast finished");
+            let _ = self.send_membership(reliable_sender_handle).await;
+
+            log::info!("Membership sent");
         }
     }
 
+    pub(crate) async fn send_membership(&self, sender: ReliableSenderHandle) {
+        log::info!("Sending membership information");
+        let protocol_service =
+            protocol::Protocol::new(self.get_node_id().clone(), self.state.clone());
+        let reliable_sender_service = ReliableSend::new(protocol_service, sender);
+        let timeout_layer = tower::timeout::TimeoutLayer::new(tokio::time::Duration::from_millis(
+            self.delivery_timeout,
+        ));
+        let res = timeout_layer
+            .layer(reliable_sender_service)
+            .oneshot(MembershipMessage::new(self.get_node_id().clone(), None).into())
+            .await;
+        log::debug!("Membership sending result {:?}", res);
+    }
+
     /// Connect to all peers and start reader writer tasks
     pub async fn connect_to_seeds(&mut self) -> Result<(), Box<dyn Error>> {
         log::debug!("Connecting to seeds...");

src/node/protocol.rs

@@ -168,7 +168,7 @@ impl Service<Message> for Protocol {
                     BoxService::new(Heartbeat::new(sender_id))
                 }
                 Message::UnicastMessage(Unicast::Membership(_m)) => {
-                    BoxService::new(Membership::new(sender_id))
+                    BoxService::new(Membership::new(sender_id, state))
                 }
                 Message::BroadcastMessage(Broadcast::RoundOnePackage(_m, _)) => {
                     BoxService::new(RoundOnePackage::new(sender_id, state))

src/node/protocol/membership.rs

@@ -16,36 +16,41 @@
 // along with Frost-Federation. If not, see
 // <https://www.gnu.org/licenses/>.
 
-use crate::node::protocol::Message;
+use crate::node::protocol::{Message, Unicast};
+use crate::node::state::State;
 use futures::{Future, FutureExt};
 use serde::{Deserialize, Serialize};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use tower::{BoxError, Service};
 
-#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, Default)]
+#[derive(Debug, Serialize, Deserialize, PartialEq, Clone)]
 pub struct MembershipMessage {
     pub sender_id: String,
-    pub message: Vec<String>,
+    pub message: Option<Vec<String>>,
 }
 
 impl MembershipMessage {
-    pub fn new(sender_id: String, members: Vec<String>) -> Self {
+    pub fn new(sender_id: String, members: Option<Vec<String>>) -> Self {
         MembershipMessage {
             sender_id,
             message: members,
         }
     }
 }
 
-#[derive(Debug, Clone, Default)]
+#[derive(Clone)]
 pub struct Membership {
     sender_id: String,
+    state: State,
 }
 
 impl Membership {
-    pub fn new(node_id: String) -> Self {
-        Membership { sender_id: node_id }
+    pub fn new(node_id: String, state: State) -> Self {
+        Membership {
+            sender_id: node_id,
+            state,
+        }
     }
 }
 
@@ -61,50 +66,98 @@ impl Service<Message> for Membership {
 
     /// Membership doesn't respond with anything. It is pushed by a
     /// node when anyone connects to it.
-    fn call(&mut self, _msg: Message) -> Self::Future {
-        async move { Ok(None) }.boxed()
+    fn call(&mut self, msg: Message) -> Self::Future {
+        let state = self.state.clone();
+        async move {
+            match msg {
+                Message::UnicastMessage(Unicast::Membership(MembershipMessage {
+                    message,
+                    sender_id,
+                })) => match message {
+                    None => {
+                        let members = state
+                            .membership_handle
+                            .get_members()
+                            .await
+                            .unwrap()
+                            .into_keys()
+                            .collect();
+                        Ok(Some(
+                            MembershipMessage {
+                                sender_id,
+                                message: Some(members),
+                            }
+                            .into(),
+                        ))
+                    }
+                    _ => Ok(None),
+                },
+                _ => Ok(None),
+            }
+        }
+        .boxed()
     }
 }
 
 #[cfg(test)]
 mod membership_tests {
 
     use super::Membership;
+    use crate::node::protocol::message_id_generator::MessageIdGenerator;
     use crate::node::protocol::MembershipMessage;
+    #[mockall_double::double]
+    use crate::node::reliable_sender::ReliableSenderHandle;
+    use crate::node::MembershipHandle;
+    use crate::node::State;
     use tower::{Service, ServiceExt};
 
     #[tokio::test]
-    async fn it_should_create_membership_as_service_and_respond_to_none_with_membership() {
-        let mut p = Membership::new("local".to_string());
+    async fn it_should_create_membership_as_service_and_respond_to_default_with_membership() {
+        let message_id_generator = MessageIdGenerator::new("localhost".to_string());
+        let membership_handle = MembershipHandle::start("localhost".to_string()).await;
+        let mut reliable_sender_handle = ReliableSenderHandle::default();
+        reliable_sender_handle
+            .expect_clone()
+            .returning(ReliableSenderHandle::default);
+        let _ = membership_handle
+            .add_member("a".into(), reliable_sender_handle)
+            .await;
+        let state = State::new(membership_handle, message_id_generator);
+
+        let mut p = Membership::new("local".to_string(), state);
         let res = p
             .ready()
             .await
             .unwrap()
-            .call(MembershipMessage::default().into())
+            .call(
+                MembershipMessage {
+                    sender_id: "local".into(),
+                    message: None,
+                }
+                .into(),
+            )
             .await
             .unwrap();
-        assert!(res.is_none());
-        // assert_eq!(
-        //     res,
-        //     Some(MembershipMessage::new("local".to_string(), vec!["a".to_string()]).into())
-        // );
+        assert_eq!(
+            res,
+            Some(MembershipMessage::new("local".to_string(), Some(vec!["a".to_string()])).into())
+        );
     }
 
     #[tokio::test]
     async fn it_should_create_membership_as_service_and_respond_to_membership_with_none() {
-        let mut p = Membership::new("local".to_string());
+        let message_id_generator = MessageIdGenerator::new("localhost".to_string());
+        let membership_handle = MembershipHandle::start("localhost".to_string()).await;
+        let state = State::new(membership_handle, message_id_generator);
+
+        let mut p = Membership::new("local".to_string(), state);
         let res = p
             .ready()
             .await
             .unwrap()
-            .call(MembershipMessage::new("local".to_string(), vec!["a".to_string()]).into())
+            .call(MembershipMessage::new("local".to_string(), Some(vec!["a".to_string()])).into())
             .await
             .unwrap();
         assert!(res.is_none());
     }
-
-    #[test]
-    fn it_should_create_default_membership_message() {
-        assert_eq!(MembershipMessage::default().sender_id, "".to_string())
-    }
 }