From 182cced285058c1767516846b3334e7998564c0d Mon Sep 17 00:00:00 2001 From: Lucas Kent Date: Fri, 10 May 2024 08:26:46 +1000 Subject: [PATCH] Port RedisSinkSingle to try_recv --- shotover/src/transforms/redis/sink_single.rs | 25 +++++++------------- 1 file changed, 9 insertions(+), 16 deletions(-) diff --git a/shotover/src/transforms/redis/sink_single.rs b/shotover/src/transforms/redis/sink_single.rs index ccd5b804a..ef90bb2bc 100644 --- a/shotover/src/transforms/redis/sink_single.rs +++ b/shotover/src/transforms/redis/sink_single.rs @@ -131,6 +131,7 @@ impl Transform for RedisSinkSingle { } let mut responses = vec![]; + if requests_wrapper.requests.is_empty() { // there are no requests, so no point sending any, but we should check for any responses without awaiting // TODO: handle errors here @@ -147,31 +148,23 @@ impl Transform for RedisSinkSingle { } } } else { - let requests_count = requests_wrapper.requests.len(); self.connection .as_mut() .unwrap() .send(requests_wrapper.requests)?; - let mut responses_count = 0; - while responses_count < requests_count { - let responses_len_old = responses.len(); - self.connection - .as_mut() - .unwrap() - .recv_into(&mut responses) - .await?; + self.connection + .as_mut() + .unwrap() + .try_recv_into(&mut responses)?; - for response in &mut responses[responses_len_old..] { - if let Some(Frame::Redis(RedisFrame::Error(_))) = response.frame() { - self.failed_requests.increment(1); - } - if response.request_id().is_some() { - responses_count += 1; - } + for response in &mut responses { + if let Some(Frame::Redis(RedisFrame::Error(_))) = response.frame() { + self.failed_requests.increment(1); } } } + Ok(responses) } }