Upgrade deps to rmqtt-raft = "0.3.4"

rmqtt · Oct 14, 2023 · a6d5c8b · a6d5c8b
1 parent 49fad0d
commit a6d5c8b
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 7 deletions.
diff --git a/rmqtt-plugins/rmqtt-cluster-raft/Cargo.toml b/rmqtt-plugins/rmqtt-cluster-raft/Cargo.toml
@@ -6,7 +6,7 @@ edition = "2021"
 
 [dependencies]
 rmqtt = "0.2"
-rmqtt-raft = { version = "0.3.3", features = ["reuse"] }
+rmqtt-raft = { version = "0.3.4", features = ["reuse"] }
 #rmqtt-raft = { path = "../../../rmqtt-raft", features = ["reuse"] }
 serde = { version = "1.0", features = ["derive"] }
 backoff = { version = "0.4", features = ["futures", "tokio"] }
diff --git a/rmqtt-plugins/rmqtt-cluster-raft/src/handler.rs b/rmqtt-plugins/rmqtt-cluster-raft/src/handler.rs
@@ -43,7 +43,7 @@ impl Handler for HookHandler {
                         if let Err(e) = retry(BACKOFF_STRATEGY.clone(), || async {
                             let msg = msg.clone();
                             let mailbox = raft_mailbox.clone();
-                            let res = async move { mailbox.send(msg).await }
+                            let res = async move { mailbox.send_proposal(msg).await }
                                 .spawn(task_exec_queue())
                                 .result()
                                 .await
@@ -73,7 +73,7 @@ impl Handler for HookHandler {
                     if let Err(e) = retry(BACKOFF_STRATEGY.clone(), || async {
                         let msg = msg.clone();
                         let mailbox = raft_mailbox.clone();
-                        let res = async move { mailbox.send(msg).await }
+                        let res = async move { mailbox.send_proposal(msg).await }
                             .spawn(task_exec_queue())
                             .result()
                             .await

diff --git a/rmqtt-plugins/rmqtt-cluster-raft/src/router.rs b/rmqtt-plugins/rmqtt-cluster-raft/src/router.rs
@@ -119,7 +119,7 @@ impl Router for &'static ClusterRouter {
 
         let msg = Message::Add { topic_filter, id, opts }.encode()?;
         let mailbox = self.raft_mailbox().await;
-        let _ = async move { mailbox.send(msg).await.map_err(anyhow::Error::new) }
+        let _ = async move { mailbox.send_proposal(msg).await.map_err(anyhow::Error::new) }
             .spawn(task_exec_queue())
             .result()
             .await
@@ -136,7 +136,7 @@ impl Router for &'static ClusterRouter {
             if let Err(e) = retry(BACKOFF_STRATEGY.clone(), || async {
                 let msg = msg.clone();
                 let mailbox = raft_mailbox.clone();
-                let res = async move { mailbox.send(msg).await }
+                let res = async move { mailbox.send_proposal(msg).await }
                     .spawn(task_exec_queue())
                     .result()
                     .await

diff --git a/rmqtt-plugins/rmqtt-cluster-raft/src/shared.rs b/rmqtt-plugins/rmqtt-cluster-raft/src/shared.rs
@@ -53,7 +53,7 @@ impl Entry for ClusterLockEntry {
     async fn try_lock(&self) -> Result<Box<dyn Entry>> {
         let msg = RaftMessage::HandshakeTryLock { id: self.id() }.encode()?;
         let raft_mailbox = self.cluster_shared.router.raft_mailbox().await;
-        let reply = raft_mailbox.send(msg).await.map_err(anyhow::Error::new)?;
+        let reply = raft_mailbox.send_proposal(msg).await.map_err(anyhow::Error::new)?;
         let mut prev_node_id = None;
         if !reply.is_empty() {
             match RaftMessageReply::decode(&reply)? {
@@ -92,7 +92,7 @@ impl Entry for ClusterLockEntry {
     async fn set(&mut self, session: Session, tx: Tx, conn: ClientInfo) -> Result<()> {
         let msg = RaftMessage::Connected { id: session.id.clone() }.encode()?;
         let raft_mailbox = self.cluster_shared.router.raft_mailbox().await;
-        let reply = raft_mailbox.send(msg).await.map_err(anyhow::Error::new)?;
+        let reply = raft_mailbox.send_proposal(msg).await.map_err(anyhow::Error::new)?;
         if !reply.is_empty() {
             let reply = RaftMessageReply::decode(&reply)?;
             match reply {