fix(rust): prevent `Router` from stopping when message handling error encountered
@@ -109,10 +109,8 @@ impl Router {
}
}
- /// Block current task running this router. Return fatal errors
- async fn run_inner(&mut self) -> Result<()> {
+ async fn handle_msg(&mut self, msg: NodeMessage) -> Result<bool> {
use NodeMessage::*;
- while let Some(msg) = self.receiver.recv().await {
match msg {
// Successful router registration command
Router(tt, addr, sender) if !self.external.contains_key(&tt) => {
@@ -143,9 +141,7 @@ impl Router {
StartProcessor(addr, senders, ref reply) => {
start_processor::exec(self, addr, senders, reply).await?
}
- StopProcessor(ref addr, ref reply) => {
- stop_processor::exec(self, addr, reply).await?
- }
+ StopProcessor(ref addr, ref reply) => stop_processor::exec(self, addr, reply).await?,
//// ==! Core node controls
StopNode(ShutdownType::Graceful(timeout), reply) => {
@@ -156,13 +152,13 @@ impl Router {
.send(NodeReply::ok())
.await
.map_err(|_| Error::InternalIOFailure)?;
- break;
+ return Ok(true);
};
}
}
StopNode(ShutdownType::Immediate, reply) => {
shutdown::immediate(self, reply).await?;
- break;
+ return Ok(true);
}
AbortNode => {
@@ -172,7 +168,7 @@ impl Router {
.await
.map_err(|_| Error::InternalIOFailure)?;
self.map.internal.clear();
- break;
+ return Ok(true);
}
}
@@ -193,7 +189,7 @@ impl Router {
.send(NodeReply::ok())
.await
.map_err(|_| Error::InternalIOFailure)?;
- break;
+ return Ok(true);
}
}
}
@@ -217,15 +213,31 @@ impl Router {
// Handle route/ sender requests
SenderReq(ref addr, ref reply) => match determine_type(addr) {
- RouteType::Internal(ref addr) => {
- utils::resolve(self, addr, reply, false).await?
- }
+ RouteType::Internal(ref addr) => utils::resolve(self, addr, reply, false).await?,
RouteType::External(tt) => {
let addr = utils::router_addr(self, tt)?;
utils::resolve(self, &addr, reply, true).await?
}
},
}
+
+ Ok(false)
+ }
+
+ /// Block current task running this router. Return fatal errors
+ async fn run_inner(&mut self) -> Result<()> {
+ while let Some(msg) = self.receiver.recv().await {
+ let msg_str = format!("{}", msg);
+ match self.handle_msg(msg).await {
+ Ok(should_break) => {
+ if should_break {
+ break;
+ }
+ }
+ Err(err) => {
+ warn!("Router error: {} while handling {}", err, msg_str);
+ }
+ }
}
Ok(())
