Remove HashMap::entry
@@ -21,7 +21,8 @@ use tokio_io::AsyncRead;
use std::net::SocketAddr;
use std::time::Duration;
-use std::collections::hash_map::{HashMap, Entry};
+use std::collections::HashMap;
+use std::rc::Rc;
use messages::{Any, Connect, RawMessage};
use node::{ExternalMessage, NodeTimeout};
@@ -112,16 +113,16 @@ impl NetworkPart {
let requests_handle = self.network_requests.1.for_each(|request| {
match request {
NetworkRequest::SendMessage(peer, msg) => {
- let conn_tx = match outgoing_connections.entry(peer) {
- Entry::Occupied(entry) => entry.get().clone(),
- Entry::Vacant(entry) => {
+ let conn_tx = if let Some(conn_tx) = outgoing_connections.get(&peer).cloned() {
+ conn_tx
+ } else {
let (conn_tx, conn_rx) = mpsc::channel(10);
- let socket = TcpStream::connect(&peer, &handle);
+ outgoing_connections.insert(peer, conn_tx.clone());
let requests_tx = requests_tx.clone();
- let connect_handle = socket
+ let connect_handle = TcpStream::connect(&peer, &handle)
.and_then(move |sock| {
- info!("Tokio: Established connection with peer={}", peer);
+ info!("Established connection with peer={}", peer);
let stream = sock.framed(MessagesCodec);
let (sink, stream) = stream.split();
@@ -139,22 +140,19 @@ impl NetworkPart {
Either::B((_, _writer)) => Ok(()).into_future(),
})
})
- .map_err(log_error)
- .then(move |_| {
- info!("Tokio: Connection with peer={} closed", peer);
+ .then(move |res| {
+ info!("Connection with peer={} closed, reason={:?}", peer, res);
+ // outgoing_connections.remove(&peer);
+
let request = NetworkRequest::DisconnectWithPeer(peer);
requests_tx
.clone()
.send(request)
.map(forget_result)
.map_err(into_other)
- })
- .map_err(log_error);
+ }).map_err(log_error);
handle.spawn(connect_handle);
-
- entry.insert(conn_tx.clone());
conn_tx
- }
};
let duration = Duration::from_secs(5);
@@ -196,7 +194,7 @@ impl NetworkPart {
let server = listener
.incoming()
.fold(network_tx, move |network_tx, (sock, addr)| {
- info!("Tokio: Accepted incoming connection with peer={}", addr);
+ info!("Accepted incoming connection with peer={}", addr);
let stream = sock.framed(MessagesCodec);
let (_, stream) = stream.split();
@@ -213,7 +211,7 @@ impl NetworkPart {
}
})
.and_then(move |(connect, stream)| {
- info!("Tokio: Received handshake message={:?}", connect);
+ info!("Received handshake message={:?}", connect);
let event = NetworkEvent::PeerConnected(addr, connect);
let connect_event = network_tx.clone().send(event).map_err(into_other);
