DATAREDIS-1119 - Support XINFO via RedisStreamCommands.

Original pull request: spring-projects#519.

Author: christophstrobl

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

@@ -41,6 +41,9 @@
 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.XInfoConsumers;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroups;
+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.StringRecord;
@@ -3666,7 +3669,8 @@ public List<RecordId> xClaimJustId(String key, String group, String consumer, XC
  */
 @Override
 public List<StringRecord> xClaim(String key, String group, String consumer, XClaimOptions options) {
-return convertAndReturn(delegate.xClaim(serialize(key), group, consumer, options), listByteMapRecordToStringMapRecordConverter);
+return convertAndReturn(delegate.xClaim(serialize(key), group, consumer, options),
+listByteMapRecordToStringMapRecordConverter);
 }
 
 /*
@@ -3705,6 +3709,33 @@ public Boolean xGroupDestroy(String key, String group) {
 return convertAndReturn(delegate.xGroupDestroy(serialize(key), group), identityConverter);
 }
 
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.StringRedisConnection#xInfo(java.lang.String)
+ */
+@Override
+public XInfoStream xInfo(String key) {
+return convertAndReturn(delegate.xInfo(serialize(key)), identityConverter);
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.StringRedisConnection#xInfoGroups(java.lang.String)
+ */
+@Override
+public XInfoGroups xInfoGroups(String key) {
+return convertAndReturn(delegate.xInfoGroups(serialize(key)), identityConverter);
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.StringRedisConnection#xInfoConsumers(java.lang.String, java.lang.String)
+ */
+@Override
+public XInfoConsumers xInfoConsumers(String key, String groupName) {
+return convertAndReturn(delegate.xInfoConsumers(serialize(key), groupName), identityConverter);
+}
+
 /*
  * (non-Javadoc)
  * @see org.springframework.data.redis.connection.StringRedisConnection#xLen(java.lang.String)
@@ -3875,6 +3906,33 @@ public Boolean xGroupDestroy(byte[] key, String groupName) {
 return delegate.xGroupDestroy(key, groupName);
 }
 
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfo(byte[])
+ */
+@Override
+public XInfoStream xInfo(byte[] key) {
+return delegate.xInfo(key);
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfoGroups(byte[])
+ */
+@Override
+public XInfoGroups xInfoGroups(byte[] key) {
+return delegate.xInfoGroups(key);
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfoConsumers(byte[], java.lang.String)
+ */
+@Override
+public XInfoConsumers xInfoConsumers(byte[] key, String groupName) {
+return delegate.xInfoConsumers(key, groupName);
+}
+
 /*
  * (non-Javadoc)
  * @see org.springframework.data.redis.connection.RedisStreamCommands#xLen(byte[])

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

@@ -35,6 +35,9 @@
 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.XInfoConsumers;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroups;
+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.core.Cursor;
@@ -495,6 +498,27 @@ default Boolean xGroupDestroy(byte[] key, String groupName) {
 return streamCommands().xGroupDestroy(key, groupName);
 }
 
+/** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */
+@Override
+@Deprecated
+default XInfoStream xInfo(byte[] key) {
+return streamCommands().xInfo(key);
+}
+
+/** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */
+@Override
+@Deprecated
+default XInfoGroups xInfoGroups(byte[] key) {
+return streamCommands().xInfoGroups(key);
+}
+
+/** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */
+@Override
+@Deprecated
+default XInfoConsumers xInfoConsumers(byte[] key, String groupName) {
+return streamCommands().xInfoConsumers(key, groupName);
+}
+
 /** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */
 @Override
 @Deprecated

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

@@ -26,6 +26,9 @@
 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.XInfoConsumers;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroups;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
@@ -488,6 +491,39 @@ default Boolean xGroupDelConsumer(byte[] key, String groupName, String consumerN
 @Nullable
 Boolean xGroupDestroy(byte[] key, String groupName);
 
+/**
+ * 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
+ */
+@Nullable
+XInfoStream xInfo(byte[] key);
+
+/**
+ * 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
+ */
+@Nullable
+XInfoGroups xInfoGroups(byte[] key);
+
+/**
+ * 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 {@literal null} when used in pipeline / transaction.
+ * @since 2.3
+ */
+@Nullable
+XInfoConsumers xInfoConsumers(byte[] key, String groupName);
+
 /**
  * Get the length of a stream.
  *

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

@@ -34,6 +34,9 @@
 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.XInfoConsumers;
+import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroups;
+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;
@@ -2128,6 +2131,39 @@ default Long xDel(String key, String... entryIds) {
 @Nullable
 Boolean xGroupDestroy(String key, String group);
 
+/**
+ * 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
+ */
+@Nullable
+XInfoStream xInfo(String key);
+
+/**
+ * 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
+ */
+@Nullable
+XInfoGroups xInfoGroups(String key);
+
+/**
+ * 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 {@literal null} when used in pipeline / transaction.
+ * @since 2.3
+ */
+@Nullable
+XInfoConsumers xInfoConsumers(String key, String groupName);
+
 /**
  * Get the length of a stream.
  *

src/main/java/org/springframework/data/redis/connection/convert/Converters.java

@@ -17,10 +17,22 @@
 
 import lombok.RequiredArgsConstructor;
 
+import java.nio.ByteBuffer;
 import java.time.Duration;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.geo.GeoResult;
@@ -38,8 +50,10 @@
 import org.springframework.data.redis.connection.RedisNode.NodeType;
 import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
 import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.NumberUtils;
 import org.springframework.util.ObjectUtils;
@@ -55,6 +69,9 @@
  */
 abstract public class Converters {
 
+// private static final Log LOGGER = LogFactory.getLog(Converters.class);
+private static final Log LOGGER = LogFactory.getLog(Converters.class);
+
 private static final byte[] ONE = new byte[] { '1' };
 private static final byte[] ZERO = new byte[] { '0' };
 private static final String CLUSTER_NODES_LINE_SEPARATOR = "\n";
@@ -426,7 +443,103 @@ public static <K, V> Converter<Map<K, V>, Properties> mapToPropertiesConverter()
 @Nullable
 public static Duration secondsToDuration(@Nullable Long seconds) {
 return seconds != null ? Duration.ofSeconds(seconds) : null;
+}
+
+/**
+ * Parse a rather generic Redis response, such as a list of something into a meaningful structure applying best effort
+ * conversion of {@code byte[]} and {@link ByteBuffer}.
+ * 
+ * @param source the source to parse
+ * @param targetType eg. {@link Map}, {@link String},...
+ * @param <T>
+ * @return
+ * @since 2.3
+ */
+public static <T> T parse(Object source, Class<T> targetType) {
+return targetType.cast(parse(source, "root", Collections.singletonMap("root", targetType)));
+}
+
+/**
+ * Parse a rather generic Redis response, such as a list of something into a meaningful structure applying best effort
+ * conversion of {@code byte[]} and {@link ByteBuffer} based on the {@literal sourcePath} and a {@literal typeHintMap}
+ *
+ * @param source the source to parse
+ * @param sourcePath the current path (use "root", for level 0).
+ * @param typeHintMap source path to target type hints allowing wildcards ({@literal *}).
+ * @return
+ * @since 2.3
+ */
+public static Object parse(Object source, String sourcePath, Map<String, Class<?>> typeHintMap) {
+
+String path = sourcePath;
+Class<?> targetType = typeHintMap.get(path);
+
+if (targetType == null) {
+
+String alternatePath = sourcePath.contains(".") ? sourcePath.substring(0, sourcePath.lastIndexOf(".")) + ".*"
+: sourcePath;
+targetType = typeHintMap.get(alternatePath);
+if (targetType == null) {
+
+if (sourcePath.endsWith("[]")) {
+targetType = String.class;
+} else {
+targetType = source.getClass();
+}
+
+} else {
+if (targetType == Map.class && sourcePath.endsWith("[]")) {
+targetType = String.class;
+} else {
+path = alternatePath;
+}
+}
+}
+
+if (LOGGER.isDebugEnabled()) {
+LOGGER.debug(String.format("parsing %s (%s) as %s.", sourcePath, path, targetType));
+}
+
+if (targetType == Object.class) {
+return source;
+}
+
+if (ClassUtils.isAssignable(String.class, targetType)) {
+
+if (source instanceof String) {
+return source.toString();
+}
+if (source instanceof byte[]) {
+return new String((byte[]) source);
+}
+if (source instanceof ByteBuffer) {
+return new String(ByteUtils.getBytes((ByteBuffer) source));
+}
+}
+
+if (ClassUtils.isAssignable(List.class, targetType) && source instanceof List) {
+
+List<Object> sourceCollection = (List<Object>) source;
+List<Object> targetList = new ArrayList<>();
+for (int i = 0; i < sourceCollection.size(); i++) {
+targetList.add(parse(sourceCollection.get(i), sourcePath + ".[" + i + "]", typeHintMap));
+}
+return targetList;
+}
+
+if (ClassUtils.isAssignable(Map.class, targetType) && source instanceof List) {
+
+List<Object> sourceCollection = ((List<Object>) source);
+Map<String, Object> targetMap = new LinkedHashMap<>();
+for (int i = 0; i < sourceCollection.size(); i = i + 2) {
+
+String key = parse(sourceCollection.get(i), path + ".[]", typeHintMap).toString();
+targetMap.put(key, parse(sourceCollection.get(i + 1), path + "." + key, typeHintMap));
+}
+return targetMap;
+}
 
+return source;
 }
 
 /**