lx/netapp
1
1
Fork 5
Code Issues 2 Pull requests 1 Releases 1 Wiki Activity

Don't close connection at first (stupid) excuse

Browse source
This commit is contained in:
Alex 2020-12-04 17:12:10 +01:00
parent 45766aa5d7
commit b247f02c29
3 changed files with 13 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
examples/fullmesh.rs
View file

@ -1,4 +1,5 @@
use std::net::SocketAddr; use std::net::SocketAddr;
use std::io::Write;
use log::info; use log::info;

8
src/conn.rs
View file

@ -216,7 +216,9 @@ impl ClientConn {
if let Some((id, resp)) = resp { if let Some((id, resp)) = resp {
trace!("dispatch_resp: got resp to {}, {} bytes", id, resp.len()); trace!("dispatch_resp: got resp to {}, {} bytes", id, resp.len());
if let Some(ch) = resp_notify.remove(&id) { if let Some(ch) = resp_notify.remove(&id) {
ch.send(resp).map_err(|_| Error::Message("Could not dispatch reply".to_string()))?; if ch.send(resp).is_err() {
debug!("Could not dispatch reply (channel probably closed, happens if request was canceled)");
}
} else { } else {
resps.insert(id, resp); resps.insert(id, resp);
} }
@ -226,7 +228,9 @@ impl ClientConn {
if let Some((id, resp_ch)) = resp_ch { if let Some((id, resp_ch)) = resp_ch {
trace!("dispatch_resp: got resp_ch {}", id); trace!("dispatch_resp: got resp_ch {}", id);
if let Some(rs) = resps.remove(&id) { if let Some(rs) = resps.remove(&id) {
resp_ch.send(rs).map_err(|_| Error::Message("Could not dispatch reply".to_string()))?; if resp_ch.send(rs).is_err() {
debug!("Could not dispatch reply (channel probably closed, happens if request was canceled)");
}
} else { } else {
resp_notify.insert(id, resp_ch); resp_notify.insert(id, resp_ch);
} }

12
src/peering/basalt.rs
View file

@ -3,7 +3,7 @@ use std::net::SocketAddr;
use std::sync::{Arc, RwLock}; use std::sync::{Arc, RwLock};
use std::time::Duration; use std::time::Duration;
use log::{debug, info, warn}; use log::{trace, debug, info, warn};
use lru::LruCache; use lru::LruCache;
use rand::{thread_rng, Rng}; use rand::{thread_rng, Rng};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@ -303,7 +303,7 @@ impl Basalt {
.sample(count) .sample(count)
.iter() .iter()
.map(|p| { .map(|p| {
info!("KYEV S {}", hex::encode(p.id)); debug!("KYEV S {}", hex::encode(p.id));
p.id p.id
}) })
.collect::<Vec<_>>() .collect::<Vec<_>>()
@ -339,7 +339,7 @@ impl Basalt {
{ {
Ok(resp) => { Ok(resp) => {
self.handle_peer_list(&resp.peers[..]); self.handle_peer_list(&resp.peers[..]);
info!("KYEV PEXi {}", hex::encode(peer)); trace!("KYEV PEXi {}", hex::encode(peer));
} }
Err(e) => { Err(e) => {
warn!("Error during pull exchange: {}", e); warn!("Error during pull exchange: {}", e);
@ -351,7 +351,7 @@ impl Basalt {
let push_msg = self.make_push_message(); let push_msg = self.make_push_message();
match self.netapp.request(&peer, push_msg, PRIO_NORMAL).await { match self.netapp.request(&peer, push_msg, PRIO_NORMAL).await {
Ok(_) => { Ok(_) => {
info!("KYEV PEXo {}", hex::encode(peer)); trace!("KYEV PEXo {}", hex::encode(peer));
} }
Err(e) => { Err(e) => {
warn!("Error during push exchange: {}", e); warn!("Error during push exchange: {}", e);
@ -371,7 +371,7 @@ impl Basalt {
tokio::time::delay_for(self.param.reset_interval).await; tokio::time::delay_for(self.param.reset_interval).await;
{ {
info!("KYEV R {}", self.param.reset_count); debug!("KYEV R {}", self.param.reset_count);
let mut view = self.view.write().unwrap(); let mut view = self.view.write().unwrap();
let prev_peers = view.current_peers(); let prev_peers = view.current_peers();
@ -419,7 +419,7 @@ impl Basalt {
attempts.insert(peer); attempts.insert(peer);
} }
let res = self.netapp.clone().try_connect(peer.addr, peer.id).await; let res = self.netapp.clone().try_connect(peer.addr, peer.id).await;
debug!("Connection attempt to {}: {:?}", peer.addr, res); trace!("Connection attempt to {}: {:?}", peer.addr, res);
self.current_attempts.write().unwrap().remove(&peer); self.current_attempts.write().unwrap().remove(&peer);