From 3f5533ea9a7ee728bbe5059112be2dae9c30351a Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Fri, 27 Mar 2020 08:47:27 +0100 Subject: [PATCH] DATAREDIS-1119 - Support XINFO via ReactiveStreamCommands. Original pull request: #519. --- .../connection/ReactiveStreamCommands.java | 98 +++++++++++++++ .../LettuceReactiveStreamCommands.java | 38 ++++++ .../redis/connection/stream/StreamInfo.java | 4 + .../core/DefaultReactiveStreamOperations.java | 45 ++++++- .../redis/core/ReactiveStreamOperations.java | 33 +++++ .../AbstractConnectionIntegrationTests.java | 14 +-- .../LettuceReactiveStreamCommandsTests.java | 113 ++++++++++++++++++ 7 files changed, 335 insertions(+), 10 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java index 30d54a63d..fec9e7b73 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java +++ b/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>> read(Publisher 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 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> xInfo(Publisher 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 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>> xInfoGroups(Publisher 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 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>> xInfoConsumers(Publisher commands); + class GroupCommand extends KeyCommand { private final GroupCommandAction action; diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java index 74cbd6c6e..c0b19f9df 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java +++ b/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> xInfo(Publisher 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>> xInfoGroups(Publisher 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) it))); + })); + } + + @Override + public Flux>> xInfoConsumers(Publisher 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) it))); + })); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xRevRange(org.reactivestreams.Publisher) 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 3af011a1e..b6ae09a29 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 @@ -315,6 +315,10 @@ public class StreamInfo { private XInfoGroup(List raw) { super(raw, DEFAULT_TYPE_HINTS); } + + public static XInfoGroup fromList(List raw) { + return new XInfoGroup(raw); + } /** * The {@literal consumer group} name. Corresponds to {@literal name}. diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java index 8360ed59e..95bae722f 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java +++ b/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 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 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 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 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 implements ReactiveStreamOperat return createMono(connection -> connection.xPending(rawKey, group)); } - /* * (non-Javadoc) * @see org.springframework.data.redis.core.ReactiveStreamOperations#size(java.lang.Object) diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java index 1c4e9305f..447d73f9f 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java +++ b/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 extends HashMapperProvider< */ Mono 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 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 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 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 38e38bede..2de72bc0b 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/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() { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java index 327223d45..837c9c635 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java +++ b/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(); + } + }