Add lag and entries read to StreamInfo

Author: kkocel

src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java

@@ -361,6 +361,26 @@ 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
+		 */
+		public Long entriesRead() {
+			return getRequired("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("entries-read", Long.class);
+		}
 	}
 
 	public static class XInfoConsumers implements Streamable<XInfoConsumer> {

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()).isZero();
+		assertThat(info.get(0).entriesRead()).isZero();
 		assertThat(info.get(0).lastDeliveredId()).isEqualTo("0-0");
 	}
 

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()).isZero();
+					assertThat(info.lag()).isZero();
 					assertThat(info.lastDeliveredId()).isEqualTo("0-0");
 				}).verifyComplete();
 	}