DATAREDIS-1119 - Support XINFO via ReactiveStreamCommands.

Original pull request: #519.

src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java

@@ -41,6 +41,9 @@ import org.springframework.data.redis.connection.stream.PendingMessages;
 import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
 import org.springframework.data.redis.connection.stream.ReadOffset;
 import org.springframework.data.redis.connection.stream.RecordId;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumer;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroup;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream;
 import org.springframework.data.redis.connection.stream.StreamOffset;
 import org.springframework.data.redis.connection.stream.StreamReadOptions;
 import org.springframework.data.redis.connection.stream.StreamRecords;
@@ -1030,6 +1033,101 @@ public interface ReactiveStreamCommands {
 	 */
 	Flux<CommandResponse<ReadCommand, Flux<ByteBufferRecord>>> read(Publisher<ReadCommand> commands);
 
+	/**
+	 * @author Christoph Strobl
+	 * @since 2.3
+	 */
+	class XInfoCommand extends KeyCommand {
+
+		private final @Nullable String groupName;
+
+		public XInfoCommand(@Nullable ByteBuffer key, @Nullable String groupName) {
+
+			super(key);
+			this.groupName = groupName;
+		}
+
+		public static XInfoCommand xInfo() {
+			return new XInfoCommand(null, null);
+		}
+
+		public XInfoCommand of(ByteBuffer key) {
+			return new XInfoCommand(key, groupName);
+		}
+
+		public XInfoCommand consumersIn(String groupName) {
+			return new XInfoCommand(getKey(), groupName);
+		}
+
+		public String getGroupName() {
+			return groupName;
+		}
+	}
+
+	/**
+	 * Obtain general information about the stream stored at the specified {@literal key}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @return a {@link Mono} emitting {@link XInfoStream} when ready.
+	 * @since 2.3
+	 */
+	default Mono<XInfoStream> xInfo(ByteBuffer key) {
+		return xInfo(Mono.just(XInfoCommand.xInfo().of(key))).next().map(CommandResponse::getOutput);
+	}
+
+	/**
+	 * Obtain general information about the stream stored at the specified {@literal key}.
+	 *
+	 * @param commands must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.3
+	 */
+	Flux<CommandResponse<XInfoCommand, XInfoStream>> xInfo(Publisher<XInfoCommand> commands);
+
+	/**
+	 * Obtain general information about the stream stored at the specified {@literal key}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @return a {@link Flux} emitting consumer group info one by one.
+	 * @since 2.3
+	 */
+	default Flux<XInfoGroup> xInfoGroups(ByteBuffer key) {
+		return xInfoGroups(Mono.just(XInfoCommand.xInfo().of(key))).next().flatMapMany(CommandResponse::getOutput);
+	}
+
+	/**
+	 * Obtain general information about the stream stored at the specified {@literal key}.
+	 *
+	 * @param commands must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.3
+	 */
+	Flux<CommandResponse<XInfoCommand, Flux<XInfoGroup>>> xInfoGroups(Publisher<XInfoCommand> commands);
+
+	/**
+	 * Obtain information about every consumer in a specific {@literal consumer group} for the stream stored at the
+	 * specified {@literal key}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param groupName name of the {@literal consumer group}.
+	 * @return a {@link Flux} emitting consumer info one by one.
+	 * @since 2.3
+	 */
+	default Flux<XInfoConsumer> xInfoConsumers(ByteBuffer key, String groupName) {
+		return xInfoConsumers(Mono.just(XInfoCommand.xInfo().of(key).consumersIn(groupName))).next()
+				.flatMapMany(CommandResponse::getOutput);
+	}
+
+	/**
+	 * Obtain information about every consumer in a specific {@literal consumer group} for the stream stored at the
+	 * specified {@literal key}.
+	 *
+	 * @param commands must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.3
+	 */
+	Flux<CommandResponse<XInfoCommand, Flux<XInfoConsumer>>> xInfoConsumers(Publisher<XInfoCommand> commands);
+
 	class GroupCommand extends KeyCommand {
 
 		private final GroupCommandAction action;

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java

@@ -20,6 +20,7 @@ import io.lettuce.core.XClaimArgs;
 import io.lettuce.core.XReadArgs;
 import io.lettuce.core.XReadArgs.StreamOffset;
 import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumer;
 import reactor.core.publisher.Flux;
 
 import java.nio.ByteBuffer;
@@ -39,6 +40,8 @@ import org.springframework.data.redis.connection.stream.Consumer;
 import org.springframework.data.redis.connection.stream.PendingMessages;
 import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
 import org.springframework.data.redis.connection.stream.RecordId;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroup;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream;
 import org.springframework.data.redis.connection.stream.StreamReadOptions;
 import org.springframework.data.redis.connection.stream.StreamRecords;
 import org.springframework.data.redis.util.ByteUtils;
@@ -344,6 +347,41 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands {
 				.map(it -> StreamRecords.newRecord().in(it.getStream()).withId(it.getId()).ofBuffer(it.getBody()));
 	}
 
+	@Override
+	public Flux<CommandResponse<XInfoCommand, XInfoStream>> xInfo(Publisher<XInfoCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
+
+			Assert.notNull(command.getKey(), "Key must not be null!");
+
+			return cmd.xinfoStream(command.getKey()).collectList().map(XInfoStream::fromList)
+					.map(it -> new CommandResponse<>(command, it));
+		}));
+
+	}
+
+	@Override
+	public Flux<CommandResponse<XInfoCommand, Flux<XInfoGroup>>> xInfoGroups(Publisher<XInfoCommand> commands) {
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			Assert.notNull(command.getKey(), "Key must not be null!");
+
+			return new CommandResponse(command, cmd.xinfoGroups(command.getKey()).map(it -> XInfoGroup.fromList((List<Object>) it)));
+		}));
+	}
+
+	@Override
+	public Flux<CommandResponse<XInfoCommand, Flux<XInfoConsumer>>> xInfoConsumers(Publisher<XInfoCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			Assert.notNull(command.getKey(), "Key must not be null!");
+
+			ByteBuffer groupName = ByteUtils.getByteBuffer(command.getGroupName());
+			return new CommandResponse(command, cmd.xinfoConsumers(command.getKey(), groupName).map(it -> new XInfoConsumer(command.getGroupName(), (List<Object>) it)));
+		}));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xRevRange(org.reactivestreams.Publisher)

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

@@ -315,6 +315,10 @@ public class StreamInfo {
 		private XInfoGroup(List<Object> raw) {
 			super(raw, DEFAULT_TYPE_HINTS);
 		}
+		
+		public static XInfoGroup fromList(List<Object> raw) {
+			return new XInfoGroup(raw);
+		}
 
 		/**
 		 * The {@literal consumer group} name. Corresponds to {@literal name}.

src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java

@@ -15,8 +15,6 @@
  */
 package org.springframework.data.redis.core;
 
-import org.springframework.data.redis.connection.stream.PendingMessages;
-import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -36,9 +34,14 @@ import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
 import org.springframework.data.redis.connection.stream.ByteBufferRecord;
 import org.springframework.data.redis.connection.stream.Consumer;
 import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.PendingMessages;
+import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
 import org.springframework.data.redis.connection.stream.ReadOffset;
 import org.springframework.data.redis.connection.stream.Record;
 import org.springframework.data.redis.connection.stream.RecordId;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumer;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroup;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream;
 import org.springframework.data.redis.connection.stream.StreamOffset;
 import org.springframework.data.redis.connection.stream.StreamReadOptions;
 import org.springframework.data.redis.hash.HashMapper;
@@ -185,6 +188,43 @@ class DefaultReactiveStreamOperations<K, HK, HV> implements ReactiveStreamOperat
 		return createMono(connection -> connection.xGroupDestroy(rawKey(key), group));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveStreamOperations#consumers(java.lang.Object)
+	 */
+	@Override
+	public Flux<XInfoConsumer> consumers(K key, String group) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(group, "Group must not be null!");
+
+		return createFlux(connection -> connection.xInfoConsumers(rawKey(key), group));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveStreamOperations#info(java.lang.Object)
+	 */
+	@Override
+	public Mono<XInfoStream> info(K key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return createMono(connection -> connection.xInfo(rawKey(key)));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveStreamOperations#groups(java.lang.Object)
+	 */
+	@Override
+	public Flux<XInfoGroup> groups(K key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return createFlux(connection -> connection.xInfoGroups(rawKey(key)));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.StreamOperations#pending(java.lang.Object, java.lang.String, org.springframework.data.domain.Range, java.lang.Long)
@@ -218,7 +258,6 @@ class DefaultReactiveStreamOperations<K, HK, HV> implements ReactiveStreamOperat
 		return createMono(connection -> connection.xPending(rawKey, group));
 	}
 
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveStreamOperations#size(java.lang.Object)

src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java

@@ -25,6 +25,9 @@ import org.reactivestreams.Publisher;
 import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
 import org.springframework.data.redis.connection.stream.*;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumer;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroup;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream;
 import org.springframework.data.redis.hash.HashMapper;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -194,6 +197,36 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
 	 */
 	Mono<String> destroyGroup(K key, String group);
 
+	/**
+	 * Obtain information about every consumer in a specific {@literal consumer group} for the stream stored at the
+	 * specified {@literal key}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param group name of the {@literal consumer group}.
+	 * @return {@literal null} when used in pipeline / transaction.
+	 * @since 2.3
+	 */
+	Flux<XInfoConsumer> consumers(K key, String group);
+
+	/**
+	 * Obtain information about {@literal consumer groups} associated with the stream stored at the specified
+	 * {@literal key}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @return {@literal null} when used in pipeline / transaction.
+	 * @since 2.3
+	 */
+	Flux<XInfoGroup> groups(K key);
+
+	/**
+	 * Obtain general information about the stream stored at the specified {@literal key}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @return {@literal null} when used in pipeline / transaction.
+	 * @since 2.3
+	 */
+	Mono<XInfoStream> info(K key);
+
 	/**
 	 * Obtain the {@link PendingMessagesSummary} for a given {@literal consumer group}.
 	 *

src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java

@@ -3246,7 +3246,7 @@ public abstract class AbstractConnectionIntegrationTests {
 		assertThat(claimed).containsAll(messages);
 	}
 
-	@Test // DATAREDIS-1084
+	@Test // DATAREDIS-1119
 	@IfProfileValue(name = "redisVersion", value = "5.0")
 	@WithRedisDriver({ RedisDriver.LETTUCE })
 	public void xinfo() {
@@ -3274,7 +3274,7 @@ public abstract class AbstractConnectionIntegrationTests {
 		assertThat(info.lastEntryId()).isEqualTo(lastRecord.getValue());
 	}
 
-	@Test // DATAREDIS-1084
+	@Test // DATAREDIS-1119
 	@IfProfileValue(name = "redisVersion", value = "5.0")
 	@WithRedisDriver({ RedisDriver.LETTUCE })
 	public void xinfoNoGroup() {
@@ -3299,7 +3299,7 @@ public abstract class AbstractConnectionIntegrationTests {
 		assertThat(info.lastEntryId()).isEqualTo(lastRecord.getValue());
 	}
 
-	@Test // DATAREDIS-1084
+	@Test // DATAREDIS-1119
 	@IfProfileValue(name = "redisVersion", value = "5.0")
 	@WithRedisDriver({ RedisDriver.LETTUCE })
 	public void xinfoGroups() {
@@ -3324,7 +3324,7 @@ public abstract class AbstractConnectionIntegrationTests {
 		assertThat(info.get(0).lastDeliveredId()).isEqualTo(lastRecord.getValue());
 	}
 
-	@Test // DATAREDIS-1084
+	@Test // DATAREDIS-1119
 	@IfProfileValue(name = "redisVersion", value = "5.0")
 	@WithRedisDriver({ RedisDriver.LETTUCE })
 	public void xinfoGroupsNoGroup() {
@@ -3342,7 +3342,7 @@ public abstract class AbstractConnectionIntegrationTests {
 
 	}
 
-	@Test // DATAREDIS-1084
+	@Test // DATAREDIS-1119
 	@IfProfileValue(name = "redisVersion", value = "5.0")
 	@WithRedisDriver({ RedisDriver.LETTUCE })
 	public void xinfoGroupsNoConsumer() {
@@ -3365,7 +3365,7 @@ public abstract class AbstractConnectionIntegrationTests {
 		assertThat(info.get(0).lastDeliveredId()).isEqualTo("0-0");
 	}
 
-	@Test // DATAREDIS-1084
+	@Test // DATAREDIS-1119
 	@IfProfileValue(name = "redisVersion", value = "5.0")
 	@WithRedisDriver({ RedisDriver.LETTUCE })
 	public void xinfoConsumers() {
@@ -3389,7 +3389,7 @@ public abstract class AbstractConnectionIntegrationTests {
 		assertThat(info.get(0).idleTimeMs()).isCloseTo(1L, Offset.offset(200L));
 	}
 
-	@Test // DATAREDIS-1084
+	@Test // DATAREDIS-1119
 	@IfProfileValue(name = "redisVersion", value = "5.0")
 	@WithRedisDriver({ RedisDriver.LETTUCE })
 	public void xinfoConsumersNoConsumer() {

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java

@@ -24,6 +24,7 @@ import reactor.test.StepVerifier;
 import java.time.Duration;
 import java.util.Collections;
 
+import org.assertj.core.data.Offset;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -40,6 +41,7 @@ import org.springframework.data.redis.connection.stream.StreamOffset;
  * Integration tests for {@link LettuceReactiveStreamCommands}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  */
 public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsTestsBase {
 
@@ -361,4 +363,115 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT
 				.verifyComplete();
 	}
 
+	@Test // DATAREDIS-1119
+	public void xinfo() {
+
+		String firstRecord = nativeCommands.xadd(KEY_1, KEY_2, VALUE_2);
+		String lastRecord = nativeCommands.xadd(KEY_1, KEY_3, VALUE_3);
+		nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, "0"), "my-group");
+		nativeCommands.xreadgroup(io.lettuce.core.Consumer.from("my-group", "my-consumer"),
+				XReadArgs.StreamOffset.from(KEY_1, ">"));
+
+		connection.streamCommands().xInfo(KEY_1_BBUFFER).as(StepVerifier::create) //
+				.consumeNextWith(info -> {
+					assertThat(info.streamLength()).isEqualTo(2L);
+					assertThat(info.radixTreeKeySize()).isOne();
+					assertThat(info.radixTreeNodesSize()).isEqualTo(2L);
+					assertThat(info.groupCount()).isOne();
+					assertThat(info.lastGeneratedId()).isEqualTo(lastRecord);
+					assertThat(info.firstEntryId()).isEqualTo(firstRecord);
+					assertThat(info.lastEntryId()).isEqualTo(lastRecord);
+				}).verifyComplete();
+	}
+
+	@Test // DATAREDIS-1119
+	public void xinfoNoGroup() {
+
+		String firstRecord = nativeCommands.xadd(KEY_1, KEY_2, VALUE_2);
+		String lastRecord = nativeCommands.xadd(KEY_1, KEY_3, VALUE_3);
+
+		connection.streamCommands().xInfo(KEY_1_BBUFFER).as(StepVerifier::create) //
+				.consumeNextWith(info -> {
+					assertThat(info.streamLength()).isEqualTo(2L);
+					assertThat(info.radixTreeKeySize()).isOne();
+					assertThat(info.radixTreeNodesSize()).isEqualTo(2L);
+					assertThat(info.groupCount()).isZero();
+					assertThat(info.lastGeneratedId()).isEqualTo(lastRecord);
+					assertThat(info.firstEntryId()).isEqualTo(firstRecord);
+					assertThat(info.lastEntryId()).isEqualTo(lastRecord);
+				}).verifyComplete();
+	}
+
+	@Test // DATAREDIS-1119
+	public void xinfoGroups() {
+
+		nativeCommands.xadd(KEY_1, KEY_2, VALUE_2);
+		String lastRecord = nativeCommands.xadd(KEY_1, KEY_3, VALUE_3);
+		nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, "0"), "my-group");
+		nativeCommands.xreadgroup(io.lettuce.core.Consumer.from("my-group", "my-consumer"),
+				XReadArgs.StreamOffset.from(KEY_1, ">"));
+
+		connection.streamCommands().xInfoGroups(KEY_1_BBUFFER).as(StepVerifier::create) //
+				.consumeNextWith(info -> {
+					assertThat(info.groupName()).isEqualTo("my-group");
+					assertThat(info.consumerCount()).isEqualTo(1L);
+					assertThat(info.pendingCount()).isEqualTo(2L);
+					assertThat(info.lastDeliveredId()).isEqualTo(lastRecord);
+				}).verifyComplete();
+	}
+
+	@Test // DATAREDIS-1119
+	public void xinfoGroupsNoGroup() {
+
+		nativeCommands.xadd(KEY_1, KEY_2, VALUE_2);
+		String lastRecord = nativeCommands.xadd(KEY_1, KEY_3, VALUE_3);
+
+		connection.streamCommands().xInfoGroups(KEY_1_BBUFFER).as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	@Test // DATAREDIS-1119
+	public void xinfoGroupsNoConsumer() {
+
+		nativeCommands.xadd(KEY_1, KEY_2, VALUE_2);
+		String lastRecord = nativeCommands.xadd(KEY_1, KEY_3, VALUE_3);
+		nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, "0"), "my-group");
+
+		connection.streamCommands().xInfoGroups(KEY_1_BBUFFER).as(StepVerifier::create) //
+				.consumeNextWith(info -> {
+					assertThat(info.groupName()).isEqualTo("my-group");
+					assertThat(info.consumerCount()).isZero();
+					assertThat(info.pendingCount()).isZero();
+					assertThat(info.lastDeliveredId()).isEqualTo("0-0");
+				}).verifyComplete();
+	}
+
+	@Test // DATAREDIS-1119
+	public void xinfoConsumers() {
+
+		nativeCommands.xadd(KEY_1, KEY_2, VALUE_2);
+		nativeCommands.xadd(KEY_1, KEY_3, VALUE_3);
+		nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, "0"), "my-group");
+		nativeCommands.xreadgroup(io.lettuce.core.Consumer.from("my-group", "my-consumer"),
+				XReadArgs.StreamOffset.from(KEY_1, ">"));
+
+		connection.streamCommands().xInfoConsumers(KEY_1_BBUFFER, "my-group").as(StepVerifier::create) //
+				.consumeNextWith(info -> {
+					assertThat(info.groupName()).isEqualTo("my-group");
+					assertThat(info.consumerName()).isEqualTo("my-consumer");
+					assertThat(info.pendingCount()).isEqualTo(2L);
+					assertThat(info.idleTimeMs()).isCloseTo(1L, Offset.offset(200L));
+				}).verifyComplete();
+	}
+
+	@Test // DATAREDIS-1119
+	public void xinfoConsumersNoConsumer() {
+
+		nativeCommands.xadd(KEY_1, KEY_2, VALUE_2);
+		nativeCommands.xadd(KEY_1, KEY_3, VALUE_3);
+		nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, "0"), "my-group");
+
+		connection.streamCommands().xInfoConsumers(KEY_1_BBUFFER, "my-group").as(StepVerifier::create).verifyComplete();
+	}
+
 }