Redis remove pushed messages

shotover-proxy/tests/redis_int_tests/mod.rs

@@ -62,6 +62,7 @@ async fn passthrough_redis_down() {
 
  shotover
  .shutdown_and_then_consume_events(&[
+ // Error occurs when client sends a message to shotover
  EventMatcher::new()
  .with_level(Level::Error)
  .with_target("shotover::server")
@@ -75,6 +76,19 @@ Caused by:
  3: Connection refused (os error 111)"#,
  )
  .with_count(Count::Times(2)),
+ // When the chain is flushed on client connection close we hit the same error again
+ EventMatcher::new()
+ .with_level(Level::Error)
+ .with_target("shotover::server")
+ .with_message(
+ r#"encountered an error when flushing the chain redis for shutdown
+
+Caused by:
+ 0: RedisSinkSingle transform failed
+ 1: Failed to connect to destination "127.0.0.1:1111"
+ 2: Connection refused (os error 111)"#,
+ )
+ .with_count(Count::Times(3)),
  invalid_frame_event(),
  ])
  .await;

shotover/src/server.rs

@@ -444,7 +444,7 @@ async fn spawn_websocket_read_write_tasks<
  );
 }
 
-fn spawn_read_write_tasks<
+pub fn spawn_read_write_tasks<
  C: CodecBuilder + 'static,
  R: AsyncRead + Unpin + Send + 'static,
  W: AsyncWrite + Unpin + Send + 'static,
@@ -455,6 +455,7 @@ fn spawn_read_write_tasks<
  in_tx: mpsc::Sender<Messages>,
  mut out_rx: UnboundedReceiver<Messages>,
  out_tx: UnboundedSender<Messages>,
+ force_run_chain: Option<Arc<Notify>>,
 ) {
  let (decoder, encoder) = codec.build();
  let mut reader = FramedRead::new(rx, decoder);
@@ -487,6 +488,9 @@ fn spawn_read_write_tasks<
  // main task has shutdown, this task is no longer needed
  return;
  }
+ if let Some(force_run_chain) = force_run_chain.as_ref() {
+ force_run_chain.notify_one();
+ }
  }
  Err(CodecReadError::RespondAndThenCloseConnection(messages)) => {
  if let Err(err) = out_tx.send(messages) {
@@ -652,6 +656,7 @@ impl<C: CodecBuilder + 'static> Handler<C> {
  in_tx,
  out_rx,
  out_tx.clone(),
+ None,
  );
  } else {
  let (rx, tx) = stream.into_split();
@@ -662,6 +667,7 @@ impl<C: CodecBuilder + 'static> Handler<C> {
  in_tx,
  out_rx,
  out_tx.clone(),
+ None,
  );
  };
  }