diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java index 3fa246685..7c4dc34b9 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.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.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 class DefaultStringRedisConnection implements StringRedisConnection, Deco */ @Override public List 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 class DefaultStringRedisConnection implements StringRedisConnection, Deco 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 class DefaultStringRedisConnection implements StringRedisConnection, Deco 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[]) diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java index c0c2466d2..0cf72c98a 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java @@ -35,6 +35,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.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 @@ public interface DefaultedRedisConnection extends RedisConnection { 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 diff --git a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java index 83fe68abd..e03ce7af8 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java @@ -26,6 +26,9 @@ import java.util.stream.Collectors; 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 @@ public interface RedisStreamCommands { @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. * diff --git a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java index 45dde6f5a..d0bd98855 100644 --- a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java @@ -34,6 +34,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.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 @@ public interface StringRedisConnection extends RedisConnection { @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. * diff --git a/src/main/java/org/springframework/data/redis/connection/convert/Converters.java b/src/main/java/org/springframework/data/redis/connection/convert/Converters.java index 4d39b372c..c773e8cfc 100644 --- a/src/main/java/org/springframework/data/redis/connection/convert/Converters.java +++ b/src/main/java/org/springframework/data/redis/connection/convert/Converters.java @@ -17,10 +17,22 @@ package org.springframework.data.redis.connection.convert; 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.RedisGeoCommands.GeoLocation; 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 @@ import org.springframework.util.StringUtils; */ 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 @@ abstract public class Converters { @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 + * @return + * @since 2.3 + */ + public static T parse(Object source, Class 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> 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 sourceCollection = (List) source; + List 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 sourceCollection = ((List) source); + Map 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; } /** diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java index 7568641ee..4eb49cd80 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java @@ -38,6 +38,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.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.util.Assert; @@ -108,6 +111,7 @@ class LettuceStreamCommands implements RedisStreamCommands { RecordId::of)); return null; } + return RecordId.of(getConnection().xadd(record.getStream(), args, record.getValue())); } catch (Exception ex) { @@ -290,6 +294,83 @@ class LettuceStreamCommands implements RedisStreamCommands { } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfo(byte[]) + */ + @Override + public XInfoStream xInfo(byte[] key) { + + Assert.notNull(key, "Key must not be null!"); + + try { + if (isPipelined()) { + pipeline(connection.newLettuceResult(getAsyncConnection().xinfoStream(key), XInfoStream::fromList)); + return null; + } + if (isQueueing()) { + transaction(connection.newLettuceResult(getAsyncConnection().xinfoStream(key), XInfoStream::fromList)); + return null; + } + return XInfoStream.fromList(getConnection().xinfoStream(key)); + } catch (Exception ex) { + throw convertLettuceAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfoGroups(byte[]) + */ + @Override + public XInfoGroups xInfoGroups(byte[] key) { + + Assert.notNull(key, "Key must not be null!"); + + try { + if (isPipelined()) { + pipeline(connection.newLettuceResult(getAsyncConnection().xinfoGroups(key), XInfoGroups::fromList)); + return null; + } + if (isQueueing()) { + transaction(connection.newLettuceResult(getAsyncConnection().xinfoGroups(key), XInfoGroups::fromList)); + return null; + } + return XInfoGroups.fromList(getConnection().xinfoGroups(key)); + } catch (Exception ex) { + throw convertLettuceAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfoConsumers(byte[], java.lang.String) + */ + @Override + public XInfoConsumers xInfoConsumers(byte[] key, String groupName) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(groupName, "GroupName must not be null!"); + + byte[] binaryGroupName = LettuceConverters.toBytes(groupName); + + try { + if (isPipelined()) { + pipeline(connection.newLettuceResult(getAsyncConnection().xinfoConsumers(key, binaryGroupName), + it -> XInfoConsumers.fromList(groupName, it))); + return null; + } + if (isQueueing()) { + transaction(connection.newLettuceResult(getAsyncConnection().xinfoConsumers(key, binaryGroupName), + it -> XInfoConsumers.fromList(groupName, it))); + return null; + } + return XInfoConsumers.fromList(groupName, getConnection().xinfoConsumers(key, binaryGroupName)); + } catch (Exception ex) { + throw convertLettuceAccessException(ex); + } + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisStreamCommands#xLen(byte[]) 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 new file mode 100644 index 000000000..3af011a1e --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java @@ -0,0 +1,496 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.stream; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; +import java.util.stream.Stream; + +import org.springframework.data.redis.connection.convert.Converters; + +/** + * @author Christoph Strobl + * @since 2.3 + */ +public class StreamInfo { + + public static class XInfoObject { + + protected static final Map> DEFAULT_TYPE_HINTS; + + static { + + Map> defaults = new HashMap<>(2); + defaults.put("root", Map.class); + defaults.put("root.*", String.class); + + DEFAULT_TYPE_HINTS = Collections.unmodifiableMap(defaults); + } + + private Map raw; + + private XInfoObject(List raw, Map> typeHints) { + this((Map) Converters.parse(raw, "root", typeHints)); + } + + private XInfoObject(Map raw) { + this.raw = raw; + } + + protected T get(String entry, Class type) { + return type.cast(raw.get(entry)); + } + + public Map getRaw() { + return raw; + } + + @Override + public String toString() { + return "XInfoStream" + raw; + } + } + + /** + * Value object holding general information about a {@literal Redis Stream}. + * + * @author Christoph Strobl + */ + public static class XInfoStream extends XInfoObject { + + private static final Map> typeHints; + + static { + + typeHints = new HashMap<>(DEFAULT_TYPE_HINTS); + typeHints.put("root.first-entry", Map.class); + typeHints.put("root.first-entry.*", Map.class); + typeHints.put("root.first-entry.*.*", Object.class); + typeHints.put("root.last-entry", Map.class); + typeHints.put("root.last-entry.*", Map.class); + typeHints.put("root.last-entry.*.*", Object.class); + } + + private XInfoStream(List raw) { + super(raw, typeHints); + } + + /** + * Factory method to create a new instance of {@link XInfoStream}. + * + * @param source the raw value source. + * @return + */ + public static XInfoStream fromList(List source) { + return new XInfoStream(source); + } + + /** + * Total number of element in the stream. Corresponds to {@literal length}. + * + * @return + */ + public Long streamLength() { + return get("length", Long.class); + } + + /** + * The streams radix tree key size. Corresponds to {@literal radix-tree-keys}. + * + * @return + */ + public Long radixTreeKeySize() { + return get("radix-tree-keys", Long.class); + } + + /** + * Total number of element radix tree nodes. Corresponds to {@literal radix-tree-nodes}. + * + * @return + */ + public Long radixTreeNodesSize() { + return get("radix-tree-nodes", Long.class); + } + + /** + * The number of associated {@literal consumer groups}. Corresponds to {@literal groups}. + * + * @return + */ + public Long groupCount() { + return get("groups", Long.class); + } + + /** + * The last generated id. May not be the same as {@link #lastEntryId()}. Corresponds to + * {@literal last-generated-id}. + * + * @return + */ + public String lastGeneratedId() { + return get("last-generated-id", String.class); + } + + /** + * The id of the streams first entry. Corresponds to {@literal first-entry 1)}. + * + * @return + */ + public String firstEntryId() { + + Map firstEntryMap = get("first-entry", Map.class); + if (firstEntryMap == null) { + return null; + } + + return firstEntryMap.keySet().iterator().next().toString(); + } + + /** + * The streams first entry. Corresponds to {@literal first-entry}. + * + * @return + */ + public Map getFirstEntry() { + + Map lastEntryMap = get("first-entry", Map.class); + if (lastEntryMap == null) { + return null; + } + + return Collections.unmodifiableMap(lastEntryMap); + } + + /** + * The id of the streams last entry. Corresponds to {@literal last-entry 1)}. + * + * @return + */ + public String lastEntryId() { + + Map lastEntryMap = get("last-entry", Map.class); + if (lastEntryMap == null) { + return null; + } + + return lastEntryMap.keySet().iterator().next().toString(); + } + + /** + * The streams first entry. Corresponds to {@literal last-entry}. + * + * @return + */ + public Map getLastEntry() { + + Map firstEntryMap = get("last-entry", Map.class); + if (firstEntryMap == null) { + return null; + } + + return Collections.unmodifiableMap(firstEntryMap); + } + + } + + /** + * Value object holding general information about {@literal consumer groups} associated with a + * {@literal Redis Stream}. + * + * @author Christoph Strobl + */ + public static class XInfoGroups { + + private final List groupInfoList; + + private XInfoGroups(List raw) { + + groupInfoList = new ArrayList<>(); + for (Object entry : raw) { + groupInfoList.add(new XInfoGroup((List) entry)); + } + } + + /** + * Factory method to create a new instance of {@link XInfoGroups}. + * + * @param source the raw value source. + * @return + */ + public static XInfoGroups fromList(List source) { + return new XInfoGroups(source); + } + + /** + * Total number of associated {@literal consumer groups}. + * + * @return zero if none available. + */ + public int groupCount() { + return size(); + } + + /** + * Returns the number of {@link XInfoGroup} available. + * + * @return zero if none available. + * @see #groupCount() + */ + public int size() { + return groupInfoList.size(); + } + + /** + * @return {@literal true} if no groups associated. + */ + public boolean isEmpty() { + return groupInfoList.isEmpty(); + } + + /** + * Returns an iterator over the {@link XInfoGroup} elements. + * + * @return + */ + public Iterator iterator() { + return groupInfoList.iterator(); + } + + /** + * Returns the {@link XInfoGroup} element at the given {@literal index}. + * + * @return the element at the specified position. + * @throws IndexOutOfBoundsException if the index is out of range. + */ + public XInfoGroup get(int index) { + return groupInfoList.get(index); + } + + /** + * Returns a sequential {@code Stream} of {@link XInfoGroup}. + * + * @return + */ + public Stream stream() { + return groupInfoList.stream(); + } + + /** + * Performs the given {@literal action} on every available {@link XInfoGroup} of this {@link XInfoGroups}. + * + * @param action + */ + public void forEach(Consumer action) { + groupInfoList.forEach(action); + } + + @Override + public String toString() { + return "XInfoGroups" + groupInfoList; + } + + } + + public static class XInfoGroup extends XInfoObject { + + private XInfoGroup(List raw) { + super(raw, DEFAULT_TYPE_HINTS); + } + + /** + * The {@literal consumer group} name. Corresponds to {@literal name}. + * + * @return + */ + public String groupName() { + return get("name", String.class); + } + + /** + * The total number of consumers in the {@literal consumer group}. Corresponds to {@literal consumers}. + * + * @return + */ + public Long consumerCount() { + return get("consumers", Long.class); + } + + /** + * The total number of pending messages in the {@literal consumer group}. Corresponds to {@literal pending}. + * + * @return + */ + public Long pendingCount() { + return get("pending", Long.class); + } + + /** + * The id of the last delivered message. Corresponds to {@literal last-delivered-id}. + * + * @return + */ + public String lastDeliveredId() { + return get("last-delivered-id", String.class); + } + } + + public static class XInfoConsumers { + + private final List consumerInfoList; + + public XInfoConsumers(String groupName, List raw) { + + consumerInfoList = new ArrayList<>(); + for (Object entry : raw) { + consumerInfoList.add(new XInfoConsumer(groupName, (List) entry)); + } + } + + public static XInfoConsumers fromList(String groupName, List source) { + return new XInfoConsumers(groupName, source); + } + + /** + * Total number of {@literal consumers} in the {@literal consumer group}. + * + * @return zero if none available. + */ + public int getConsumerCount() { + return consumerInfoList.size(); + } + + /** + * Returns the number of {@link XInfoConsumer} available. + * + * @return zero if none available. + * @see #getConsumerCount() + */ + public int size() { + return consumerInfoList.size(); + } + + /** + * @return {@literal true} if no groups associated. + */ + public boolean isEmpty() { + return consumerInfoList.isEmpty(); + } + + /** + * Returns an iterator over the {@link XInfoConsumer} elements. + * + * @return + */ + public Iterator iterator() { + return consumerInfoList.iterator(); + } + + /** + * Returns the {@link XInfoConsumer} element at the given {@literal index}. + * + * @return the element at the specified position. + * @throws IndexOutOfBoundsException if the index is out of range. + */ + public XInfoConsumer get(int index) { + return consumerInfoList.get(index); + } + + /** + * Returns a sequential {@code Stream} of {@link XInfoConsumer}. + * + * @return + */ + public Stream stream() { + return consumerInfoList.stream(); + } + + /** + * Performs the given {@literal action} on every available {@link XInfoConsumer} of this {@link XInfoConsumers}. + * + * @param action + */ + public void forEach(Consumer action) { + consumerInfoList.forEach(action); + } + + @Override + public String toString() { + return "XInfoConsumers" + consumerInfoList; + } + } + + public static class XInfoConsumer extends XInfoObject { + + private final String groupName; + + public XInfoConsumer(String groupName, List raw) { + + super(raw, DEFAULT_TYPE_HINTS); + this.groupName = groupName; + } + + /** + * The {@literal consumer group} name. + * + * @return + */ + public String groupName() { + return groupName; + } + + /** + * The {@literal consumer} name. Corresponds to {@literal name}. + * + * @return + */ + public String consumerName() { + return get("name", String.class); + } + + /** + * The idle time (in millis). Corresponds to {@literal idle}. + * + * @return + */ + public Long idleTimeMs() { + return get("idle", Long.class); + } + + /** + * The idle time. Corresponds to {@literal idle}. + * + * @return + */ + public Duration idleTime() { + return Duration.ofMillis(idleTimeMs()); + } + + /** + * The number of pending messages. Corresponds to {@literal pending}. + * + * @return + */ + public Long pendingCount() { + return get("pending", Long.class); + } + } +} diff --git a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java index 6b15159c4..aa6f422b1 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.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.Record; 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.hash.HashMapper; @@ -181,6 +184,39 @@ class DefaultStreamOperations extends AbstractOperations i return execute(connection -> connection.xGroupDestroy(rawKey, group), true); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.StreamOperations#info(java.lang.Object) + */ + @Override + public XInfoStream info(K key) { + + byte[] rawKey = rawKey(key); + return execute(connection -> connection.xInfo(rawKey), true); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.StreamOperations#consumers(java.lang.Object, java.lang.String) + */ + @Override + public XInfoConsumers consumers(K key, String group) { + + byte[] rawKey = rawKey(key); + return execute(connection -> connection.xInfoConsumers(rawKey, group), true); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.StreamOperations#groups(java.lang.Object) + */ + @Override + public XInfoGroups groups(K key) { + + byte[] rawKey = rawKey(key); + return execute(connection -> connection.xInfoGroups(rawKey), true); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.core.StreamOperations#pending(java.lang.Object, java.lang.String, org.springframework.data.domain.Range, java.lang.Long) diff --git a/src/main/java/org/springframework/data/redis/core/StreamOperations.java b/src/main/java/org/springframework/data/redis/core/StreamOperations.java index a1137b0c3..5096f560b 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/StreamOperations.java @@ -24,6 +24,9 @@ import java.util.Map; 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.data.redis.hash.HashMapper; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -193,6 +196,36 @@ public interface StreamOperations extends HashMapperProvider @Nullable Boolean 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 + */ + XInfoConsumers 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 + */ + XInfoGroups 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 + */ + XInfoStream info(K key); + /** * Obtain the {@link PendingMessagesSummary} for a given {@literal consumer group}. * 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 32d9dad6d..38e38bede 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -72,6 +72,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.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.core.Cursor; import org.springframework.data.redis.core.ScanOptions; @@ -3243,6 +3246,166 @@ public abstract class AbstractConnectionIntegrationTests { assertThat(claimed).containsAll(messages); } + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xinfo() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1, ReadOffset.lastConsumed()))); + + actual.add(connection.xInfo(KEY_1)); + + List results = getResults(); + assertThat(results).hasSize(5); + RecordId firstRecord = (RecordId) results.get(0); + RecordId lastRecord = (RecordId) results.get(1); + XInfoStream info = (XInfoStream) results.get(4); + + assertThat(info.streamLength()).isEqualTo(2L); + assertThat(info.radixTreeKeySize()).isOne(); + assertThat(info.radixTreeNodesSize()).isEqualTo(2L); + assertThat(info.groupCount()).isOne(); + assertThat(info.lastGeneratedId()).isEqualTo(lastRecord.getValue()); + assertThat(info.firstEntryId()).isEqualTo(firstRecord.getValue()); + assertThat(info.lastEntryId()).isEqualTo(lastRecord.getValue()); + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xinfoNoGroup() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); + + actual.add(connection.xInfo(KEY_1)); + + List results = getResults(); + assertThat(results).hasSize(3); + RecordId firstRecord = (RecordId) results.get(0); + RecordId lastRecord = (RecordId) results.get(1); + XInfoStream info = (XInfoStream) results.get(2); + + assertThat(info.streamLength()).isEqualTo(2L); + assertThat(info.radixTreeKeySize()).isOne(); + assertThat(info.radixTreeNodesSize()).isEqualTo(2L); + assertThat(info.groupCount()).isZero(); + assertThat(info.lastGeneratedId()).isEqualTo(lastRecord.getValue()); + assertThat(info.firstEntryId()).isEqualTo(firstRecord.getValue()); + assertThat(info.lastEntryId()).isEqualTo(lastRecord.getValue()); + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xinfoGroups() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1, ReadOffset.lastConsumed()))); + + actual.add(connection.xInfoGroups(KEY_1)); + + List results = getResults(); + assertThat(results).hasSize(5); + RecordId lastRecord = (RecordId) results.get(1); + XInfoGroups info = (XInfoGroups) results.get(4); + + assertThat(info.size()).isOne(); + 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).lastDeliveredId()).isEqualTo(lastRecord.getValue()); + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xinfoGroupsNoGroup() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); + + actual.add(connection.xInfoGroups(KEY_1)); + + List results = getResults(); + assertThat(results).hasSize(3); + XInfoGroups info = (XInfoGroups) results.get(2); + + assertThat(info.size()).isZero(); + + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xinfoGroupsNoConsumer() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + + actual.add(connection.xInfoGroups(KEY_1)); + + List results = getResults(); + assertThat(results).hasSize(4); + XInfoGroups info = (XInfoGroups) results.get(3); + + assertThat(info.size()).isOne(); + + assertThat(info.get(0).groupName()).isEqualTo("my-group"); + assertThat(info.get(0).consumerCount()).isZero(); + assertThat(info.get(0).pendingCount()).isZero(); + assertThat(info.get(0).lastDeliveredId()).isEqualTo("0-0"); + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xinfoConsumers() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1, ReadOffset.lastConsumed()))); + + actual.add(connection.xInfoConsumers(KEY_1, "my-group")); + + List results = getResults(); + assertThat(results).hasSize(5); + XInfoConsumers info = (XInfoConsumers) results.get(4); + + assertThat(info.size()).isOne(); + assertThat(info.get(0).groupName()).isEqualTo("my-group"); + assertThat(info.get(0).consumerName()).isEqualTo("my-consumer"); + assertThat(info.get(0).pendingCount()).isEqualTo(2L); + assertThat(info.get(0).idleTimeMs()).isCloseTo(1L, Offset.offset(200L)); + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xinfoConsumersNoConsumer() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + actual.add(connection.xInfoConsumers(KEY_1, "my-group")); + + List results = getResults(); + assertThat(results).hasSize(4); + XInfoConsumers info = (XInfoConsumers) results.get(3); + + assertThat(info.size()).isZero(); + } + protected void verifyResults(List expected) { assertThat(getResults()).isEqualTo(expected); }