Skip to content

(feat) Streams: "_map" methods to return an Option for empty/null xread response #364

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
danwritecode wants to merge 4 commits into
base: main
Choose a base branch
from
Open

(feat) Streams: "_map" methods to return an Option for empty/null xread response #364

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
483db89
updated "_map" functions to return an Option in event of NULL value
danwritecode Oct 22, 2025
ac26182
removed commented code
danwritecode Oct 22, 2025
2d82b98
updated to check for an empty array in addition to null Value
danwritecode Oct 22, 2025
7eb9f9a
updated tests and docs
danwritecode Oct 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion examples/streams.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,11 @@ async fn main() {
let mut count = 0;
loop {
// call XREAD for new records in a loop, blocking up to 10 sec each time
let entry: XReadResponse<Str, Str, Str, Str> = client.xread_map(Some(1), Some(10_000), "foo", "$").await?;
// Note: _map return an Option in the result of an empty stream response.
let entry: XReadResponse<Str, Str, Str, Str> = client.xread_map(Some(1), Some(10_000), "foo", "$")
.await?
.expect("Expected Some found None.");

count += 1;

for (key, records) in entry.into_iter() {
Expand Down
39 changes: 31 additions & 8 deletions src/commands/interfaces/streams.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -370,7 +370,7 @@ pub trait StreamsInterface: ClientLike + Sized {
block: Option<u64>,
keys: K,
ids: I,
) -> impl Future<Output = FredResult<XReadResponse<Rk1, Rk2, Rk3, Rv>>> + Send
) -> impl Future<Output = FredResult<Option<XReadResponse<Rk1, Rk2, Rk3, Rv>>>> + Send
where
Rk1: FromKey + Hash + Eq,
Rk2: FromValue,
Expand All @@ -381,9 +381,20 @@ pub trait StreamsInterface: ClientLike + Sized {
{
async move {
into!(keys, ids);
commands::streams::xread(self, count, block, keys, ids)
.await?
.into_xread_response()
let value = commands::streams::xread(self, count, block, keys, ids)
.await?;

if let Value::Array(values) = &value {
if values.len() == 0 {
return Ok(None);
}
}

if let Value::Null = value {
return Ok(None);
}

Ok(Some(value.into_xread_response()?))
}
}

Expand Down Expand Up @@ -540,7 +551,7 @@ pub trait StreamsInterface: ClientLike + Sized {
noack: bool,
keys: K,
ids: I,
) -> impl Future<Output = FredResult<XReadResponse<Rk1, Rk2, Rk3, Rv>>> + Send
) -> impl Future<Output = FredResult<Option<XReadResponse<Rk1, Rk2, Rk3, Rv>>>> + Send
where
Rk1: FromKey + Hash + Eq,
Rk2: FromValue,
Expand All @@ -553,9 +564,21 @@ pub trait StreamsInterface: ClientLike + Sized {
{
async move {
into!(group, consumer, keys, ids);
commands::streams::xreadgroup(self, group, consumer, count, block, noack, keys, ids)
.await?
.into_xread_response()
let value = commands::streams::xreadgroup(self, group, consumer, count, block, noack, keys, ids)
.await?;

if let Value::Array(values) = &value {
if values.len() == 0 {
return Ok(None);
}
}

if let Value::Null = value {
return Ok(None);
}

Ok(Some(value.into_xread_response()?))

}
}

Expand Down
37 changes: 27 additions & 10 deletions tests/integration/streams/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,7 +266,9 @@ pub async fn should_xread_map_one_key(client: Client, _: Config) -> Result<(), E
create_fake_group_and_stream(&client, "foo{1}").await?;
let _ = add_stream_entries(&client, "foo{1}", 3).await?;

let result: XReadResponse<String, String, String, usize> = client.xread_map(None, None, "foo{1}", "0").await?;
let result: XReadResponse<String, String, String, usize> = client.xread_map(None, None, "foo{1}", "0")
.await?
.expect("Expected Some found None.");

for (idx, (_, record)) in result.get("foo{1}").unwrap().iter().enumerate() {
let count = record.get("count").expect("Failed to read count");
Expand Down Expand Up @@ -415,7 +417,8 @@ pub async fn should_xreadgroup_one_stream(client: Client, _: Config) -> Result<(

let result: XReadResponse<String, String, String, usize> = client
.xreadgroup_map("group1", "consumer1", None, None, false, "foo{1}", ">")
.await?;
.await?
.expect("Expected Some found None.");

assert_eq!(result.len(), 1);
for (idx, (_, record)) in result.get("foo{1}").unwrap().iter().enumerate() {
Expand Down Expand Up @@ -444,7 +447,8 @@ pub async fn should_xreadgroup_multiple_stream(client: Client, _: Config) -> Res
vec!["foo{1}", "bar{1}"],
vec![">", ">"],
)
.await?;
.await?
.expect("Expected Some found None.");

assert_eq!(result.len(), 2);
for (idx, (_, record)) in result.get("foo{1}").unwrap().iter().enumerate() {
Expand Down Expand Up @@ -475,7 +479,8 @@ pub async fn should_xreadgroup_block(client: Client, _: Config) -> Result<(), Er

let mut result: XReadResponse<String, String, String, usize> = client
.xreadgroup_map("group1", "consumer1", None, Some(10_000), false, "foo{1}", ">")
.await?;
.await?
.expect("Expected Some found None.");

assert_eq!(result.len(), 1);
let records = result.remove("foo{1}").unwrap();
Expand All @@ -493,7 +498,9 @@ pub async fn should_xack_one_id(client: Client, _: Config) -> Result<(), Error>

let result: XReadResponse<String, String, String, usize> = client
.xreadgroup_map("group1", "consumer1", None, None, false, "foo{1}", ">")
.await?;
.await?
.expect("Expected Some found None.");

assert_eq!(result.len(), 1);
let records = result.get("foo{1}").unwrap();
let id = records[0].0.clone();
Expand All @@ -510,7 +517,9 @@ pub async fn should_xack_multiple_ids(client: Client, _: Config) -> Result<(), E

let result: XReadResponse<String, String, String, usize> = client
.xreadgroup_map("group1", "consumer1", None, None, false, "foo{1}", ">")
.await?;
.await?
.expect("Expected Some found None.");

assert_eq!(result.len(), 1);
let records = result.get("foo{1}").unwrap();
let ids: Vec<String> = records.iter().map(|(id, _)| id.clone()).collect();
Expand All @@ -528,7 +537,9 @@ pub async fn should_xclaim_one_id(client: Client, _: Config) -> Result<(), Error

let mut result: XReadResponse<String, String, String, usize> = client
.xreadgroup_map("group1", "consumer1", Some(1), None, false, "foo{1}", ">")
.await?;
.await?
.expect("Expected Some found None.");

assert_eq!(result.len(), 1);
assert_eq!(result.get("foo{1}").unwrap().len(), 1);
let first_read_id = result.get_mut("foo{1}").unwrap().pop().unwrap().0;
Expand Down Expand Up @@ -574,7 +585,9 @@ pub async fn should_xclaim_multiple_ids(client: Client, _: Config) -> Result<(),

let mut result: XReadResponse<String, String, String, usize> = client
.xreadgroup_map("group1", "consumer1", Some(2), None, false, "foo{1}", ">")
.await?;
.await?
.expect("Expected Some found None.");

assert_eq!(result.len(), 1);
assert_eq!(result.get("foo{1}").unwrap().len(), 2);
let second_read_id = result.get_mut("foo{1}").unwrap().pop().unwrap().0;
Expand Down Expand Up @@ -626,7 +639,9 @@ pub async fn should_xclaim_with_justid(client: Client, _: Config) -> Result<(),

let mut result: XReadResponse<String, String, String, usize> = client
.xreadgroup_map("group1", "consumer1", Some(2), None, false, "foo{1}", ">")
.await?;
.await?
.expect("Expected Some found None.");

assert_eq!(result.len(), 1);
assert_eq!(result.get("foo{1}").unwrap().len(), 2);
let second_read_id = result.get_mut("foo{1}").unwrap().pop().unwrap().0;
Expand Down Expand Up @@ -671,7 +686,9 @@ pub async fn should_xautoclaim_default(client: Client, _: Config) -> Result<(),

let mut result: XReadResponse<String, String, String, usize> = client
.xreadgroup_map("group1", "consumer1", Some(2), None, false, "foo{1}", ">")
.await?;
.await?
.expect("Expected Some found None.");

assert_eq!(result.len(), 1);
assert_eq!(result.get("foo{1}").unwrap().len(), 2);
let second_read_id = result.get_mut("foo{1}").unwrap().pop().unwrap().0;
Expand Down