Remove handler from listener of NetworkPart
@@ -158,12 +158,8 @@ impl NetworkPart {
connections limit reached.",
peer
);
- Self::send_event(
- &handle,
- &network_tx,
- NetworkEvent::PeerDisconnected(peer),
- );
- return Ok(());
+ let event = NetworkEvent::PeerDisconnected(peer);
+ handle.spawn(network_tx.clone().send(event).map(drop).map_err(log_error));
}
// Register outgoing channel.
let (conn_tx, conn_rx) = mpsc::channel(10);
@@ -230,7 +226,8 @@ impl NetworkPart {
}
NetworkRequest::DisconnectWithPeer(peer) => {
outgoing_connections.remove(&peer);
- Self::send_event(&handle, &network_tx, NetworkEvent::PeerDisconnected(peer));
+ let event = NetworkEvent::PeerDisconnected(peer);
+ handle.spawn(network_tx.clone().send(event).map(drop).map_err(log_error));
}
// Immediately stop the event loop.
NetworkRequest::Shutdown => {
@@ -256,11 +253,9 @@ impl NetworkPart {
// TODO Don't use unwrap here!
let listener = TcpListener::bind(&self.listen_address, &handle_orig).unwrap();
let network_tx = self.network_tx.clone();
- let handle = handle_orig.clone();
let server = listener
.incoming()
.for_each(move |(sock, addr)| {
- let handle = handle.clone();
// Increment reference counter
let holder = Rc::downgrade(&incoming_connections_counter);
// Check incoming connections count
@@ -271,8 +266,8 @@ impl NetworkPart {
connections limit reached.",
addr
);
- return Ok(());
- }
+ tobox(future::ok(()))
+ } else {
trace!("Accepted incoming connection with peer={}", addr);
let stream = sock.framed(MessagesCodec);
let (_, stream) = stream.split();
@@ -307,10 +302,9 @@ impl NetworkPart {
.map(|_| {
// Ensure that holder lives until the stream ends.
let _holder = holder;
- })
- .map_err(log_error);
- handle.spawn(connection_handler);
- Ok(())
+ });
+ tobox(connection_handler)
+ }
})
.map_err(log_error);
@@ -322,10 +316,6 @@ impl NetworkPart {
];
tobox(future::join_all(tasks))
}
-
- fn send_event(handle: &Handle, network_tx: &mpsc::Sender<NetworkEvent>, event: NetworkEvent) {
- handle.spawn(network_tx.clone().send(event).map(drop).map_err(log_error));
- }
}
fn tobox<F: Future + 'static>(f: F) -> Box<Future<Item=(), Error=F::Error>> {
