Skip to content
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.

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

Add lag and entries-read to StreamInfo (XINFO GROUPS) #2949

Open
wants to merge 8 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,8 @@ public Map<Object, Object> getLastEntry() {
* {@literal Redis Stream}.
*
* @author Christoph Strobl
* @author Mark Paluch
* @author Krzysztof Kocel
*/
public static class XInfoGroups implements Streamable<XInfoGroup> {

Expand Down Expand Up @@ -361,6 +363,27 @@ public Long pendingCount() {
public String lastDeliveredId() {
return getRequired("last-delivered-id", String.class);
}

/**
* The logical "read counter" of the last entry delivered to the group's consumers. Corresponds to {@literal entries-read}.
*
* @return
*/
@Nullable
public Long entriesRead() {
return get("entries-read", Long.class);
}

/**
* The number of entries in the stream that are still waiting to be delivered to the group's consumers,
* or a NULL when that number can't be determined. Corresponds to {@literal lag}.
*
* @return
*/
@Nullable
public Long lag() {
return get("lag", Long.class);
}
}

public static class XInfoConsumers implements Streamable<XInfoConsumer> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4017,6 +4017,8 @@ public void xinfoGroups() {
assertThat(info.get(0).groupName()).isEqualTo("my-group");
assertThat(info.get(0).consumerCount()).isEqualTo(1L);
assertThat(info.get(0).pendingCount()).isEqualTo(2L);
assertThat(info.get(0).lag()).isEqualTo(0L);
assertThat(info.get(0).entriesRead()).isEqualTo(2L);
assertThat(info.get(0).lastDeliveredId()).isEqualTo(lastRecord.getValue());
}

Expand Down Expand Up @@ -4055,6 +4057,8 @@ public void xinfoGroupsNoConsumer() {
assertThat(info.get(0).groupName()).isEqualTo("my-group");
assertThat(info.get(0).consumerCount()).isZero();
assertThat(info.get(0).pendingCount()).isZero();
assertThat(info.get(0).lag()).isEqualTo(2);
assertThat(info.get(0).entriesRead()).isNull();
assertThat(info.get(0).lastDeliveredId()).isEqualTo("0-0");
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -429,6 +429,8 @@ void xinfoGroups() {
assertThat(info.groupName()).isEqualTo("my-group");
assertThat(info.consumerCount()).isEqualTo(1L);
assertThat(info.pendingCount()).isEqualTo(2L);
assertThat(info.lag()).isZero();
assertThat(info.entriesRead()).isEqualTo(2L);
assertThat(info.lastDeliveredId()).isEqualTo(lastRecord);
}).verifyComplete();
}
Expand All @@ -455,6 +457,8 @@ void xinfoGroupsNoConsumer() {
assertThat(info.groupName()).isEqualTo("my-group");
assertThat(info.consumerCount()).isZero();
assertThat(info.pendingCount()).isZero();
assertThat(info.entriesRead()).isNull();
assertThat(info.lag()).isEqualTo(2);
assertThat(info.lastDeliveredId()).isEqualTo("0-0");
}).verifyComplete();
}
Expand Down