Merge pull request #15 from Miaxos/feat-add-set-name

Feat: Add client setname

app/roster/Cargo.toml

@@ -33,6 +33,7 @@ config.workspace = true
 derive_builder.workspace = true
 dotenv.workspace = true
 futures = "0.3"
+futures-locks = "0.7"
 local-sync = "0.1"
 monoio = { workspace = true, features = ["bytes", "sync", "iouring"] }
 rustc-hash = "1.1.0"

app/roster/src/application/server/cmd/client/list.rs

@@ -105,19 +105,17 @@ impl ClientList {
         let connections = ctx.supervisor.get_normal_connection().await;
 
         // TODO(@miaxos): lot of things missing here
-        let conn_frames = connections
-            .into_iter()
-            .map(|x| {
-                ByteString::from(format!(
-                    "id={id} addr={addr} laddr={laddr} fd={fd}",
-                    id = &x.id,
-                    addr = &x.addr,
-                    laddr = &x.laddr,
-                    fd = &x.fd
-                ))
-            })
-            .map(Frame::Simple)
-            .collect();
+        let mut conn_frames = Vec::with_capacity(connections.len());
+        for conn in connections {
+            conn_frames.push(Frame::Simple(ByteString::from(format!(
+                "id={id} addr={addr} laddr={laddr} fd={fd} name={name}",
+                id = &conn.id,
+                addr = &conn.addr,
+                laddr = &conn.laddr,
+                fd = &conn.fd,
+                name = &conn.name().await.unwrap_or(ByteString::new()),
+            ))));
+        }
 
         let response = Frame::Array(conn_frames);
         dst.write_frame(&response).await?;

app/roster/src/application/server/cmd/client/mod.rs

@@ -9,11 +9,13 @@ use crate::application::server::frame::Frame;
 mod id;
 mod list;
 mod set_info;
+mod set_name;
 
 #[derive(Debug)]
 pub enum Client {
     Help,
     SetInfo(set_info::ClientSetInfo),
+    SetName(set_name::ClientSetName),
     List(list::ClientList),
     Id(id::ClientID),
 }
@@ -28,6 +30,8 @@ LIST [options ...]
     Return information about client connections. Options:
     * TYPE (NORMAL|MASTER|REPLICA|PUBSUB)
       Return clients of specified type.
+SETNAME <name>
+    Assign the name <name> to the current connection.
 SETINFO <option> <value>
     Set client meta attr. Options are:
     * LIB-NAME: the client lib name.
@@ -54,6 +58,9 @@ impl SubcommandRegistry for Client {
             "setinfo" => Command::Client(Client::SetInfo(
                 set_info::ClientSetInfo::parse_frames(&mut parse)?,
             )),
+            "setname" => Command::Client(Client::SetName(
+                set_name::ClientSetName::parse_frames(&mut parse)?,
+            )),
             "id" => Command::Client(Client::Id(id::ClientID::parse_frames(
                 &mut parse,
             )?)),
@@ -105,6 +112,7 @@ impl CommandExecution for Client {
         match self {
             Client::Help => Client::help(dst, ctx).await,
             Client::SetInfo(cmd) => cmd.apply(dst, ctx).await,
+            Client::SetName(cmd) => cmd.apply(dst, ctx).await,
             Client::Id(cmd) => cmd.apply(dst, ctx).await,
             Client::List(cmd) => cmd.apply(dst, ctx).await,
         }

app/roster/src/application/server/cmd/client/set_info.rs

@@ -1,5 +1,3 @@
-use std::time::Duration;
-
 use bytes::Bytes;
 use bytestring::ByteString;
 
@@ -102,8 +100,6 @@ impl ClientSetInfo {
         dst: &mut WriteConnection,
         _ctx: Context,
     ) -> anyhow::Result<()> {
-        monoio::time::sleep(Duration::from_secs(3)).await;
-
         let response = Frame::Simple(ByteString::from_static("OK"));
         dst.write_frame(&response).await?;
 

app/roster/src/application/server/cmd/client/set_name.rs

@@ -0,0 +1,62 @@
+use bytestring::ByteString;
+
+use super::super::parse::Parse;
+use crate::application::server::connection::WriteConnection;
+use crate::application::server::context::Context;
+use crate::application::server::frame::Frame;
+
+/// The CLIENT SETNAME command assigns a name to the current connection.
+///
+/// The assigned name is displayed in the output of CLIENT LIST so that it is
+/// possible to identify the client that performed a given connection.
+///
+/// For instance when Redis is used in order to implement a queue, producers and
+/// consumers of messages may want to set the name of the connection according
+/// to their role.
+///
+/// There is no limit to the length of the name that can be assigned if not the
+/// usual limits of the Redis string type (512 MB). However it is not possible
+/// to use spaces in the connection name as this would violate the format of the
+/// CLIENT LIST reply.
+///
+/// It is possible to entirely remove the connection name setting it to the
+/// empty string, that is not a valid connection name since it serves to this
+/// specific purpose.
+///
+/// The connection name can be inspected using CLIENT GETNAME.
+///
+/// Every new connection starts without an assigned name.
+///
+/// Tip: setting names to connections is a good way to debug connection leaks
+/// due to bugs in the application using Redis.
+#[derive(Debug, Default)]
+pub struct ClientSetName {
+    name: ByteString,
+}
+
+impl ClientSetName {
+    /// Create a new `ClientSetName` command with optional `msg`.
+    pub fn new(name: ByteString) -> ClientSetName {
+        ClientSetName { name }
+    }
+
+    pub(crate) fn parse_frames(
+        parse: &mut Parse,
+    ) -> anyhow::Result<ClientSetName> {
+        let name = parse.next_string()?;
+        Ok(ClientSetName::new(name))
+    }
+
+    pub(crate) async fn apply(
+        self,
+        dst: &mut WriteConnection,
+        ctx: Context,
+    ) -> anyhow::Result<()> {
+        ctx.connection.set_name(self.name).await;
+
+        let response = Frame::Simple(ByteString::from_static("OK"));
+        dst.write_frame(&response).await?;
+
+        Ok(())
+    }
+}

app/roster/src/application/server/supervisor/mod.rs

@@ -2,6 +2,8 @@ use std::net::SocketAddr;
 use std::sync::atomic::{AtomicBool, AtomicU64};
 use std::sync::Arc;
 
+use bytestring::ByteString;
+use futures_locks::RwLock;
 use scc::HashMap;
 
 /// [Supervisor] is the Applicative layer that allow you to interact with the
@@ -51,6 +53,7 @@ impl Supervisor {
             id,
             kind: MetadataConnectionKind::Normal,
             stopped: AtomicBool::new(false),
+            name: RwLock::new(None),
             addr,
             laddr,
             fd,
@@ -92,6 +95,8 @@ pub struct MetadataConnection {
     pub id: u64,
     /// Describe the connection kind
     pub kind: MetadataConnectionKind,
+    /// the name set by the client with CLIENT SETNAME
+    name: RwLock<Option<ByteString>>,
     /// Tell if the connection is stopped
     pub stopped: AtomicBool,
     /// Address/Port of the client
@@ -108,6 +113,17 @@ impl MetadataConnection {
         self.stopped
             .store(true, std::sync::atomic::Ordering::Relaxed);
     }
+
+    /// Set the name of the connection
+    pub async fn set_name(&self, name: ByteString) {
+        let mut lock = self.name.write().await;
+        *lock = Some(name);
+    }
+
+    /// Set the name of the connection
+    pub async fn name(&self) -> Option<ByteString> {
+        self.name.read().await.clone()
+    }
 }
 
 impl MetadataConnection {

app/roster/tests/client.rs

@@ -37,34 +37,8 @@ pub async fn test_simple_client_help() {
         Return information about client connections. Options:
         * TYPE (NORMAL|MASTER|REPLICA|PUBSUB)
           Return clients of specified type.
-    SETINFO <option> <value>
-        Set client meta attr. Options are:
-        * LIB-NAME: the client lib name.
-        * LIB-VER: the client lib version.
-    HELP
-        Print this help.
-    "###);
-}
-
-#[tokio::test]
-pub async fn test_simple_client() {
-    let addr = utils::start_simple_server();
-
-    let connection = utils::connect_without_auth(addr).await;
-
-    let res_f: Vec<String> =
-        connection.send(resp_array!["CLIENT"]).await.unwrap();
-
-    let joined = res_f.join("\n");
-
-    insta::assert_display_snapshot!(joined, @r###"
-    CLIENT <subcommand> [<arg> [value] [opt] ...]. subcommands are:
-    ID
-        Return the ID of the current connection.
-    LIST [options ...]
-        Return information about client connections. Options:
-        * TYPE (NORMAL|MASTER|REPLICA|PUBSUB)
-          Return clients of specified type.
+    SETNAME <name>
+        Assign the name <name> to the current connection.
     SETINFO <option> <value>
         Set client meta attr. Options are:
         * LIB-NAME: the client lib name.

app/roster/tests/client_list.rs

@@ -5,7 +5,7 @@ use regex::Regex;
 #[tokio::test]
 pub async fn client_list() {
     let test_re: Regex =
-        Regex::new(r"^id=0 addr=.*? laddr=.*? fd=.*$").unwrap();
+        Regex::new(r"^id=0 addr=.*? laddr=.*? fd=.*? name=$").unwrap();
     let addr = utils::start_simple_server();
 
     let connection = utils::connect_without_auth(addr).await;

app/roster/tests/client_setname.rs

@@ -0,0 +1,39 @@
+mod utils;
+use redis_async::resp_array;
+use regex::Regex;
+
+#[tokio::test]
+pub async fn client_setname() {
+    let test_re: Regex =
+        Regex::new(r"^id=0 addr=.*? laddr=.*? fd=.*? name=$").unwrap();
+    let addr = utils::start_simple_server();
+
+    let connection = utils::connect_without_auth(addr).await;
+
+    let mut res_f: Vec<String> = connection
+        .send(resp_array!["CLIENT", "LIST"])
+        .await
+        .unwrap();
+
+    assert_eq!(res_f.len(), 1);
+    let first_value = res_f.pop().unwrap();
+    assert!(test_re.is_match(&first_value));
+
+    let res_ok: String = connection
+        .send(resp_array!["CLIENT", "SETNAME", "newname"])
+        .await
+        .unwrap();
+
+    assert_eq!(res_ok, "OK");
+
+    let mut res_f: Vec<String> = connection
+        .send(resp_array!["CLIENT", "LIST"])
+        .await
+        .unwrap();
+
+    let test_re: Regex =
+        Regex::new(r"^id=0 addr=.*? laddr=.*? fd=.*? name=newname$").unwrap();
+    assert_eq!(res_f.len(), 1);
+    let first_value = res_f.pop().unwrap();
+    assert!(test_re.is_match(&first_value));
+}

docs/cmd_list.md

@@ -52,8 +52,8 @@ the command or open an issue.
 - [ ] CLIENT NO TOUCH
 - [ ] CLIENT PAUSE
 - [ ] CLIENT REPLY
-- [ ] CLIENT SETINFO
-- [ ] CLIENT SETNAME
+- [x] CLIENT SETINFO
+- [x] CLIENT SETNAME
 - [ ] CLIENT TRACKING
 - [ ] CLIENT TRACKINGINFO
 - [ ] CLIENT UNBLOCK