|
@@ -3,7 +3,6 @@ use async_std::sync::{Arc, Mutex};
|
|
|
use async_executor::Executor;
|
|
use async_executor::Executor;
|
|
|
use async_trait::async_trait;
|
|
use async_trait::async_trait;
|
|
|
use log::debug;
|
|
use log::debug;
|
|
|
-use url::Url;
|
|
|
|
|
|
|
|
|
|
use crate::{net, Result};
|
|
use crate::{net, Result};
|
|
|
|
|
|
|
@@ -16,7 +15,6 @@ pub struct ProtocolRaft {
|
|
|
msg_sub: net::MessageSubscription<NetMsg>,
|
|
msg_sub: net::MessageSubscription<NetMsg>,
|
|
|
p2p: net::P2pPtr,
|
|
p2p: net::P2pPtr,
|
|
|
msgs: Arc<Mutex<Vec<u64>>>,
|
|
msgs: Arc<Mutex<Vec<u64>>>,
|
|
|
- channel_address: Url,
|
|
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
impl ProtocolRaft {
|
|
impl ProtocolRaft {
|
|
@@ -31,7 +29,6 @@ impl ProtocolRaft {
|
|
|
message_subsytem.add_dispatch::<NetMsg>().await;
|
|
message_subsytem.add_dispatch::<NetMsg>().await;
|
|
|
|
|
|
|
|
let msg_sub = channel.subscribe_msg::<NetMsg>().await.expect("Missing NetMsg dispatcher!");
|
|
let msg_sub = channel.subscribe_msg::<NetMsg>().await.expect("Missing NetMsg dispatcher!");
|
|
|
- let channel_address = channel.address();
|
|
|
|
|
|
|
|
|
|
Arc::new(Self {
|
|
Arc::new(Self {
|
|
|
id,
|
|
id,
|
|
@@ -40,13 +37,11 @@ impl ProtocolRaft {
|
|
|
jobsman: net::ProtocolJobsManager::new("ProtocolRaft", channel),
|
|
jobsman: net::ProtocolJobsManager::new("ProtocolRaft", channel),
|
|
|
p2p,
|
|
p2p,
|
|
|
msgs,
|
|
msgs,
|
|
|
- channel_address,
|
|
|
|
|
})
|
|
})
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
async fn handle_receive_msg(self: Arc<Self>) -> Result<()> {
|
|
async fn handle_receive_msg(self: Arc<Self>) -> Result<()> {
|
|
|
debug!(target: "raft", "ProtocolRaft::handle_receive_msg() [START]");
|
|
debug!(target: "raft", "ProtocolRaft::handle_receive_msg() [START]");
|
|
|
- let exclude_list = vec![self.channel_address.clone()];
|
|
|
|
|
loop {
|
|
loop {
|
|
|
let msg = self.msg_sub.receive().await?;
|
|
let msg = self.msg_sub.receive().await?;
|
|
|
|
|
|
|
@@ -56,14 +51,16 @@ impl ProtocolRaft {
|
|
|
&msg.id, &msg.method
|
|
&msg.id, &msg.method
|
|
|
);
|
|
);
|
|
|
|
|
|
|
|
- if self.msgs.lock().await.contains(&msg.id) {
|
|
|
|
|
- continue
|
|
|
|
|
|
|
+ {
|
|
|
|
|
+ let mut msgs = self.msgs.lock().await;
|
|
|
|
|
+ if msgs.contains(&msg.id) {
|
|
|
|
|
+ continue
|
|
|
|
|
+ }
|
|
|
|
|
+ msgs.push(msg.id);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- self.msgs.lock().await.push(msg.id);
|
|
|
|
|
-
|
|
|
|
|
let msg = (*msg).clone();
|
|
let msg = (*msg).clone();
|
|
|
- self.p2p.broadcast_with_exclude(msg.clone(), &exclude_list).await?;
|
|
|
|
|
|
|
+ self.p2p.broadcast(msg.clone()).await?;
|
|
|
|
|
|
|
|
match (self.id.clone(), msg.recipient_id.clone()) {
|
|
match (self.id.clone(), msg.recipient_id.clone()) {
|
|
|
// check if the ids are equal when both
|
|
// check if the ids are equal when both
|