Skip to content

Commit

Permalink
fix(server): remove bounded channel check (#1209)
Browse files Browse the repository at this point in the history
This was intended to propogate the backpressure all the way down the
underlying socket but it's weird and doesn't work very well.

For subscriptions the backpressure will be handled by
implementation itself and just rely on that now.
  • Loading branch information
niklasad1 authored Sep 28, 2023
1 parent 9ad7342 commit 08d6a18
Showing 1 changed file with 0 additions and 29 deletions.
29 changes: 0 additions & 29 deletions server/src/transport/ws.rs
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

0 comments on commit 08d6a18

Please sign in to comment.