Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

graceful shutdown: distinguish between stopped and conn closed #1220

Merged
merged 2 commits into from
Oct 19, 2023
Merged

graceful shutdown: distinguish between stopped and conn closed #1220

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8899e00
graceful shutdown: distinguish between stopped and conn closed
niklasad1 Oct 19, 2023
539ab67
grumbles: ws_stream fuse
niklasad1 Oct 19, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions server/src/transport/ws.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -288,13 +288,15 @@ pub(crate) async fn background_task<L: Logger>(sender: Sender, receiver: Receive
// Return the `Closed` error to avoid logging unnecessary warnings on clean shutdown.
Err(e) => Some((Err(e), receiver)),
}
});
})
.fuse();

tokio::pin!(ws_stream);

let result = loop {
let data = match try_recv(&mut ws_stream, stopped, ping_config).await {
Receive::Shutdown => break Ok(Shutdown::Stopped),
Receive::ConnectionClosed => break Ok(Shutdown::ConnectionClosed),
Receive::Stopped => break Ok(Shutdown::Stopped),
Receive::Ok(data, stop) => {
stopped = stop;
data
Expand Down Expand Up @@ -408,7 +410,8 @@ async fn send_task(
}

enum Receive<S> {
Shutdown,
ConnectionClosed,
Stopped,
Err(SokettoError, S),
Ok(Vec<u8>, S),
}
Expand All @@ -428,7 +431,7 @@ where
loop {
match futures_util::future::select(futs, stopped).await {
// The connection is closed.
Either::Left((Either::Left((None, _)), _)) => break Receive::Shutdown,
Either::Left((Either::Left((None, _)), _)) => break Receive::ConnectionClosed,
// The message has been received, we are done
Either::Left((Either::Left((Some(Ok(Incoming::Data(d))), _)), s)) => break Receive::Ok(d, s),
// Got a pong response, update our "last seen" timestamp.
Expand Down Expand Up @@ -457,7 +460,7 @@ where
futs = futures_util::future::select(rcv, inactivity_check);
}
// Server has been stopped.
Either::Right(_) => break Receive::Shutdown,
Either::Right(_) => break Receive::Stopped,
}
}
}
Expand Down