Move run_sender_loop to own method
Reorganize the run_sender_loop to its own function.  This will reduce
the nested complexity of the spawn closure within the build function.
// See the License for the specific language governing permissions and
// limitations under the License.
-use std::sync::mpsc::{channel, Sender};
+use std::sync::mpsc::{channel, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::thread;
@@ -290,17 +290,50 @@ where
let send_join_handle = thread::Builder::new()
.name("PeerInterconnect Sender".into())
.spawn(move || {
+ if let Err(err) =
+ run_send_loop(peers, &peer_connector, dispatched_receiver, message_sender)
+ {
+ error!("Shutting down peer interconnect sender: {}", err);
+ }
+ })
+ .map_err(|err| {
+ PeerInterconnectError::StartUpError(format!(
+ "Unable to start PeerInterconnect sender thread {}",
+ err
+ ))
+ })?;
+
+ Ok(PeerInterconnect {
+ dispatched_sender: dispatched_sender.clone(),
+ recv_join_handle,
+ send_join_handle,
+ shutdown_handle: ShutdownHandle {
+ sender: dispatched_sender,
+ dispatch_shutdown: network_dispatcher_shutdown,
+ },
+ })
+ }
+}
+
+fn run_send_loop<S>(
+ peers: Arc<Mutex<BiHashMap<String, String>>>,
+ peer_connector: &PeerManagerConnector,
+ receiver: Receiver<SendRequest>,
+ message_sender: S,
+) -> Result<(), String>
+where
+ S: MatrixSender + 'static,
+{
loop {
// receive message from internal handlers to send over the network
- let (recipient, payload) = match dispatched_receiver.recv() {
+ let (recipient, payload) = match receiver.recv() {
Ok(SendRequest::Message { recipient, payload }) => (recipient, payload),
Ok(SendRequest::Shutdown) => {
info!("Received Shutdown");
- break;
+ break Ok(());
}
Err(err) => {
- error!("Unable to receive message from handlers: {}", err);
- break;
+ break Err(format!("Unable to receive message from handlers: {}", err));
}
};
// convert recipient (peer_id) to connection_id
@@ -308,8 +341,7 @@ where
let mut peers = match peers.lock() {
Ok(recv_peers) => recv_peers,
Err(_) => {
- error!("PeerInterconnect state has been poisoned");
- break;
+ break Err("PeerInterconnect state has been poisoned".into());
}
};
@@ -323,8 +355,7 @@ where
*peers = match peer_connector.connection_ids() {
Ok(peers) => peers,
Err(err) => {
- error!("Unable to get peer map: {}", err);
- break;
+ break Err(format!("Unable to get peer map: {}", err));
}
};
@@ -339,34 +370,13 @@ where
// if peer exists, send message over the network
if !connection_id.is_empty() {
- match message_sender.send(connection_id.to_string(), payload) {
- Ok(_) => (),
- Err(err) => {
- error!("Unable to send message to {}", err);
- }
+ if let Err(err) = message_sender.send(connection_id.to_string(), payload) {
+ error!("Unable to send message to {}: {}", recipient, err);
}
} else {
error!("Cannot send message, unknown peer: {}", recipient);
}
}
- })
- .map_err(|err| {
- PeerInterconnectError::StartUpError(format!(
- "Unable to start PeerInterconnect sender thread {}",
- err
- ))
- })?;
-
- Ok(PeerInterconnect {
- dispatched_sender: dispatched_sender.clone(),
- recv_join_handle,
- send_join_handle,
- shutdown_handle: ShutdownHandle {
- sender: dispatched_sender,
- dispatch_shutdown: network_dispatcher_shutdown,
- },
- })
- }
}
#[derive(Clone)]
