Repository: Migorithm/duva
Pull request #110 (2024-12-15 15:54:30)

feat: peer information dissemination


Patch:
diff --git a/src/adapters/io/tokio_file.rs b/src/adapters/io/tokio_file.rs
index 811a8dbc..5b6a919e 100644
--- a/src/adapters/io/tokio_file.rs
+++ b/src/adapters/io/tokio_file.rs
@@ -5,7 +5,7 @@ use crate::services::stream_manager::{
 use tokio::io::AsyncWriteExt;
 
 impl TWrite for tokio::fs::File {
-    async fn write_all(&mut self, buf: &[u8]) -> Result<(), IoError> {
+    async fn write(&mut self, buf: &[u8]) -> Result<(), IoError> {
         AsyncWriteExt::write_all(&mut (self as &mut tokio::fs::File), buf)
             .await
             .map_err(|e| e.kind().into())
diff --git a/src/adapters/io/tokio_stream.rs b/src/adapters/io/tokio_stream.rs
index 863a0ea6..7824c257 100644
--- a/src/adapters/io/tokio_stream.rs
+++ b/src/adapters/io/tokio_stream.rs
@@ -3,7 +3,8 @@ use std::io::ErrorKind;
 use crate::{
     services::stream_manager::{
         error::IoError,
-        interface::{TConnectStreamFactory, TGetPeerIp, TRead, TStream, TWrite},
+        interface::{TConnectStreamFactory, TGetPeerIp, TRead, TStream},
+        query_io::{parse, QueryIO},
         PeerAddr,
     },
     TStreamListener, TStreamListenerFactory,
@@ -14,6 +15,49 @@ use tokio::{
     net::TcpListener,
 };
 
+impl TStream for tokio::net::TcpStream {
+    async fn read_value(&mut self) -> anyhow::Result<QueryIO> {
+        let mut buffer = BytesMut::with_capacity(512);
+        self.read_bytes(&mut buffer).await?;
+
+        let (query_io, _) = parse(buffer)?;
+        Ok(query_io)
+    }
+    async fn read_values(&mut self) -> anyhow::Result<Vec<QueryIO>> {
+        let mut buffer = BytesMut::with_capacity(512);
+        self.read_bytes(&mut buffer).await?;
+
+        let mut parsed_values = Vec::new();
+        let mut remaining_buffer = buffer;
+
+        while !remaining_buffer.is_empty() {
+            match parse(remaining_buffer.clone()) {
+                Ok((query_io, consumed)) => {
+                    parsed_values.push(query_io);
+
+                    // * Remove the parsed portion from the buffer
+                    remaining_buffer = remaining_buffer.split_off(consumed);
+                }
+                Err(e) => {
+                    // Handle parsing errors
+                    // You might want to log the error or handle it differently based on your use case
+                    return Err(anyhow::anyhow!("Parsing error: {:?}", e));
+                }
+            }
+        }
+        Ok(parsed_values)
+    }
+
+    async fn write(&mut self, value: QueryIO) -> Result<(), IoError> {
+        self.write_all(value.serialize().as_bytes())
+            .await
+            .map_err(|e| {
+                eprintln!("error = {:?}", e);
+                Into::<IoError>::into(e.kind())
+            })
+    }
+}
+
 impl TRead for tokio::net::TcpStream {
     // TCP doesn't inherently delimit messages.
     // The data arrives in a continuous stream of bytes. And
@@ -44,14 +88,6 @@ impl TRead for tokio::net::TcpStream {
     }
 }
 
-impl TWrite for tokio::net::TcpStream {
-    async fn write_all(&mut self, buf: &[u8]) -> Result<(), IoError> {
-        let mut stream = self as &mut tokio::net::TcpStream;
-        AsyncWriteExt::write_all(&mut stream, buf)
-            .await
-            .map_err(|e| e.kind().into())
-    }
-}
 impl TGetPeerIp for tokio::net::TcpStream {
     fn get_peer_ip(&self) -> Result<String, IoError> {
         let addr = self.peer_addr().map_err(|error| {
@@ -115,3 +151,32 @@ fn test_socket_to_string() {
     //THEN
     assert_eq!(socket.ip().to_string(), "127.0.0.1")
 }
+
+#[tokio::test]
+async fn test_read_values() {
+    let mut buffer = BytesMut::with_capacity(512);
+    // add a simple string to buffer
+    buffer.extend_from_slice(b"+FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0\r\n");
+    buffer.extend_from_slice(b"+PEERS localhost:6378\r\n");
+    // add an integer to buffer
+
+    let mut parsed_values = vec![];
+    while !buffer.is_empty() {
+        if let Ok((query_io, consumed)) = parse(buffer.clone()) {
+            parsed_values.push(query_io);
+
+            // * Remove the parsed portion from the buffer
+            buffer = buffer.split_off(consumed);
+        }
+    }
+
+    assert_eq!(parsed_values.len(), 2);
+    assert_eq!(
+        parsed_values[0],
+        QueryIO::SimpleString("FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0".to_string())
+    );
+    assert_eq!(
+        parsed_values[1],
+        QueryIO::SimpleString("PEERS localhost:6378".to_string())
+    );
+}
diff --git a/src/lib.rs b/src/lib.rs
index b9c9873a..24212217 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -94,7 +94,7 @@ where
         tokio::spawn(async move {
             loop {
                 match replication_listener.listen().await {
-                    Ok((peer_stream, _)) => {
+                    Ok((peer_stream, _socket_addr)) => {
                         let query_manager =
                             StreamManager::new(peer_stream, replication_request_controller);
 
diff --git a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
index 3b8ace14..dfd66920 100644
--- a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
+++ b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
@@ -27,7 +27,7 @@ impl<T: TWrite> SavingProcessor<T> {
             encode_metadata(Vec::from([("redis-ver", "6.0.16")]))?,
             encode_database_info(0)?,
         ];
-        self.writer.write_all(&meta.concat()).await?;
+        self.writer.write(&meta.concat()).await?;
         Ok(())
     }
     pub async fn handle_cmd(&mut self, cmd: SaveCommand) -> Result<bool> {
@@ -41,7 +41,7 @@ impl<T: TWrite> SavingProcessor<T> {
                 self.meta.num_of_saved_table_size_actor -= 1;
                 if self.meta.num_of_saved_table_size_actor == 0 {
                     self.writer
-                        .write_all(&encode_database_table_size(
+                        .write(&encode_database_table_size(
                             self.meta.total_key_value_table_size,
                             self.meta.total_expires_table_size,
                         )?)
@@ -59,7 +59,7 @@ impl<T: TWrite> SavingProcessor<T> {
                 if self.meta.num_of_cache_actors == 0 {
                     self.encode_chunk_queue().await?;
                     let checksum = encode_checksum(&[0; 8])?;
-                    self.writer.write_all(&checksum).await?;
+                    self.writer.write(&checksum).await?;
                     return Ok(true);
                 }
             }
@@ -71,7 +71,7 @@ impl<T: TWrite> SavingProcessor<T> {
         while let Some(chunk) = self.meta.chunk_queue.pop_front() {
             for kvs in chunk {
                 let encoded_chunk = kvs.encode_with_key()?;
-                self.writer.write_all(&encoded_chunk).await?;
+                self.writer.write(&encoded_chunk).await?;
             }
         }
         Ok(())
diff --git a/src/services/stream_manager/error.rs b/src/services/stream_manager/error.rs
index be24e8a9..c3458ef9 100644
--- a/src/services/stream_manager/error.rs
+++ b/src/services/stream_manager/error.rs
@@ -1,3 +1,4 @@
+#[derive(Debug)]
 pub enum IoError {
     ConnectionRefused,
     ConnectionReset,
diff --git a/src/services/stream_manager/interface.rs b/src/services/stream_manager/interface.rs
index b5721ef1..47050726 100644
--- a/src/services/stream_manager/interface.rs
+++ b/src/services/stream_manager/interface.rs
@@ -1,19 +1,26 @@
 use bytes::BytesMut;
 
-use super::{error::IoError, PeerAddr};
+use super::{error::IoError, query_io::QueryIO, PeerAddr};
 
-pub trait TStream: TRead + TWrite + TGetPeerIp {}
-impl<T> TStream for T where T: TRead + TWrite + TGetPeerIp {}
+pub trait TStream: TGetPeerIp + Send + Sync + 'static {
+    // TODO deprecated
+    fn read_value(&mut self) -> impl std::future::Future<Output = anyhow::Result<QueryIO>> + Send;
+    fn read_values(&mut self) -> impl std::future::Future<Output = anyhow::Result<Vec<QueryIO>>>;
+    fn write(
+        &mut self,
+        value: QueryIO,
+    ) -> impl std::future::Future<Output = Result<(), IoError>> + Send;
+}
 
-pub trait TRead: Send + Sync + 'static {
+pub trait TRead {
     fn read_bytes(
         &mut self,
         buf: &mut BytesMut,
     ) -> impl std::future::Future<Output = Result<(), std::io::Error>> + Send;
 }
 
-pub trait TWrite: Send + Sync + 'static {
-    fn write_all(
+pub trait TWrite {
+    fn write(
         &mut self,
         buf: &[u8],
     ) -> impl std::future::Future<Output = Result<(), IoError>> + Send;
@@ -43,7 +50,7 @@ pub trait TStreamListener: Sync + Send + 'static {
     > + Send;
 }
 
-pub trait TGetPeerIp: Send + Sync + 'static {
+pub trait TGetPeerIp {
     fn get_peer_ip(&self) -> Result<String, IoError>;
 }
 
diff --git a/src/services/stream_manager/mod.rs b/src/services/stream_manager/mod.rs
index 6b5d6814..a73a64e6 100644
--- a/src/services/stream_manager/mod.rs
+++ b/src/services/stream_manager/mod.rs
@@ -7,7 +7,6 @@ use crate::services::stream_manager::client_request_controllers::client_request:
 use crate::services::stream_manager::client_request_controllers::ClientRequestController;
 
 use anyhow::Context;
-use bytes::BytesMut;
 use client_request_controllers::arguments::ClientRequestArguments;
 use error::IoError;
 use interface::{
@@ -20,6 +19,7 @@ use replication_request_controllers::{
     arguments::PeerRequestArguments, replication_request::HandShakeRequest,
     ReplicationRequestController,
 };
+use tokio::task::yield_now;
 
 /// Controller is a struct that will be used to read and write values to the client.
 pub struct StreamManager<T, U>
@@ -38,18 +38,8 @@ where
         StreamManager { stream, controller }
     }
 
-    // crlf
-    pub async fn read_value(&mut self) -> anyhow::Result<QueryIO> {
-        let mut buffer = BytesMut::with_capacity(512);
-        self.stream.read_bytes(&mut buffer).await?;
-
-        let (query_io, _) = query_io::parse(buffer)?;
-        Ok(query_io)
-    }
-
-    pub async fn write_value(&mut self, value: QueryIO) -> Result<(), IoError> {
-        self.stream.write_all(value.serialize().as_bytes()).await?;
-        Ok(())
+    pub async fn send(&mut self, value: QueryIO) -> Result<(), IoError> {
+        self.stream.write(value).await
     }
 }
 
@@ -58,9 +48,8 @@ where
     T: TStream,
 {
     async fn extract_query(&mut self) -> anyhow::Result<(ClientRequest, ClientRequestArguments)> {
-        let query_io = self.read_value().await?;
+        let query_io = self.stream.read_value().await?;
         match query_io {
-            // TODO refactor
             QueryIO::Array(value_array) => Ok((
                 value_array
                     .first()
@@ -98,8 +87,8 @@ where
                 .handle::<F>(cancellation_token, request, query_args)
                 .await
             {
-                Ok(response) => self.write_value(response).await,
-                Err(e) => self.write_value(QueryIO::Err(e.to_string())).await,
+                Ok(response) => self.send(response).await,
+                Err(e) => self.send(QueryIO::Err(e.to_string())).await,
             };
 
             if let Err(e) = res {
@@ -111,6 +100,7 @@ where
     }
 }
 
+#[derive(Debug)]
 pub struct PeerAddr(pub String);
 impl<T> StreamManager<T, &'static ReplicationRequestController>
 where
@@ -121,7 +111,7 @@ where
         R: TryFrom<String>,
         anyhow::Error: From<R::Error>,
     {
-        let query_io = self.read_value().await?;
+        let query_io = self.stream.read_value().await?;
         match query_io {
             // TODO refactor
             QueryIO::Array(value_array) => Ok((
@@ -137,12 +127,11 @@ where
         }
     }
     async fn send_simple_string(&mut self, value: &str) -> Result<(), IoError> {
-        self.write_value(QueryIO::SimpleString(value.to_string()))
-            .await
+        self.send(QueryIO::SimpleString(value.to_string())).await
     }
 
     // Temporal coupling: each handling functions inside the following method must be in order
-    async fn establish_threeway_handshake(&mut self) -> anyhow::Result<PeerAddr> {
+    pub async fn establish_threeway_handshake(&mut self) -> anyhow::Result<PeerAddr> {
         self.handle_ping().await?;
 
         let port = self.handle_replconf_listening_port().await?;
@@ -154,6 +143,10 @@ where
         // TODO find use of psync info?
         let (_repl_id, _offset) = self.handle_psync().await?;
 
+        // ! TODO: STRANGE BEHAVIOUR
+        // if not for the following, message is sent with the previosly sent message
+        // even with this, it shows flaking behaviour
+        yield_now().await;
         Ok(PeerAddr(format!("{}:{}", self.stream.get_peer_ip()?, port)))
     }
 
@@ -203,11 +196,12 @@ where
         let peers = self.controller.config_manager.get_peers().await?;
         Ok(peers)
     }
-    async fn disseminate_peers(&mut self) -> anyhow::Result<()> {
+    pub async fn disseminate_peers(&mut self) -> anyhow::Result<()> {
         let peers = self.get_peers().await?;
         if peers.is_empty() {
             return Ok(());
         }
+
         self.send_simple_string(
             format!(
                 "PEERS {}",
@@ -240,11 +234,12 @@ where
         mut self,
         connect_stream_factory: impl TConnectStreamFactory,
     ) -> anyhow::Result<()> {
-        let peer_addr = self.establish_threeway_handshake().await?;
+        let peer_addr = self.establish_threeway_handshake().await.unwrap();
+
+        // Send the peer address to the query manager to be used for replication.
+        self.disseminate_peers().await.unwrap();
 
-        self.disseminate_peers().await?;
-        // TODO the following may be skipped if connection has already been made
-        let out_stream = connect_stream_factory.connect(peer_addr).await?;
+        let out_stream = connect_stream_factory.connect(peer_addr).await;
 
         // Following infinite loop may need to be changed once replica information is given
         loop {
@@ -252,8 +247,8 @@ where
 
             // * Having error from the following will not a concern as liveness concern is on cluster manager
             let _ = match self.controller.handle(request, query_args).await {
-                Ok(response) => self.write_value(response).await,
-                Err(e) => self.write_value(QueryIO::Err(e.to_string())).await,
+                Ok(response) => self.send(response).await,
+                Err(e) => self.send(QueryIO::Err(e.to_string())).await,
             };
         }
     }
diff --git a/src/services/stream_manager/query_io.rs b/src/services/stream_manager/query_io.rs
index d3d45421..a3ee8bc7 100644
--- a/src/services/stream_manager/query_io.rs
+++ b/src/services/stream_manager/query_io.rs
@@ -63,7 +63,7 @@ impl From<String> for QueryIO {
     }
 }
 
-pub(in crate::services) fn parse(buffer: BytesMut) -> Result<(QueryIO, usize)> {
+pub fn parse(buffer: BytesMut) -> Result<(QueryIO, usize)> {
     match buffer[0] as char {
         '+' => parse_simple_string(buffer),
         '*' => parse_array(buffer),
diff --git a/src/services/stream_manager/replication_request_controllers/mod.rs b/src/services/stream_manager/replication_request_controllers/mod.rs
index 2f502fc0..947170dc 100644
--- a/src/services/stream_manager/replication_request_controllers/mod.rs
+++ b/src/services/stream_manager/replication_request_controllers/mod.rs
@@ -22,15 +22,7 @@ impl ReplicationRequestController {
         _args: PeerRequestArguments,
     ) -> anyhow::Result<QueryIO> {
         // handle replication request
-        let response = match request {
-            ReplicationRequest::Ping => {
-                // ! HACK to test out if ping was given from replica
-                self.config_manager
-                    .route_command(ConfigCommand::ReplicaPing)
-                    .await?;
-                QueryIO::SimpleString("PONG".to_string())
-            }
-        };
+        let response = match request {};
         Ok(response)
     }
 }
diff --git a/src/services/stream_manager/replication_request_controllers/replication_request.rs b/src/services/stream_manager/replication_request_controllers/replication_request.rs
index 1ea93f95..18490018 100644
--- a/src/services/stream_manager/replication_request_controllers/replication_request.rs
+++ b/src/services/stream_manager/replication_request_controllers/replication_request.rs
@@ -1,6 +1,4 @@
-pub enum ReplicationRequest {
-    Ping,
-}
+pub enum ReplicationRequest {}
 
 pub enum HandShakeRequest {
     Ping,
@@ -12,7 +10,6 @@ impl TryFrom<String> for ReplicationRequest {
     type Error = anyhow::Error;
     fn try_from(value: String) -> Result<Self, Self::Error> {
         match value.to_lowercase().as_str() {
-            "ping" => Ok(ReplicationRequest::Ping),
             _ => Err(anyhow::anyhow!("Invalid command")),
         }
     }


Test patch:
diff --git a/tests/common.rs b/tests/common.rs
index dc8d7f8d..8bb807b9 100644
--- a/tests/common.rs
+++ b/tests/common.rs
@@ -1,11 +1,12 @@
 use redis_starter_rust::adapters::io::tokio_stream::TokioConnectStreamFactory;
 use redis_starter_rust::services::config::config_manager::ConfigManager;
-use redis_starter_rust::services::stream_manager::interface::TCancellationTokenFactory;
+use redis_starter_rust::services::stream_manager::interface::{TCancellationTokenFactory, TStream};
 use redis_starter_rust::services::stream_manager::query_io::QueryIO;
 use redis_starter_rust::{
     adapters::io::tokio_stream::TokioStreamListenerFactory, services::config::config_actor::Config,
 };
 use redis_starter_rust::{StartUpFacade, TNotifyStartUp};
+use tokio::net::TcpStream;
 
 use std::sync::Arc;
 use tokio::{
@@ -30,12 +31,26 @@ impl<'a> From<(ReadHalf<'a>, WriteHalf<'a>)> for TestStreamHandler<'a> {
 impl<'a> TestStreamHandler<'a> {
     pub async fn send(&mut self, operation: &[u8]) {
         self.write.write_all(operation).await.unwrap();
+        self.write.flush().await.unwrap();
     }
 
+    // read response from the server
     pub async fn get_response(&mut self) -> String {
-        let mut buffer = [0; 1024];
-        let n = self.read.read(&mut buffer).await.unwrap();
-        String::from_utf8(buffer[0..n].to_vec()).unwrap()
+        let mut buffer = Vec::new();
+        let mut temp_buffer = [0; 1024];
+
+        loop {
+            let bytes_read = self.read.read(&mut temp_buffer).await.unwrap();
+            if bytes_read == 0 {
+                break;
+            }
+            buffer.extend_from_slice(&temp_buffer[..bytes_read]);
+            if bytes_read < temp_buffer.len() {
+                break;
+            }
+        }
+
+        String::from_utf8_lossy(&buffer).into_owned()
     }
 }
 
@@ -149,3 +164,51 @@ pub fn get_command(key: &str) -> Vec<u8> {
 pub fn save_command() -> Vec<u8> {
     array(vec!["SAVE"]).into_bytes()
 }
+
+pub async fn threeway_handshake_helper(stream_handler: &mut TcpStream, client_port: u16) {
+    // client sends PING command
+    stream_handler
+        .write_all(b"*1\r\n$4\r\nPING\r\n")
+        .await
+        .unwrap();
+
+    let val = stream_handler.read_values().await.unwrap()[0].clone();
+    assert_eq!(val.serialize(), "+PONG\r\n");
+
+    // client sends REPLCONF listening-port command
+    stream_handler
+        .write_all(
+            format!(
+                "*3\r\n$8\r\nREPLCONF\r\n$14\r\nlistening-port\r\n$4\r\n{}\r\n",
+                client_port
+            )
+            .as_bytes(),
+        )
+        .await
+        .unwrap();
+    let val = stream_handler.read_values().await.unwrap()[0].clone();
+    assert_eq!(val.serialize(), "+OK\r\n");
+
+    // client sends REPLCONF capa command
+    stream_handler
+        .write_all(b"*3\r\n$8\r\nREPLCONF\r\n$4\r\ncapa\r\n$6\r\npsync2\r\n")
+        .await
+        .unwrap();
+    let val = stream_handler.read_values().await.unwrap()[0].clone();
+    assert_eq!(val.serialize(), "+OK\r\n");
+
+    // THEN - client receives OK
+
+    // client sends PSYNC command
+    stream_handler
+        .write_all(b"*3\r\n$5\r\nPSYNC\r\n$1\r\n?\r\n$2\r\n-1\r\n")
+        .await
+        .unwrap();
+    let values = stream_handler.read_values().await.unwrap();
+
+    assert_eq!(
+        values[0].serialize(),
+        "+FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0\r\n"
+    );
+    assert_eq!(values.len(), 1);
+}
diff --git a/tests/test_disseminate_peers.rs b/tests/test_disseminate_peers.rs
index 9b9e1ce8..ab6aeb8c 100644
--- a/tests/test_disseminate_peers.rs
+++ b/tests/test_disseminate_peers.rs
@@ -1,33 +1,38 @@
-/// This is to test if the server successfully disseminates the peers to the client
-/// The client should be able to get the list of peers from the server
+/// After three-way handshake, client will receive peers from the master server
 mod common;
-use common::{find_free_port_in_range, start_test_server, TestStreamHandler};
+use common::{
+    find_free_port_in_range, start_test_server, threeway_handshake_helper, TestStreamHandler,
+};
 use redis_starter_rust::{
     adapters::cancellation_token::CancellationTokenFactory,
-    services::config::{config_actor::Config, config_manager::ConfigManager},
+    services::{
+        config::{config_actor::Config, config_manager::ConfigManager},
+        stream_manager::interface::TStream,
+    },
 };
 use tokio::net::TcpStream;
 
+// TODO Flaky!
 #[tokio::test]
-#[ignore = "not yet ready"]
 async fn test_disseminate_peers() {
     // GIVEN - master server configuration
     let mut config = Config::default();
-    config.peers = vec!["localhost:6378".to_string()];
+    let peer_address_to_test = "localhost:6378";
+    config.peers = vec![peer_address_to_test.to_string()];
     let mut manager = ConfigManager::new(config);
 
     // ! `peer_bind_addr` is bind_addr dedicated for peer connections
     manager.port = find_free_port_in_range(6000, 6553).await.unwrap();
 
     let master_cluster_bind_addr = manager.peer_bind_addr();
-
     let _ = start_test_server(CancellationTokenFactory, manager).await;
 
     let mut client_stream = TcpStream::connect(master_cluster_bind_addr).await.unwrap();
 
-    let mut h: TestStreamHandler = client_stream.split().into();
-    h.send(b"*1\r\n$4\r\nPING\r\n").await;
+    let client_fake_port = 6889;
+    threeway_handshake_helper(&mut client_stream, client_fake_port).await;
 
-    // THEN
-    assert_eq!(h.get_response().await, "+PONG\r\n");
+    //THEN
+    let values = client_stream.read_value().await.unwrap();
+    assert_eq!(values.serialize(), "+PEERS localhost:6378\r\n");
 }
diff --git a/tests/test_handshake.rs b/tests/test_handshake.rs
index 12b5ca3e..42349e06 100644
--- a/tests/test_handshake.rs
+++ b/tests/test_handshake.rs
@@ -66,36 +66,3 @@ async fn test_threeway_handshake() {
         "+FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0\r\n"
     );
 }
-
-pub async fn make_threeway_handshake(stream_handler: &mut TestStreamHandler<'_>, client_port: u16) {
-    // client sends PING command
-    stream_handler.send(b"*1\r\n$4\r\nPING\r\n").await;
-    stream_handler.get_response().await;
-
-    // client sends REPLCONF listening-port command
-    stream_handler
-        .send(
-            format!(
-                "*3\r\n$8\r\nREPLCONF\r\n$14\r\nlistening-port\r\n$4\r\n{}\r\n",
-                client_port
-            )
-            .as_bytes(),
-        )
-        .await;
-
-    stream_handler.get_response().await;
-
-    // client sends REPLCONF capa command
-    stream_handler
-        .send(b"*3\r\n$8\r\nREPLCONF\r\n$4\r\ncapa\r\n$6\r\npsync2\r\n")
-        .await;
-
-    // THEN - client receives OK
-    stream_handler.get_response().await;
-    // client sends PSYNC command
-    stream_handler
-        .send(b"*3\r\n$5\r\nPSYNC\r\n$1\r\n?\r\n$2\r\n-1\r\n")
-        .await;
-
-    stream_handler.get_response().await;
-}