Refactor retry connection handler in NetworkPart
@@ -144,24 +144,20 @@ impl NetworkPart {
let handle = handle_orig.clone();
let requests_handle = self.network_requests.1.for_each(move |request| {
let handle = handle.clone();
- let network_tx = network_tx.clone();
let outgoing_connections = outgoing_connections.clone();
match request {
NetworkRequest::SendMessage(peer, msg) => {
- let conn_tx =
- if let Some(conn_tx) = outgoing_connections.get(peer) {
- conn_tx
- } else {
- // Check limit
+ let network_tx_cloned = network_tx.clone();
+ let conn_tx = outgoing_connections.get(peer)
+ .or_else(move || {
if outgoing_connections.len() >= outgoing_connections_limit {
warn!(
"Rejected outgoing connection with peer={}, \
connections limit reached.",
peer
);
- let event = NetworkEvent::PeerDisconnected(peer);
- handle.spawn(network_tx.clone().send(event).map(drop).map_err(log_error));
- }
+ None
+ } else {
// Register outgoing channel.
let (conn_tx, conn_rx) = mpsc::channel(10);
outgoing_connections.insert(peer, &conn_tx);
@@ -208,18 +204,25 @@ impl NetworkPart {
.then(move |res| {
trace!("Connection with peer={} closed, reason={:?}", peer, res);
outgoing_connections.clone().remove(&peer);
- network_tx
+ network_tx_cloned
.clone()
.send(NetworkEvent::PeerDisconnected(peer))
.map(drop)
})
.map_err(log_error);
handle.spawn(connect_handle);
- conn_tx
- };
-
+ Some(conn_tx)
+ }
+ });
+ if let Some(conn_tx) = conn_tx {
let send_handle = conn_tx.send(msg).map_err(log_error).map(drop);
tobox(send_handle)
+ } else {
+ let event = NetworkEvent::PeerDisconnected(peer);
+ let fut = network_tx.clone().send(event).map_err(log_error)
+ .into_future();
+ tobox(fut)
+ }
}
NetworkRequest::DisconnectWithPeer(peer) => {
outgoing_connections.remove(&peer);
