fix(server): remove bounded channel check (#1209)

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.

CHANGELOG.md

@@ -4,6 +4,17 @@ The format is based on [Keep a Changelog].
 
 [Keep a Changelog]: http://keepachangelog.com/en/1.0.0/
 
+## [v0.20.2] - 2023-10-13
+
+This release removes the bounded buffer check which was intended to provide
+backpressure all the way down to the TCP layer but it didn't work well.
+
+For subscriptions the backpressure will be handled by implementation itself 
+and just rely on that.
+
+### [Changed]
+- server: remove bounded channel check ([#1209](https://github.com/paritytech/jsonrpsee/pull/1209))
+
 ## [v0.20.1] - 2023-09-15
 
 This release adds support for `synchronous subscriptions` and fixes a leak in WebSocket server

server/src/transport/ws.rs

@@ -277,19 +277,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 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).await {
 			Receive::Shutdown => break Ok(Shutdown::Stopped),
 			Receive::Ok(stop) => {
@@ -411,22 +398,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>, stopped: S) -> Receive<S>
 where