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

fix(server): remove bounded channel check #1209

Merged
merged 1 commit into from
Sep 28, 2023
Merged

fix(server): remove bounded channel check #1209

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
29 changes: 0 additions & 29 deletions server/src/transport/ws.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -278,19 +278,6 @@ pub(crate) async fn background_task<L: Logger>(sender: Sender, mut receiver: Rec
let result = loop {
data.clear();

// This is a guard to ensure that the underlying socket is only read if there is space in
// the buffer for messages to be sent back to them.
//
// Thus, this check enforces that if the client can't keep up with receiving messages,
// then no new messages will be read from them.
//
// TCP retransmission mechanism will take care of the rest and adjust the window size accordingly.
let Some(stop) = wait_until_connection_buffer_has_capacity(&sink, stopped).await else {
break Ok(Shutdown::ConnectionClosed);
};

stopped = stop;

match try_recv(&mut receiver, &mut data, stopped, ping_config).await {
Receive::Shutdown => break Ok(Shutdown::Stopped),
Receive::Ok(stop) => {
Expand Down Expand Up @@ -410,22 +397,6 @@ enum Receive<S> {
Ok(S),
}

// Wait until there is capacity in connection buffer to send one message.
//
// Fails if the server was stopped.
async fn wait_until_connection_buffer_has_capacity<S>(sink: &MethodSink, stopped: S) -> Option<S>
where
S: Future<Output = ()> + Unpin,
{
let reserve = sink.has_capacity();
tokio::pin!(reserve);

match futures_util::future::select(reserve, stopped).await {
Either::Left((Ok(_), s)) => Some(s),
_ => None,
}
}

/// Attempts to read data from WebSocket fails if the server was stopped.
async fn try_recv<S>(receiver: &mut Receiver, data: &mut Vec<u8>, mut stopped: S, ping_config: PingConfig) -> Receive<S>
where
Expand Down
Loading