fix(router): retry messages when server is not ready (#1651)
* fix(router): retry messages when server is not ready * style(fmt): rustfmt
This commit is contained in:
parent
dba5dcbd83
commit
e910db9bad
1 changed files with 146 additions and 114 deletions
|
|
@ -445,6 +445,22 @@ fn route_action(
|
||||||
should_break
|
should_break
|
||||||
}
|
}
|
||||||
|
|
||||||
|
macro_rules! send_to_screen_or_retry_queue {
|
||||||
|
($rlocked_sessions:expr, $message:expr, $instruction: expr, $retry_queue:expr) => {{
|
||||||
|
match $rlocked_sessions.as_ref() {
|
||||||
|
Some(session_metadata) => {
|
||||||
|
session_metadata.senders.send_to_screen($message).unwrap();
|
||||||
|
},
|
||||||
|
None => {
|
||||||
|
log::warn!("Server not ready, trying to place instruction in retry queue...");
|
||||||
|
if let Some(retry_queue) = $retry_queue.as_mut() {
|
||||||
|
retry_queue.push($instruction);
|
||||||
|
}
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}};
|
||||||
|
}
|
||||||
|
|
||||||
pub(crate) fn route_thread_main(
|
pub(crate) fn route_thread_main(
|
||||||
session_data: Arc<RwLock<Option<SessionMetaData>>>,
|
session_data: Arc<RwLock<Option<SessionMetaData>>>,
|
||||||
session_state: Arc<RwLock<SessionState>>,
|
session_state: Arc<RwLock<SessionState>>,
|
||||||
|
|
@ -453,12 +469,16 @@ pub(crate) fn route_thread_main(
|
||||||
mut receiver: IpcReceiverWithContext<ClientToServerMsg>,
|
mut receiver: IpcReceiverWithContext<ClientToServerMsg>,
|
||||||
client_id: ClientId,
|
client_id: ClientId,
|
||||||
) {
|
) {
|
||||||
loop {
|
let mut retry_queue = vec![];
|
||||||
|
'route_loop: loop {
|
||||||
match receiver.recv() {
|
match receiver.recv() {
|
||||||
Some((instruction, err_ctx)) => {
|
Some((instruction, err_ctx)) => {
|
||||||
err_ctx.update_thread_ctx();
|
err_ctx.update_thread_ctx();
|
||||||
let rlocked_sessions = session_data.read().unwrap();
|
let rlocked_sessions = session_data.read().unwrap();
|
||||||
|
let handle_instruction = |instruction: ClientToServerMsg,
|
||||||
|
mut retry_queue: Option<&mut Vec<ClientToServerMsg>>|
|
||||||
|
-> bool {
|
||||||
|
let mut should_break = false;
|
||||||
match instruction {
|
match instruction {
|
||||||
ClientToServerMsg::Action(action, maybe_client_id) => {
|
ClientToServerMsg::Action(action, maybe_client_id) => {
|
||||||
let client_id = maybe_client_id.unwrap_or(client_id);
|
let client_id = maybe_client_id.unwrap_or(client_id);
|
||||||
|
|
@ -476,7 +496,7 @@ pub(crate) fn route_thread_main(
|
||||||
&to_server,
|
&to_server,
|
||||||
client_id,
|
client_id,
|
||||||
) {
|
) {
|
||||||
break;
|
should_break = true;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|
@ -498,14 +518,13 @@ pub(crate) fn route_thread_main(
|
||||||
.unwrap();
|
.unwrap();
|
||||||
},
|
},
|
||||||
ClientToServerMsg::TerminalPixelDimensions(pixel_dimensions) => {
|
ClientToServerMsg::TerminalPixelDimensions(pixel_dimensions) => {
|
||||||
rlocked_sessions
|
// this is experimental, please be cautious implementing this elsewhere
|
||||||
.as_ref()
|
send_to_screen_or_retry_queue!(
|
||||||
.unwrap()
|
rlocked_sessions,
|
||||||
.senders
|
ScreenInstruction::TerminalPixelDimensions(pixel_dimensions),
|
||||||
.send_to_screen(ScreenInstruction::TerminalPixelDimensions(
|
instruction,
|
||||||
pixel_dimensions,
|
retry_queue
|
||||||
))
|
);
|
||||||
.unwrap();
|
|
||||||
},
|
},
|
||||||
ClientToServerMsg::BackgroundColor(background_color_instruction) => {
|
ClientToServerMsg::BackgroundColor(background_color_instruction) => {
|
||||||
rlocked_sessions
|
rlocked_sessions
|
||||||
|
|
@ -563,23 +582,36 @@ pub(crate) fn route_thread_main(
|
||||||
// we don't unwrap this because we don't really care if there's an error here (eg.
|
// we don't unwrap this because we don't really care if there's an error here (eg.
|
||||||
// if the main server thread exited before this router thread did)
|
// if the main server thread exited before this router thread did)
|
||||||
let _ = to_server.send(ServerInstruction::RemoveClient(client_id));
|
let _ = to_server.send(ServerInstruction::RemoveClient(client_id));
|
||||||
break;
|
return true;
|
||||||
},
|
},
|
||||||
ClientToServerMsg::KillSession => {
|
ClientToServerMsg::KillSession => {
|
||||||
to_server.send(ServerInstruction::KillSession).unwrap();
|
to_server.send(ServerInstruction::KillSession).unwrap();
|
||||||
},
|
},
|
||||||
ClientToServerMsg::ConnStatus => {
|
ClientToServerMsg::ConnStatus => {
|
||||||
let _ = to_server.send(ServerInstruction::ConnStatus(client_id));
|
let _ = to_server.send(ServerInstruction::ConnStatus(client_id));
|
||||||
break;
|
should_break = true;
|
||||||
},
|
},
|
||||||
ClientToServerMsg::DetachSession(client_id) => {
|
ClientToServerMsg::DetachSession(client_id) => {
|
||||||
let _ = to_server.send(ServerInstruction::DetachSession(client_id));
|
let _ = to_server.send(ServerInstruction::DetachSession(client_id));
|
||||||
break;
|
should_break = true;
|
||||||
},
|
},
|
||||||
ClientToServerMsg::ListClients => {
|
ClientToServerMsg::ListClients => {
|
||||||
let _ = to_server.send(ServerInstruction::ActiveClients(client_id));
|
let _ = to_server.send(ServerInstruction::ActiveClients(client_id));
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
should_break
|
||||||
|
};
|
||||||
|
for instruction_to_retry in retry_queue.drain(..) {
|
||||||
|
log::warn!("Server ready, retrying sending instruction.");
|
||||||
|
let should_break = handle_instruction(instruction_to_retry, None);
|
||||||
|
if should_break {
|
||||||
|
break 'route_loop;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let should_break = handle_instruction(instruction, Some(&mut retry_queue));
|
||||||
|
if should_break {
|
||||||
|
break 'route_loop;
|
||||||
|
}
|
||||||
},
|
},
|
||||||
None => {
|
None => {
|
||||||
log::error!("Received empty message from client");
|
log::error!("Received empty message from client");
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue