diff --git a/src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java b/src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java index c822fcea47..19fd29963d 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java @@ -219,6 +219,8 @@ public Map getLastEntry() { * {@literal Redis Stream}. * * @author Christoph Strobl + * @author Mark Paluch + * @author Krzysztof Kocel */ public static class XInfoGroups implements Streamable { @@ -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 { diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index b6b9cd6ed4..36a16d8f90 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -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()); } @@ -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"); } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsIntegrationTests.java index 9fb4d8401b..a98a3b6350 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsIntegrationTests.java @@ -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(); } @@ -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(); }