diff --git a/crates/ember-client/Cargo.toml b/crates/ember-client/Cargo.toml index 17ae56ce..0a82db6f 100644 --- a/crates/ember-client/Cargo.toml +++ b/crates/ember-client/Cargo.toml @@ -30,3 +30,6 @@ tls = [ "dep:rustls-pki-types", "dep:rustls-native-certs", ] +# Vector set commands (VADD, VSIM, etc.). Requires the server to be built +# with the same feature enabled. +vector = [] diff --git a/crates/ember-client/src/commands.rs b/crates/ember-client/src/commands.rs index 5761f1e2..a157d459 100644 --- a/crates/ember-client/src/commands.rs +++ b/crates/ember-client/src/commands.rs @@ -16,6 +16,33 @@ use ember_protocol::types::Frame; use crate::connection::{Client, ClientError}; use crate::pipeline::Pipeline; +use crate::subscriber::Subscriber; + +// --- public types --- + +/// A page of keys returned by [`Client::scan`]. +/// +/// Iterate until `cursor` is `0` to walk the full keyspace. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ScanPage { + /// Cursor for the next call. `0` means iteration is complete. + pub cursor: u64, + /// Keys returned in this page. + pub keys: Vec, +} + +/// A single entry from [`Client::slowlog_get`]. +#[derive(Debug, Clone)] +pub struct SlowlogEntry { + /// Monotonically increasing log entry ID. + pub id: i64, + /// Unix timestamp (seconds) when the command was logged. + pub timestamp: i64, + /// Execution time in microseconds. + pub duration_us: i64, + /// The command and its arguments as raw bytes. + pub command: Vec, +} // --- frame construction helpers --- @@ -282,6 +309,213 @@ fn scored_members(frame: Frame) -> Result, ClientError> { Ok(result) } +/// Decodes a simple string or bulk string response into a `String`. +/// +/// Used by: TYPE, INFO, ECHO, BGSAVE, BGREWRITEAOF. +fn string_value(frame: Frame) -> Result { + match frame { + Frame::Simple(s) => Ok(s), + Frame::Bulk(b) => String::from_utf8(b.to_vec()) + .map_err(|_| ClientError::Protocol("response is not valid UTF-8".into())), + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected simple or bulk string, got {other:?}" + ))), + } +} + +/// Decodes a float returned as a bulk string. +/// +/// Used by: INCRBYFLOAT. +fn float_value(frame: Frame) -> Result { + match frame { + Frame::Bulk(b) => { + let s = std::str::from_utf8(&b) + .map_err(|_| ClientError::Protocol("float response is not valid UTF-8".into()))?; + s.parse::() + .map_err(|_| ClientError::Protocol(format!("not a valid float: {s:?}"))) + } + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected bulk float, got {other:?}" + ))), + } +} + +/// Decodes a SCAN response into a `ScanPage`. +/// +/// RESP3 layout: `Array([Bulk(cursor), Array([Bulk(key), ...])])`. +fn scan_page(frame: Frame) -> Result { + let elems = match frame { + Frame::Array(e) => e, + Frame::Error(e) => return Err(ClientError::Server(e)), + other => { + return Err(ClientError::Protocol(format!( + "expected array for SCAN, got {other:?}" + ))) + } + }; + + if elems.len() != 2 { + return Err(ClientError::Protocol(format!( + "SCAN response must have 2 elements, got {}", + elems.len() + ))); + } + + let mut iter = elems.into_iter(); + let cursor_frame = iter.next().unwrap(); + let keys_frame = iter.next().unwrap(); + + let cursor = match cursor_frame { + Frame::Bulk(b) => { + let s = std::str::from_utf8(&b) + .map_err(|_| ClientError::Protocol("SCAN cursor is not valid UTF-8".into()))?; + s.parse::() + .map_err(|_| ClientError::Protocol(format!("SCAN cursor is not a u64: {s:?}")))? + } + other => { + return Err(ClientError::Protocol(format!( + "expected bulk cursor in SCAN, got {other:?}" + ))) + } + }; + + let keys = bytes_vec(keys_frame)?; + Ok(ScanPage { cursor, keys }) +} + +/// Decodes a SLOWLOG GET response into a list of entries. +/// +/// Each entry: `Array([Integer(id), Integer(ts), Integer(us), Array([Bulk(cmd), ...])])`. +fn slowlog_entries(frame: Frame) -> Result, ClientError> { + let outer = match frame { + Frame::Array(e) => e, + Frame::Null => return Ok(Vec::new()), + Frame::Error(e) => return Err(ClientError::Server(e)), + other => { + return Err(ClientError::Protocol(format!( + "expected array for SLOWLOG, got {other:?}" + ))) + } + }; + + outer + .into_iter() + .map(|entry_frame| { + let entry = match entry_frame { + Frame::Array(e) => e, + other => { + return Err(ClientError::Protocol(format!( + "expected array for slowlog entry, got {other:?}" + ))) + } + }; + + if entry.len() < 4 { + return Err(ClientError::Protocol(format!( + "slowlog entry too short: {} elements", + entry.len() + ))); + } + + let id = match &entry[0] { + Frame::Integer(n) => *n, + other => { + return Err(ClientError::Protocol(format!( + "expected integer id in slowlog, got {other:?}" + ))) + } + }; + let timestamp = match &entry[1] { + Frame::Integer(n) => *n, + other => { + return Err(ClientError::Protocol(format!( + "expected integer timestamp in slowlog, got {other:?}" + ))) + } + }; + let duration_us = match &entry[2] { + Frame::Integer(n) => *n, + other => { + return Err(ClientError::Protocol(format!( + "expected integer duration in slowlog, got {other:?}" + ))) + } + }; + let command = match entry.into_iter().nth(3).unwrap() { + Frame::Array(parts) => parts + .into_iter() + .map(|p| match p { + Frame::Bulk(b) => Ok(b), + other => Err(ClientError::Protocol(format!( + "expected bulk in slowlog command, got {other:?}" + ))), + }) + .collect::, _>>()?, + other => { + return Err(ClientError::Protocol(format!( + "expected array for slowlog command, got {other:?}" + ))) + } + }; + + Ok(SlowlogEntry { + id, + timestamp, + duration_us, + command, + }) + }) + .collect() +} + +/// Decodes a PUBSUB NUMSUB response into channel/count pairs. +/// +/// Layout: `Array([Bulk(channel), Integer(count), ...])` — alternating. +fn numsub_pairs(frame: Frame) -> Result, ClientError> { + let elems = match frame { + Frame::Array(e) => e, + Frame::Null => return Ok(Vec::new()), + Frame::Error(e) => return Err(ClientError::Server(e)), + other => { + return Err(ClientError::Protocol(format!( + "expected array for PUBSUB NUMSUB, got {other:?}" + ))) + } + }; + + if elems.len() % 2 != 0 { + return Err(ClientError::Protocol(format!( + "PUBSUB NUMSUB array has odd length ({})", + elems.len() + ))); + } + + let mut result = Vec::with_capacity(elems.len() / 2); + let mut iter = elems.into_iter(); + while let (Some(ch_frame), Some(cnt_frame)) = (iter.next(), iter.next()) { + let channel = match ch_frame { + Frame::Bulk(b) => b, + other => { + return Err(ClientError::Protocol(format!( + "expected bulk channel in PUBSUB NUMSUB, got {other:?}" + ))) + } + }; + let count = match cnt_frame { + Frame::Integer(n) => n, + other => { + return Err(ClientError::Protocol(format!( + "expected integer count in PUBSUB NUMSUB, got {other:?}" + ))) + } + }; + result.push((channel, count)); + } + Ok(result) +} + // --- typed command API --- impl Client { @@ -728,6 +962,321 @@ impl Client { ok(frame) } + // --- more string commands --- + + /// Returns the length of the string at `key`. Returns 0 for missing keys. + pub async fn strlen(&mut self, key: &str) -> Result { + let frame = self.send_frame(cmd2(b"STRLEN", key.as_bytes())).await?; + integer(frame) + } + + /// Increments the float stored at `key` by `delta`. Returns the new value. + pub async fn incr_by_float(&mut self, key: &str, delta: f64) -> Result { + let d = delta.to_string(); + let frame = self + .send_frame(cmd3(b"INCRBYFLOAT", key.as_bytes(), d.as_bytes())) + .await?; + float_value(frame) + } + + // --- key commands --- + + /// Returns the type of the value stored at `key` as a string: + /// `"string"`, `"list"`, `"set"`, `"zset"`, `"hash"`, or `"none"`. + pub async fn key_type(&mut self, key: &str) -> Result { + let frame = self.send_frame(cmd2(b"TYPE", key.as_bytes())).await?; + string_value(frame) + } + + /// Returns all keys matching `pattern`. + /// + /// Use `"*"` to return every key. This is a blocking O(N) scan — prefer + /// [`Client::scan`] in production. + pub async fn keys(&mut self, pattern: &str) -> Result, ClientError> { + let frame = self.send_frame(cmd2(b"KEYS", pattern.as_bytes())).await?; + bytes_vec(frame) + } + + /// Renames `key` to `newkey`. Returns an error if `key` does not exist. + pub async fn rename(&mut self, key: &str, newkey: &str) -> Result<(), ClientError> { + let frame = self + .send_frame(cmd3(b"RENAME", key.as_bytes(), newkey.as_bytes())) + .await?; + ok(frame) + } + + /// Incrementally iterates keys in the keyspace. + /// + /// Pass `cursor: 0` to start a new iteration. Continue calling with the + /// returned cursor until the cursor is `0` again. An optional `pattern` + /// filters by glob and `count` hints at the page size (server may return + /// more or fewer). + pub async fn scan( + &mut self, + cursor: u64, + count: Option, + pattern: Option<&str>, + ) -> Result { + let cur = cursor.to_string(); + let mut parts = Vec::with_capacity(6); + parts.push(Frame::Bulk(Bytes::from_static(b"SCAN"))); + parts.push(Frame::Bulk(Bytes::copy_from_slice(cur.as_bytes()))); + if let Some(pat) = pattern { + parts.push(Frame::Bulk(Bytes::from_static(b"MATCH"))); + parts.push(Frame::Bulk(Bytes::copy_from_slice(pat.as_bytes()))); + } + if let Some(n) = count { + let ns = n.to_string(); + parts.push(Frame::Bulk(Bytes::from_static(b"COUNT"))); + parts.push(Frame::Bulk(Bytes::copy_from_slice(ns.as_bytes()))); + } + let frame = self.send_frame(Frame::Array(parts)).await?; + scan_page(frame) + } + + /// Sets a timeout of `millis` milliseconds on `key`. Returns `true` if + /// the timeout was set, `false` if the key does not exist. + pub async fn pexpire(&mut self, key: &str, millis: u64) -> Result { + let ms = millis.to_string(); + let frame = self + .send_frame(cmd3(b"PEXPIRE", key.as_bytes(), ms.as_bytes())) + .await?; + bool_flag(frame) + } + + // --- more hash commands --- + + /// Returns values for multiple `fields` in the hash at `key`. Missing + /// fields are `None`. + pub async fn hmget( + &mut self, + key: &str, + fields: &[&str], + ) -> Result>, ClientError> { + let frame = self + .send_frame(cmd_key_and_keys(b"HMGET", key, fields)) + .await?; + optional_bytes_vec(frame) + } + + // --- more server commands --- + + /// Echoes `message` back from the server. Useful for round-trip testing. + pub async fn echo(&mut self, message: &str) -> Result { + let frame = self.send_frame(cmd2(b"ECHO", message.as_bytes())).await?; + match frame { + Frame::Bulk(b) => Ok(b), + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected bulk for ECHO, got {other:?}" + ))), + } + } + + /// Deletes keys asynchronously. Behaves like `del` but frees memory in + /// the background for large values. Returns the number of keys removed. + pub async fn unlink(&mut self, keys: &[&str]) -> Result { + let frame = self.send_frame(cmd_keys_only(b"UNLINK", keys)).await?; + integer(frame) + } + + /// Returns server information. Pass `Some("keyspace")` for a specific + /// section, or `None` for all sections. + pub async fn info(&mut self, section: Option<&str>) -> Result { + let frame = match section { + Some(s) => self.send_frame(cmd2(b"INFO", s.as_bytes())).await?, + None => { + self.send_frame(Frame::Array(vec![Frame::Bulk(Bytes::from_static(b"INFO"))])) + .await? + } + }; + match frame { + Frame::Bulk(b) => String::from_utf8(b.to_vec()) + .map_err(|_| ClientError::Protocol("INFO response is not valid UTF-8".into())), + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected bulk for INFO, got {other:?}" + ))), + } + } + + /// Triggers a background snapshot (`BGSAVE`). Returns the server status + /// message. + pub async fn bgsave(&mut self) -> Result { + let frame = self + .send_frame(Frame::Array(vec![Frame::Bulk(Bytes::from_static( + b"BGSAVE", + ))])) + .await?; + string_value(frame) + } + + /// Triggers an AOF rewrite in the background (`BGREWRITEAOF`). Returns + /// the server status message. + pub async fn bgrewriteaof(&mut self) -> Result { + let frame = self + .send_frame(Frame::Array(vec![Frame::Bulk(Bytes::from_static( + b"BGREWRITEAOF", + ))])) + .await?; + string_value(frame) + } + + // --- slowlog commands --- + + /// Returns up to `count` recent slow-log entries, or all entries if + /// `count` is `None`. + pub async fn slowlog_get( + &mut self, + count: Option, + ) -> Result, ClientError> { + let frame = match count { + Some(n) => { + let ns = n.to_string(); + self.send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"SLOWLOG")), + Frame::Bulk(Bytes::from_static(b"GET")), + Frame::Bulk(Bytes::copy_from_slice(ns.as_bytes())), + ])) + .await? + } + None => { + self.send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"SLOWLOG")), + Frame::Bulk(Bytes::from_static(b"GET")), + ])) + .await? + } + }; + slowlog_entries(frame) + } + + /// Returns the number of entries currently in the slow log. + pub async fn slowlog_len(&mut self) -> Result { + let frame = self + .send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"SLOWLOG")), + Frame::Bulk(Bytes::from_static(b"LEN")), + ])) + .await?; + integer(frame) + } + + /// Clears all entries from the slow log. + pub async fn slowlog_reset(&mut self) -> Result<(), ClientError> { + let frame = self + .send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"SLOWLOG")), + Frame::Bulk(Bytes::from_static(b"RESET")), + ])) + .await?; + ok(frame) + } + + // --- pub/sub commands (request-response only) --- + + /// Publishes `message` to `channel`. Returns the number of subscribers + /// that received the message. + pub async fn publish( + &mut self, + channel: &str, + message: impl AsRef<[u8]>, + ) -> Result { + let frame = self + .send_frame(cmd3(b"PUBLISH", channel.as_bytes(), message.as_ref())) + .await?; + integer(frame) + } + + /// Returns the names of active pub/sub channels. Pass `Some(pattern)` to + /// filter by glob, or `None` for all channels. + pub async fn pubsub_channels( + &mut self, + pattern: Option<&str>, + ) -> Result, ClientError> { + let frame = match pattern { + Some(p) => { + self.send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"PUBSUB")), + Frame::Bulk(Bytes::from_static(b"CHANNELS")), + Frame::Bulk(Bytes::copy_from_slice(p.as_bytes())), + ])) + .await? + } + None => { + self.send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"PUBSUB")), + Frame::Bulk(Bytes::from_static(b"CHANNELS")), + ])) + .await? + } + }; + bytes_vec(frame) + } + + /// Returns the subscriber counts for the given channels as + /// `(channel, count)` pairs. + pub async fn pubsub_numsub( + &mut self, + channels: &[&str], + ) -> Result, ClientError> { + let mut parts = Vec::with_capacity(2 + channels.len()); + parts.push(Frame::Bulk(Bytes::from_static(b"PUBSUB"))); + parts.push(Frame::Bulk(Bytes::from_static(b"NUMSUB"))); + for ch in channels { + parts.push(Frame::Bulk(Bytes::copy_from_slice(ch.as_bytes()))); + } + let frame = self.send_frame(Frame::Array(parts)).await?; + numsub_pairs(frame) + } + + /// Returns the number of active pattern subscriptions across all clients. + pub async fn pubsub_numpat(&mut self) -> Result { + let frame = self + .send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"PUBSUB")), + Frame::Bulk(Bytes::from_static(b"NUMPAT")), + ])) + .await?; + integer(frame) + } + + // --- pub/sub subscriber mode --- + + /// Puts the connection into subscriber mode on `channels`. + /// + /// The connection is consumed and a [`Subscriber`] is returned. Use a + /// separate [`Client`] for regular commands while subscribed. + pub async fn subscribe(mut self, channels: &[&str]) -> Result { + let mut parts = Vec::with_capacity(1 + channels.len()); + parts.push(Frame::Bulk(Bytes::from_static(b"SUBSCRIBE"))); + for ch in channels { + parts.push(Frame::Bulk(Bytes::copy_from_slice(ch.as_bytes()))); + } + self.write_frame(Frame::Array(parts)).await?; + // drain confirmation frames (one per channel) + for _ in 0..channels.len() { + self.read_response().await?; + } + Ok(Subscriber::new(self)) + } + + /// Same as [`subscribe`](Client::subscribe) but subscribes to glob + /// patterns with `PSUBSCRIBE`. + pub async fn psubscribe(mut self, patterns: &[&str]) -> Result { + let mut parts = Vec::with_capacity(1 + patterns.len()); + parts.push(Frame::Bulk(Bytes::from_static(b"PSUBSCRIBE"))); + for p in patterns { + parts.push(Frame::Bulk(Bytes::copy_from_slice(p.as_bytes()))); + } + self.write_frame(Frame::Array(parts)).await?; + // drain confirmation frames (one per pattern) + for _ in 0..patterns.len() { + self.read_response().await?; + } + Ok(Subscriber::new(self)) + } + // --- pipeline --- /// Executes all commands queued in `pipeline` as a single batch. @@ -772,6 +1321,16 @@ fn cmd_key_values>(cmd: &'static [u8], key: &str, values: &[V]) - Frame::Array(parts) } +/// `CMD key1 key2 ...` (no separate leading key argument) +fn cmd_keys_only(cmd: &'static [u8], keys: &[&str]) -> Frame { + let mut parts = Vec::with_capacity(1 + keys.len()); + parts.push(Frame::Bulk(Bytes::from_static(cmd))); + for k in keys { + parts.push(Frame::Bulk(Bytes::copy_from_slice(k.as_bytes()))); + } + Frame::Array(parts) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/ember-client/src/connection.rs b/crates/ember-client/src/connection.rs index d01a3c67..09fdb551 100644 --- a/crates/ember-client/src/connection.rs +++ b/crates/ember-client/src/connection.rs @@ -242,8 +242,20 @@ impl Client { let _ = self.transport.shutdown().await; } + /// Serializes and writes a single frame without waiting for a response. + /// + /// Used by [`Subscriber`] to send SUBSCRIBE/UNSUBSCRIBE frames before + /// draining the confirmation responses separately. + pub(crate) async fn write_frame(&mut self, frame: Frame) -> Result<(), ClientError> { + self.write_buf.clear(); + frame.serialize(&mut self.write_buf); + self.transport.write_all(&self.write_buf).await?; + self.transport.flush().await?; + Ok(()) + } + /// Reads a complete RESP3 frame from the server. - async fn read_response(&mut self) -> Result { + pub(crate) async fn read_response(&mut self) -> Result { loop { if !self.read_buf.is_empty() { match parse_frame(&self.read_buf) { diff --git a/crates/ember-client/src/lib.rs b/crates/ember-client/src/lib.rs index a729d466..178f3c6c 100644 --- a/crates/ember-client/src/lib.rs +++ b/crates/ember-client/src/lib.rs @@ -48,11 +48,18 @@ mod commands; mod connection; mod pipeline; +pub mod subscriber; #[cfg(feature = "tls")] pub mod tls; +#[cfg(feature = "vector")] +pub mod vector; +pub use commands::{ScanPage, SlowlogEntry}; pub use connection::{Client, ClientError}; pub use ember_protocol::types::Frame; pub use pipeline::Pipeline; +pub use subscriber::{Message, Subscriber}; #[cfg(feature = "tls")] pub use tls::TlsClientConfig; +#[cfg(feature = "vector")] +pub use vector::SimResult; diff --git a/crates/ember-client/src/pipeline.rs b/crates/ember-client/src/pipeline.rs index 0346d40d..4a3f4039 100644 --- a/crates/ember-client/src/pipeline.rs +++ b/crates/ember-client/src/pipeline.rs @@ -210,6 +210,66 @@ impl Pipeline { self.push(array3(b"ZSCORE", key.as_bytes(), member.as_bytes())) } + // --- more string commands --- + + /// Queues a `STRLEN key` command. + pub fn strlen(self, key: &str) -> Self { + self.push(array2(b"STRLEN", key.as_bytes())) + } + + /// Queues an `INCRBYFLOAT key delta` command. + pub fn incr_by_float(self, key: &str, delta: f64) -> Self { + let d = delta.to_string(); + self.push(array3(b"INCRBYFLOAT", key.as_bytes(), d.as_bytes())) + } + + // --- key commands --- + + /// Queues a `TYPE key` command. + pub fn key_type(self, key: &str) -> Self { + self.push(array2(b"TYPE", key.as_bytes())) + } + + /// Queues a `KEYS pattern` command. + pub fn keys(self, pattern: &str) -> Self { + self.push(array2(b"KEYS", pattern.as_bytes())) + } + + /// Queues a `RENAME key newkey` command. + pub fn rename(self, key: &str, newkey: &str) -> Self { + self.push(array3(b"RENAME", key.as_bytes(), newkey.as_bytes())) + } + + /// Queues a `PEXPIRE key millis` command. + pub fn pexpire(self, key: &str, millis: u64) -> Self { + let ms = millis.to_string(); + self.push(array3(b"PEXPIRE", key.as_bytes(), ms.as_bytes())) + } + + /// Queues an `UNLINK key [key ...]` command. + pub fn unlink(self, keys: &[&str]) -> Self { + self.push(array_with_keys(b"UNLINK", keys)) + } + + // --- more hash commands --- + + /// Queues an `HMGET key field [field ...]` command. + pub fn hmget(self, key: &str, fields: &[&str]) -> Self { + self.push(array_with_key_and_keys(b"HMGET", key, fields)) + } + + // --- server commands --- + + /// Queues an `ECHO message` command. + pub fn echo(self, message: &str) -> Self { + self.push(array2(b"ECHO", message.as_bytes())) + } + + /// Queues a `PUBLISH channel message` command. + pub fn publish(self, channel: &str, message: impl AsRef<[u8]>) -> Self { + self.push(array3(b"PUBLISH", channel.as_bytes(), message.as_ref())) + } + // --- internal --- fn push(mut self, frame: Frame) -> Self { diff --git a/crates/ember-client/src/subscriber.rs b/crates/ember-client/src/subscriber.rs new file mode 100644 index 00000000..7f7a714a --- /dev/null +++ b/crates/ember-client/src/subscriber.rs @@ -0,0 +1,185 @@ +//! Pub/sub subscriber mode. +//! +//! When a connection issues [`Client::subscribe`] or [`Client::psubscribe`] it +//! enters pub/sub mode: the server will push [`Message`] frames whenever a +//! matching publish event occurs. Normal request-response commands cannot be +//! issued on the same connection while it is in sub mode. +//! +//! Use a separate [`Client`] for regular commands while subscribed. + +use bytes::Bytes; +use ember_protocol::types::Frame; + +use crate::connection::{Client, ClientError}; + +/// A message pushed by the server to a subscribed connection. +#[derive(Debug, Clone)] +pub struct Message { + /// The channel the message was published to. + pub channel: Bytes, + /// The message payload. + pub data: Bytes, + /// Set only for pattern-matched messages (`PSUBSCRIBE`). Contains the + /// pattern that matched the channel. + pub pattern: Option, +} + +/// A connection locked into pub/sub mode. +/// +/// Obtained by calling [`Client::subscribe`] or [`Client::psubscribe`]. +/// The underlying transport is consumed — create a separate [`Client`] +/// for regular commands while this subscriber is active. +/// +/// # Example +/// +/// ```no_run +/// use ember_client::Client; +/// +/// #[tokio::main] +/// async fn main() -> Result<(), ember_client::ClientError> { +/// let mut publisher = Client::connect("127.0.0.1", 6379).await?; +/// let subscriber_conn = Client::connect("127.0.0.1", 6379).await?; +/// +/// let mut sub = subscriber_conn.subscribe(&["news"]).await?; +/// +/// publisher.publish("news", "breaking: hello world").await?; +/// +/// let msg = sub.recv().await?; +/// println!("got: {:?}", msg.data); +/// Ok(()) +/// } +/// ``` +pub struct Subscriber { + inner: Client, +} + +impl Subscriber { + pub(crate) fn new(inner: Client) -> Self { + Self { inner } + } + + /// Blocks until the next message arrives on any subscribed channel. + /// + /// Subscription confirmation frames (`subscribe`/`psubscribe`) are + /// skipped silently — only actual message frames are returned. + pub async fn recv(&mut self) -> Result { + loop { + let frame = self.inner.read_response().await?; + if let Some(msg) = try_parse_message(frame)? { + return Ok(msg); + } + // confirmation frame (subscribe/unsubscribe/psubscribe/punsubscribe) + // — loop back and wait for the next frame + } + } + + /// Subscribes to additional channels without leaving sub mode. + pub async fn subscribe(&mut self, channels: &[&str]) -> Result<(), ClientError> { + let mut parts = Vec::with_capacity(1 + channels.len()); + parts.push(Frame::Bulk(Bytes::from_static(b"SUBSCRIBE"))); + for ch in channels { + parts.push(Frame::Bulk(Bytes::copy_from_slice(ch.as_bytes()))); + } + self.inner.write_frame(Frame::Array(parts)).await?; + // drain the confirmation frames (one per channel) + for _ in 0..channels.len() { + self.inner.read_response().await?; + } + Ok(()) + } + + /// Unsubscribes from the given channels. + /// + /// When all subscriptions have been removed the inner [`Client`] is + /// returned so the connection can be reused for regular commands. + pub async fn unsubscribe(mut self, channels: &[&str]) -> Result { + let mut parts = Vec::with_capacity(1 + channels.len()); + parts.push(Frame::Bulk(Bytes::from_static(b"UNSUBSCRIBE"))); + for ch in channels { + parts.push(Frame::Bulk(Bytes::copy_from_slice(ch.as_bytes()))); + } + self.inner.write_frame(Frame::Array(parts)).await?; + // drain the unsubscribe confirmation frames + for _ in 0..channels.len() { + self.inner.read_response().await?; + } + Ok(self.inner) + } +} + +/// Tries to parse a push frame into a [`Message`]. +/// +/// Returns `Ok(Some(_))` for `message` and `pmessage` frames. +/// Returns `Ok(None)` for subscription management frames. +/// Returns `Err` for unexpected or malformed frames. +fn try_parse_message(frame: Frame) -> Result, ClientError> { + let elems = match frame { + Frame::Array(e) => e, + Frame::Error(e) => return Err(ClientError::Server(e)), + other => { + return Err(ClientError::Protocol(format!( + "expected array push frame, got {other:?}" + ))) + } + }; + + if elems.len() < 3 { + return Err(ClientError::Protocol(format!( + "push frame too short: {} elements", + elems.len() + ))); + } + + let kind = match &elems[0] { + Frame::Bulk(b) => b.clone(), + Frame::Simple(s) => Bytes::copy_from_slice(s.as_bytes()), + other => { + return Err(ClientError::Protocol(format!( + "expected bulk/simple frame kind, got {other:?}" + ))) + } + }; + + match kind.as_ref() { + b"message" => { + if elems.len() < 3 { + return Err(ClientError::Protocol( + "message frame has fewer than 3 elements".into(), + )); + } + let channel = bulk_bytes(elems[1].clone())?; + let data = bulk_bytes(elems[2].clone())?; + Ok(Some(Message { + channel, + data, + pattern: None, + })) + } + b"pmessage" => { + if elems.len() < 4 { + return Err(ClientError::Protocol( + "pmessage frame has fewer than 4 elements".into(), + )); + } + let pattern = bulk_bytes(elems[1].clone())?; + let channel = bulk_bytes(elems[2].clone())?; + let data = bulk_bytes(elems[3].clone())?; + Ok(Some(Message { + channel, + data, + pattern: Some(pattern), + })) + } + // subscribe / unsubscribe / psubscribe / punsubscribe confirmations + _ => Ok(None), + } +} + +fn bulk_bytes(frame: Frame) -> Result { + match frame { + Frame::Bulk(b) => Ok(b), + other => Err(ClientError::Protocol(format!( + "expected bulk in push frame, got {other:?}" + ))), + } +} diff --git a/crates/ember-client/src/vector.rs b/crates/ember-client/src/vector.rs new file mode 100644 index 00000000..2b4a3a26 --- /dev/null +++ b/crates/ember-client/src/vector.rs @@ -0,0 +1,338 @@ +//! Vector set commands (`VADD`, `VSIM`, etc.). +//! +//! Enabled by the `vector` cargo feature. Requires the server to be built +//! with the same feature enabled. + +use bytes::Bytes; +use ember_protocol::types::Frame; + +use crate::connection::{Client, ClientError}; + +// --- public types --- + +/// One result from a [`Client::vsim`] nearest-neighbour search. +#[derive(Debug, Clone)] +pub struct SimResult { + /// The element name. + pub element: Bytes, + /// Distance from the query vector. Lower is closer (metric-dependent). + pub distance: f32, +} + +// --- decoders --- + +fn vsim_results(frame: Frame, with_scores: bool) -> Result, ClientError> { + let elems = match frame { + Frame::Array(e) => e, + Frame::Null => return Ok(Vec::new()), + Frame::Error(e) => return Err(ClientError::Server(e)), + other => { + return Err(ClientError::Protocol(format!( + "expected array for VSIM, got {other:?}" + ))) + } + }; + + if !with_scores { + return elems + .into_iter() + .map(|e| match e { + Frame::Bulk(b) => Ok(SimResult { + element: b, + distance: 0.0, + }), + other => Err(ClientError::Protocol(format!( + "expected bulk element in VSIM, got {other:?}" + ))), + }) + .collect(); + } + + if elems.len() % 2 != 0 { + return Err(ClientError::Protocol(format!( + "VSIM WITHSCORES array has odd length ({})", + elems.len() + ))); + } + + let mut result = Vec::with_capacity(elems.len() / 2); + let mut iter = elems.into_iter(); + while let (Some(elem_frame), Some(dist_frame)) = (iter.next(), iter.next()) { + let element = match elem_frame { + Frame::Bulk(b) => b, + other => { + return Err(ClientError::Protocol(format!( + "expected bulk element in VSIM WITHSCORES, got {other:?}" + ))) + } + }; + let distance = match dist_frame { + Frame::Bulk(b) => { + let s = std::str::from_utf8(&b).map_err(|_| { + ClientError::Protocol("VSIM distance is not valid UTF-8".into()) + })?; + s.parse::().map_err(|_| { + ClientError::Protocol(format!("VSIM distance is not a valid float: {s:?}")) + })? + } + other => { + return Err(ClientError::Protocol(format!( + "expected bulk distance in VSIM WITHSCORES, got {other:?}" + ))) + } + }; + result.push(SimResult { element, distance }); + } + Ok(result) +} + +fn vget_vector(frame: Frame) -> Result>, ClientError> { + match frame { + Frame::Array(elems) => { + let floats = elems + .into_iter() + .map(|e| match e { + Frame::Bulk(b) => { + let s = std::str::from_utf8(&b).map_err(|_| { + ClientError::Protocol("vector component is not valid UTF-8".into()) + })?; + s.parse::().map_err(|_| { + ClientError::Protocol(format!( + "vector component is not a valid float: {s:?}" + )) + }) + } + other => Err(ClientError::Protocol(format!( + "expected bulk float in VGET response, got {other:?}" + ))), + }) + .collect::, _>>()?; + Ok(Some(floats)) + } + Frame::Null => Ok(None), + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected array or null for VGET, got {other:?}" + ))), + } +} + +fn vinfo_pairs(frame: Frame) -> Result, ClientError> { + match frame { + Frame::Array(elems) => { + if elems.len() % 2 != 0 { + return Err(ClientError::Protocol(format!( + "VINFO array has odd length ({})", + elems.len() + ))); + } + let mut result = Vec::with_capacity(elems.len() / 2); + let mut iter = elems.into_iter(); + while let (Some(k), Some(v)) = (iter.next(), iter.next()) { + let key = match k { + Frame::Bulk(b) => b, + Frame::Simple(s) => Bytes::copy_from_slice(s.as_bytes()), + other => { + return Err(ClientError::Protocol(format!( + "expected bulk key in VINFO, got {other:?}" + ))) + } + }; + let val = match v { + Frame::Bulk(b) => b, + Frame::Integer(n) => Bytes::copy_from_slice(n.to_string().as_bytes()), + other => { + return Err(ClientError::Protocol(format!( + "expected bulk/integer value in VINFO, got {other:?}" + ))) + } + }; + result.push((key, val)); + } + Ok(result) + } + Frame::Null => Ok(Vec::new()), + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected array for VINFO, got {other:?}" + ))), + } +} + +// --- Client methods --- + +impl Client { + /// Adds `element` with `vector` to the vector set at `key`. Returns `true` + /// if the element was newly inserted, `false` if it already existed and + /// was updated. + /// + /// Requires the server `vector` feature. + pub async fn vadd( + &mut self, + key: &str, + element: &str, + vector: &[f32], + ) -> Result { + let mut parts = Vec::with_capacity(3 + vector.len()); + parts.push(Frame::Bulk(Bytes::from_static(b"VADD"))); + parts.push(Frame::Bulk(Bytes::copy_from_slice(key.as_bytes()))); + parts.push(Frame::Bulk(Bytes::copy_from_slice(element.as_bytes()))); + for &f in vector { + parts.push(Frame::Bulk(Bytes::copy_from_slice( + f.to_string().as_bytes(), + ))); + } + let frame = self.send_frame(Frame::Array(parts)).await?; + match frame { + Frame::Integer(1) => Ok(true), + Frame::Integer(0) => Ok(false), + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected 0 or 1 for VADD, got {other:?}" + ))), + } + } + + /// Adds multiple elements in a single round-trip. Returns the number of + /// newly inserted vectors (updates do not count). + /// + /// The server requires all vectors in a batch to have the same + /// dimensionality as the vector set (or creates it from the first batch). + pub async fn vadd_batch( + &mut self, + key: &str, + dim: usize, + entries: &[(&str, &[f32])], + ) -> Result { + let mut parts = Vec::with_capacity(5 + entries.len() * (1 + dim)); + parts.push(Frame::Bulk(Bytes::from_static(b"VADD_BATCH"))); + parts.push(Frame::Bulk(Bytes::copy_from_slice(key.as_bytes()))); + parts.push(Frame::Bulk(Bytes::from_static(b"DIM"))); + parts.push(Frame::Bulk(Bytes::copy_from_slice( + dim.to_string().as_bytes(), + ))); + for (element, vector) in entries { + parts.push(Frame::Bulk(Bytes::copy_from_slice(element.as_bytes()))); + for &f in *vector { + parts.push(Frame::Bulk(Bytes::copy_from_slice( + f.to_string().as_bytes(), + ))); + } + } + let frame = self.send_frame(Frame::Array(parts)).await?; + match frame { + Frame::Integer(n) => Ok(n), + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected integer for VADD_BATCH, got {other:?}" + ))), + } + } + + /// Searches the vector set at `key` for the `count` nearest neighbours of + /// `query`. Returns results sorted by distance (closest first). + pub async fn vsim( + &mut self, + key: &str, + query: &[f32], + count: u32, + ) -> Result, ClientError> { + let mut parts = Vec::with_capacity(4 + query.len()); + parts.push(Frame::Bulk(Bytes::from_static(b"VSIM"))); + parts.push(Frame::Bulk(Bytes::copy_from_slice(key.as_bytes()))); + for &f in query { + parts.push(Frame::Bulk(Bytes::copy_from_slice( + f.to_string().as_bytes(), + ))); + } + parts.push(Frame::Bulk(Bytes::from_static(b"COUNT"))); + parts.push(Frame::Bulk(Bytes::copy_from_slice( + count.to_string().as_bytes(), + ))); + parts.push(Frame::Bulk(Bytes::from_static(b"WITHSCORES"))); + let frame = self.send_frame(Frame::Array(parts)).await?; + vsim_results(frame, true) + } + + /// Removes `element` from the vector set at `key`. Returns `true` if the + /// element was removed. + pub async fn vrem(&mut self, key: &str, element: &str) -> Result { + let frame = self + .send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"VREM")), + Frame::Bulk(Bytes::copy_from_slice(key.as_bytes())), + Frame::Bulk(Bytes::copy_from_slice(element.as_bytes())), + ])) + .await?; + match frame { + Frame::Integer(1) => Ok(true), + Frame::Integer(0) => Ok(false), + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected 0 or 1 for VREM, got {other:?}" + ))), + } + } + + /// Returns the raw vector stored for `element` in the vector set at `key`, + /// or `None` if not found. + pub async fn vget( + &mut self, + key: &str, + element: &str, + ) -> Result>, ClientError> { + let frame = self + .send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"VGET")), + Frame::Bulk(Bytes::copy_from_slice(key.as_bytes())), + Frame::Bulk(Bytes::copy_from_slice(element.as_bytes())), + ])) + .await?; + vget_vector(frame) + } + + /// Returns the number of vectors in the vector set at `key`. + pub async fn vcard(&mut self, key: &str) -> Result { + let frame = self + .send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"VCARD")), + Frame::Bulk(Bytes::copy_from_slice(key.as_bytes())), + ])) + .await?; + match frame { + Frame::Integer(n) => Ok(n), + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected integer for VCARD, got {other:?}" + ))), + } + } + + /// Returns the dimensionality of vectors in the set at `key`. + pub async fn vdim(&mut self, key: &str) -> Result { + let frame = self + .send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"VDIM")), + Frame::Bulk(Bytes::copy_from_slice(key.as_bytes())), + ])) + .await?; + match frame { + Frame::Integer(n) => Ok(n), + Frame::Error(e) => Err(ClientError::Server(e)), + other => Err(ClientError::Protocol(format!( + "expected integer for VDIM, got {other:?}" + ))), + } + } + + /// Returns metadata about the vector set at `key` as key-value pairs. + pub async fn vinfo(&mut self, key: &str) -> Result, ClientError> { + let frame = self + .send_frame(Frame::Array(vec![ + Frame::Bulk(Bytes::from_static(b"VINFO")), + Frame::Bulk(Bytes::copy_from_slice(key.as_bytes())), + ])) + .await?; + vinfo_pairs(frame) + } +} diff --git a/tests/integration/src/client_typed_api.rs b/tests/integration/src/client_typed_api.rs index 170ac479..69278afb 100644 --- a/tests/integration/src/client_typed_api.rs +++ b/tests/integration/src/client_typed_api.rs @@ -226,6 +226,274 @@ async fn pipeline_mixed_commands() { assert_eq!(frames.len(), 3); } +// --- strlen --- + +#[tokio::test] +async fn strlen_existing_and_missing() { + let (_server, mut client) = connect().await; + client.set("slen", "hello").await.unwrap(); + assert_eq!(client.strlen("slen").await.unwrap(), 5); + assert_eq!(client.strlen("nosuchkey").await.unwrap(), 0); +} + +// --- incr_by_float --- + +#[tokio::test] +async fn incr_by_float_positive_and_negative() { + let (_server, mut client) = connect().await; + client.set("flt", "10.5").await.unwrap(); + let v = client.incr_by_float("flt", 1.5).await.unwrap(); + assert!((v - 12.0).abs() < 1e-9); + let v2 = client.incr_by_float("flt", -2.0).await.unwrap(); + assert!((v2 - 10.0).abs() < 1e-9); +} + +// --- pexpire --- + +#[tokio::test] +async fn pexpire_sets_ttl_in_millis() { + let (_server, mut client) = connect().await; + client.set("px_key", "v").await.unwrap(); + let set = client.pexpire("px_key", 60_000).await.unwrap(); + assert!(set); + let pttl = client.pttl("px_key").await.unwrap(); + assert!(pttl > 0 && pttl <= 60_000); +} + +// --- hmget --- + +#[tokio::test] +async fn hmget_subset_and_missing() { + let (_server, mut client) = connect().await; + client + .hset("hm", &[("a", "1"), ("b", "2"), ("c", "3")]) + .await + .unwrap(); + let vals = client.hmget("hm", &["a", "c", "missing"]).await.unwrap(); + assert_eq!(vals.len(), 3); + assert_eq!(vals[0].as_deref(), Some(b"1" as &[u8])); + assert_eq!(vals[1].as_deref(), Some(b"3" as &[u8])); + assert!(vals[2].is_none()); +} + +// --- key_type --- + +#[tokio::test] +async fn key_type_returns_correct_types() { + let (_server, mut client) = connect().await; + client.set("str_key", "v").await.unwrap(); + assert_eq!(client.key_type("str_key").await.unwrap(), "string"); + + client.rpush("lst_key", &["item"]).await.unwrap(); + assert_eq!(client.key_type("lst_key").await.unwrap(), "list"); + + assert_eq!(client.key_type("no_such_key").await.unwrap(), "none"); +} + +// --- keys --- + +#[tokio::test] +async fn keys_glob_returns_matching_keys() { + let (_server, mut client) = connect().await; + client.set("pfx:a", "1").await.unwrap(); + client.set("pfx:b", "2").await.unwrap(); + client.set("other", "3").await.unwrap(); + let mut matched = client.keys("pfx:*").await.unwrap(); + matched.sort(); + assert_eq!(matched.len(), 2); + assert_eq!(matched[0].as_ref(), b"pfx:a"); + assert_eq!(matched[1].as_ref(), b"pfx:b"); +} + +// --- rename --- + +#[tokio::test] +async fn rename_then_old_key_gone() { + let (_server, mut client) = connect().await; + client.set("old_name", "value").await.unwrap(); + client.rename("old_name", "new_name").await.unwrap(); + assert!(client.get("old_name").await.unwrap().is_none()); + assert_eq!( + client.get("new_name").await.unwrap().as_deref(), + Some(b"value" as &[u8]) + ); +} + +// --- scan --- + +#[tokio::test] +async fn scan_full_iteration_collects_all_keys() { + let (_server, mut client) = connect().await; + let expected_keys = ["scan_a", "scan_b", "scan_c"]; + for k in &expected_keys { + client.set(k, "v").await.unwrap(); + } + + let mut all_keys = Vec::new(); + let mut cursor = 0u64; + loop { + let page = client.scan(cursor, None, Some("scan_*")).await.unwrap(); + all_keys.extend(page.keys); + cursor = page.cursor; + if cursor == 0 { + break; + } + } + + all_keys.sort(); + assert_eq!(all_keys.len(), 3); + assert_eq!(all_keys[0].as_ref(), b"scan_a"); + assert_eq!(all_keys[2].as_ref(), b"scan_c"); +} + +// --- echo --- + +#[tokio::test] +async fn echo_returns_same_bytes() { + let (_server, mut client) = connect().await; + let reply = client.echo("hello world").await.unwrap(); + assert_eq!(reply.as_ref(), b"hello world"); +} + +// --- unlink --- + +#[tokio::test] +async fn unlink_returns_count_like_del() { + let (_server, mut client) = connect().await; + client.set("ul1", "v").await.unwrap(); + client.set("ul2", "v").await.unwrap(); + let removed = client.unlink(&["ul1", "ul2", "ul_missing"]).await.unwrap(); + assert_eq!(removed, 2); + assert!(client.get("ul1").await.unwrap().is_none()); +} + +// --- info --- + +#[tokio::test] +async fn info_returns_non_empty_string() { + let (_server, mut client) = connect().await; + let output = client.info(None).await.unwrap(); + assert!(!output.is_empty()); +} + +// --- bgsave --- + +#[tokio::test] +async fn bgsave_returns_status_string() { + let (_server, mut client) = connect().await; + let status = client.bgsave().await.unwrap(); + assert!(!status.is_empty()); +} + +// --- slowlog --- + +#[tokio::test] +async fn slowlog_len_is_non_negative() { + let (_server, mut client) = connect().await; + let len = client.slowlog_len().await.unwrap(); + assert!(len >= 0); +} + +#[tokio::test] +async fn slowlog_reset_clears_log() { + let (_server, mut client) = connect().await; + client.slowlog_reset().await.unwrap(); + assert_eq!(client.slowlog_len().await.unwrap(), 0); +} + +// --- pub/sub (request-response) --- + +#[tokio::test] +async fn publish_to_channel_with_no_subscribers_returns_zero() { + let (_server, mut client) = connect().await; + let count = client.publish("unsubscribed_channel", "msg").await.unwrap(); + assert_eq!(count, 0); +} + +#[tokio::test] +async fn pubsub_channels_with_pattern_and_none() { + let (_server, mut client) = connect().await; + // no subscribers active — both should return empty + let with_pat = client.pubsub_channels(Some("*")).await.unwrap(); + let all = client.pubsub_channels(None).await.unwrap(); + assert!(with_pat.is_empty()); + assert!(all.is_empty()); +} + +#[tokio::test] +async fn pubsub_numsub_empty_channels() { + let (_server, mut client) = connect().await; + let pairs = client.pubsub_numsub(&["no_one_here"]).await.unwrap(); + assert_eq!(pairs.len(), 1); + assert_eq!(pairs[0].0.as_ref(), b"no_one_here"); + assert_eq!(pairs[0].1, 0); +} + +#[tokio::test] +async fn pubsub_numpat_is_zero_before_subscriptions() { + let (_server, mut client) = connect().await; + let n = client.pubsub_numpat().await.unwrap(); + assert_eq!(n, 0); +} + +// --- subscriber --- + +#[tokio::test] +async fn subscriber_recv_message() { + let server = crate::helpers::TestServer::start(); + let sub_conn = Client::connect("127.0.0.1", server.port) + .await + .expect("connect failed"); + let mut publisher = Client::connect("127.0.0.1", server.port) + .await + .expect("connect failed"); + + let mut sub = sub_conn.subscribe(&["typed_events"]).await.unwrap(); + publisher.publish("typed_events", "hello").await.unwrap(); + + let msg = sub.recv().await.unwrap(); + assert_eq!(msg.channel.as_ref(), b"typed_events"); + assert_eq!(msg.data.as_ref(), b"hello"); + assert!(msg.pattern.is_none()); +} + +#[tokio::test] +async fn subscriber_multiple_channels_correct_message() { + let server = crate::helpers::TestServer::start(); + let sub_conn = Client::connect("127.0.0.1", server.port) + .await + .expect("connect failed"); + let mut publisher = Client::connect("127.0.0.1", server.port) + .await + .expect("connect failed"); + + let mut sub = sub_conn.subscribe(&["ch_a", "ch_b"]).await.unwrap(); + publisher.publish("ch_b", "payload").await.unwrap(); + + let msg = sub.recv().await.unwrap(); + assert_eq!(msg.channel.as_ref(), b"ch_b"); + assert_eq!(msg.data.as_ref(), b"payload"); +} + +#[tokio::test] +async fn psubscriber_pattern_match() { + let server = crate::helpers::TestServer::start(); + let sub_conn = Client::connect("127.0.0.1", server.port) + .await + .expect("connect failed"); + let mut publisher = Client::connect("127.0.0.1", server.port) + .await + .expect("connect failed"); + + let mut sub = sub_conn.psubscribe(&["typed:*"]).await.unwrap(); + publisher.publish("typed:update", "data").await.unwrap(); + + let msg = sub.recv().await.unwrap(); + assert_eq!(msg.channel.as_ref(), b"typed:update"); + assert_eq!(msg.data.as_ref(), b"data"); + assert_eq!(msg.pattern.as_deref(), Some(b"typed:*" as &[u8])); +} + // --- error surfacing --- #[tokio::test]