feat: Implement useful callbacks

README.md

@@ -137,7 +137,7 @@ let raft_handle = tokio::spawn(raft.clone().run());
 raft.join(vec![join_ticket]).await;
 
 // ...
-tokio::try_join!(join_ticket)?;
+tokio::try_join!(raft_handle)?;
 ```
 
 ### Manipulate FSM by RaftServiceClient

examples/memstore/dynamic-members/src/main.rs

@@ -5,8 +5,7 @@ extern crate slog_term;
 
 use actix_web::{web, App, HttpServer};
 use raftify::{
-    raft::{formatter::set_custom_formatter, logger::Slogger},
-    CustomFormatter, Raft as Raft_,
+    custom_callbacks::set_on_member_changed, raft::{formatter::set_custom_formatter, logger::Slogger}, CustomFormatter, Raft as Raft_
 };
 use slog::Drain;
 use slog_envlogger::LogBuilder;
@@ -54,6 +53,11 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
         slog: slog::Logger::root(drain, o!()),
     });
 
+    set_on_member_changed(Box::new(|peers_| Box::pin(async move {
+        let peers_ = peers_.lock().await;
+        println!("Peers changed: {:?}", peers_.inner);
+    }))).await;
+
     set_custom_formatter(CustomFormatter::<LogEntry, HashStore>::new());
 
     let options = Options::from_args();

examples/memstore/static-members/src/main.rs

@@ -5,8 +5,7 @@ extern crate slog_term;
 
 use actix_web::{web, App, HttpServer};
 use raftify::{
-    raft::{formatter::set_custom_formatter, logger::Slogger},
-    ClusterJoinTicket, CustomFormatter, Raft as Raft_,
+    custom_callbacks::set_on_member_changed, raft::{formatter::set_custom_formatter, logger::{Logger, Slogger}}, CustomFormatter, Raft as Raft_
 };
 use slog::Drain;
 use slog_envlogger::LogBuilder;
@@ -57,6 +56,11 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
         slog: slog::Logger::root(drain, o!()),
     });
 
+    set_on_member_changed(Box::new(|peers_| Box::pin(async move {
+        let peers_ = peers_.lock().await;
+        println!("Peers changed: {:?}", peers_.inner);
+    }))).await;
+
     set_custom_formatter(CustomFormatter::<LogEntry, HashStore>::new());
 
     let options = Options::from_args();

raftify/Cargo.toml

@@ -29,6 +29,7 @@ tonic = "0.9.2"
 built = "0.5"
 clap = "3.0"
 chrono = "0.4.31"
+lazy_static = "1.4.0"
 
 [build-dependencies]
 tonic-build = "0.9.2"

raftify/src/custom_callbacks.rs

@@ -0,0 +1,16 @@
+use lazy_static::lazy_static;
+use std::{future::Future, pin::Pin, sync::Arc};
+use tokio::sync::Mutex;
+
+use crate::Peers;
+
+type ON_MEMBER_CHANGED_CALLBACK = dyn FnMut(Arc<Mutex<Peers>>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync;
+
+lazy_static! {
+    pub static ref ON_MEMBER_CHANGED: Arc<Mutex<Option<Box<ON_MEMBER_CHANGED_CALLBACK>>>> = Arc::new(Mutex::new(None));
+}
+
+pub async fn set_on_member_changed(callback: Box<ON_MEMBER_CHANGED_CALLBACK>) {
+    let mut global_callback = ON_MEMBER_CHANGED.lock().await;
+    *global_callback = Some(callback);
+}

raftify/src/lib.rs

@@ -18,6 +18,7 @@ mod storage;
 mod utils;
 
 pub mod cli;
+pub mod custom_callbacks;
 pub mod raft_service;
 
 pub use {

raftify/src/raft_node/mod.rs

@@ -28,6 +28,7 @@ use utils::inspect_raftnode;
 
 use crate::{
     create_client,
+    custom_callbacks::{set_on_member_changed, ON_MEMBER_CHANGED},
     error::{Result, SendMessageError},
     raft::{
         eraftpb::{
@@ -907,6 +908,11 @@ impl<
             }
         }
 
+        let mut on_member_changed = ON_MEMBER_CHANGED.lock().await;
+        if let Some(cb) = on_member_changed.as_mut() {
+            let _ = cb(self.peers.clone()).await;
+        }
+
         Ok(())
     }