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 7c4dc34b9..772b8920a 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java @@ -67,6 +67,7 @@ import org.springframework.util.ObjectUtils; * @author Thomas Darimont * @author Mark Paluch * @author Ninad Divadkar + * @author Tugdual Grall */ public class DefaultStringRedisConnection implements StringRedisConnection, DecoratedRedisConnection { @@ -3691,6 +3692,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco return convertAndReturn(delegate.xGroupCreate(serialize(key), group, readOffset), identityConverter); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#xGroupCreate(java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset, java.lang.String, boolean) + */ + @Override + public String xGroupCreate(String key, ReadOffset readOffset, String group, boolean mkStream) { + return convertAndReturn(delegate.xGroupCreate(serialize(key), group, readOffset, mkStream), identityConverter); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.StringRedisConnection#xGroupDelConsumer(java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.Consumer) @@ -3888,6 +3898,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco return delegate.xGroupCreate(key, groupName, readOffset); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupCreate(byte[], org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset, java.lang.String, boolean) + */ + @Override + public String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset, boolean mkStream) { + return delegate.xGroupCreate(key, groupName, readOffset, mkStream); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupDelConsumer(byte[], org.springframework.data.redis.connection.RedisStreamCommands.Consumer) 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 0cf72c98a..684273ab8 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java @@ -54,6 +54,7 @@ import org.springframework.lang.Nullable; * * @author Christoph Strobl * @author Mark Paluch + * @author Tugdual Grall * @since 2.0 */ public interface DefaultedRedisConnection extends RedisConnection { @@ -484,6 +485,13 @@ public interface DefaultedRedisConnection extends RedisConnection { return streamCommands().xGroupCreate(key, groupName, readOffset); } + /** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */ + @Override + @Deprecated + default String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset, boolean mkStream) { + return streamCommands().xGroupCreate(key, groupName, readOffset, mkStream); + } + /** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */ @Override @Deprecated 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 e51ffaee5..0c19b5832 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java @@ -56,6 +56,7 @@ import org.springframework.util.StringUtils; * * @author Mark Paluch * @author Christoph Strobl + * @author Tugdual Grall * @since 2.2 */ public interface ReactiveStreamCommands { @@ -1134,15 +1135,22 @@ public interface ReactiveStreamCommands { private final @Nullable String groupName; private final @Nullable String consumerName; private final @Nullable ReadOffset offset; + private final boolean mkStream; public GroupCommand(@Nullable ByteBuffer key, GroupCommandAction action, @Nullable String groupName, - @Nullable String consumerName, @Nullable ReadOffset offset) { + @Nullable String consumerName, @Nullable ReadOffset offset, boolean mkStream) { super(key); this.action = action; this.groupName = groupName; this.consumerName = consumerName; this.offset = offset; + this.mkStream = mkStream; + } + + public GroupCommand(@Nullable ByteBuffer key, GroupCommandAction action, @Nullable String groupName, + @Nullable String consumerName, @Nullable ReadOffset offset) { + this(key, action, groupName, consumerName, offset, false); } public static GroupCommand createGroup(String group) { @@ -1161,6 +1169,10 @@ public interface ReactiveStreamCommands { return new GroupCommand(null, GroupCommandAction.DELETE_CONSUMER, consumer.getGroup(), consumer.getName(), null); } + public GroupCommand makeStream(boolean mkStream) { + return new GroupCommand(getKey(), action, groupName, consumerName, offset,mkStream); + } + public GroupCommand at(ReadOffset offset) { return new GroupCommand(getKey(), action, groupName, consumerName, offset); } @@ -1173,6 +1185,10 @@ public interface ReactiveStreamCommands { return new GroupCommand(getKey(), action, groupName, consumerName, offset); } + public boolean getMkStream() { + return this.mkStream; + } + @Nullable public ReadOffset getReadOffset() { return this.offset; @@ -1211,6 +1227,20 @@ public interface ReactiveStreamCommands { .map(CommandResponse::getOutput); } + /** + * Create a consumer group. + * + * @param key key the {@literal key} the stream is stored at. + * @param groupName name of the consumer group to create. + * @param readOffset the offset to start at. + * @param mkStream if true the group will create the stream if needed (MKSTREAM) + * @return the {@link Mono} emitting {@literal ok} if successful. + */ + default Mono xGroupCreate(ByteBuffer key, String groupName, ReadOffset readOffset, boolean mkStream) { + return xGroup(Mono.just(GroupCommand.createGroup(groupName).forStream(key).at(readOffset).makeStream(mkStream))).next() + .map(CommandResponse::getOutput); + } + /** * Delete a consumer from a consumer group. * 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 97eccc122..c45f25b74 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java @@ -39,6 +39,7 @@ import org.springframework.util.StringUtils; * * @author Mark Paluch * @author Christoph Strobl + * @author Tugdual Grall * @see Redis Documentation - Streams * @since 2.2 */ @@ -464,6 +465,18 @@ public interface RedisStreamCommands { @Nullable String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset); + /** + * Create a consumer group. + * + * @param key the {@literal key} the stream is stored at. + * @param groupName name of the consumer group to create. + * @param readOffset the offset to start at. + * @param mkStream if true the group will create the stream if not already present (MKSTREAM) + * @return {@literal ok} if successful. {@literal null} when used in pipeline / transaction. + */ + @Nullable + String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset, boolean mkStream); + /** * Delete a consumer from a consumer group. * 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 d0bd98855..a856a34a6 100644 --- a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java @@ -60,6 +60,7 @@ import org.springframework.lang.Nullable; * @author David Liu * @author Mark Paluch * @author Ninad Divadkar + * @author Tugdual Grall * @see RedisCallback * @see RedisSerializer * @see StringRedisTemplate @@ -2109,6 +2110,19 @@ public interface StringRedisConnection extends RedisConnection { @Nullable String xGroupCreate(String key, ReadOffset readOffset, String group); + /** + * Create a consumer group. + * + * @param key + * @param readOffset + * @param group name of the consumer group. + * @param mkStream if true the group will create the stream if needed (MKSTREAM) + * @since + * @return {@literal true} if successful. {@literal null} when used in pipeline / transaction. + */ + @Nullable + String xGroupCreate(String key, ReadOffset readOffset, String group, boolean mkStream); + /** * Delete a consumer from a consumer group. * 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 58c317a17..45898294a 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 @@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.XAddArgs; import io.lettuce.core.XClaimArgs; +import io.lettuce.core.XGroupCreateArgs; import io.lettuce.core.XReadArgs; import io.lettuce.core.XReadArgs.StreamOffset; import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands; @@ -51,6 +52,7 @@ import org.springframework.util.Assert; * {@link ReactiveStreamCommands} implementation for {@literal Lettuce}. * * @author Mark Paluch + * @author Tugdual Grall * @since 2.2 */ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { @@ -188,8 +190,12 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { StreamOffset offset = StreamOffset.from(command.getKey(), command.getReadOffset().getOffset()); - return cmd.xgroupCreate(offset, ByteUtils.getByteBuffer(command.getGroupName())) - .map(it -> new CommandResponse<>(command, it)); + return cmd.xgroupCreate(offset, + ByteUtils.getByteBuffer(command.getGroupName()), + XGroupCreateArgs.Builder.mkstream( command.getMkStream())) + .map(it -> + new CommandResponse<>(command, it) + ); } if (command.getAction().equals(GroupCommandAction.DELETE_CONSUMER)) { 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 fffc43eb7..835d0eceb 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 @@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.XAddArgs; import io.lettuce.core.XClaimArgs; +import io.lettuce.core.XGroupCreateArgs; import io.lettuce.core.XReadArgs; import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands; import io.lettuce.core.cluster.api.sync.RedisClusterCommands; @@ -47,6 +48,7 @@ import org.springframework.util.Assert; /** * @author Mark Paluch + * @author Tugdual Grall * @since 2.2 */ @RequiredArgsConstructor @@ -216,6 +218,15 @@ class LettuceStreamCommands implements RedisStreamCommands { */ @Override public String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset) { + return xGroupCreate(key, groupName, readOffset, false); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupCreate(byte[], org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset, java.lang.String, boolean) + */ + @Override + public String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset, boolean mkSteam) { Assert.notNull(key, "Key must not be null!"); Assert.hasText(groupName, "Group name must not be null or empty!"); @@ -226,15 +237,17 @@ class LettuceStreamCommands implements RedisStreamCommands { if (isPipelined()) { pipeline(connection - .newLettuceResult(getAsyncConnection().xgroupCreate(streamOffset, LettuceConverters.toBytes(groupName)))); + .newLettuceResult(getAsyncConnection().xgroupCreate(streamOffset, LettuceConverters.toBytes(groupName), + XGroupCreateArgs.Builder.mkstream(mkSteam)))); return null; } if (isQueueing()) { transaction(connection - .newLettuceResult(getAsyncConnection().xgroupCreate(streamOffset, LettuceConverters.toBytes(groupName)))); + .newLettuceResult(getAsyncConnection().xgroupCreate(streamOffset, LettuceConverters.toBytes(groupName), + XGroupCreateArgs.Builder.mkstream(mkSteam)))); return null; } - return getConnection().xgroupCreate(streamOffset, LettuceConverters.toBytes(groupName)); + return getConnection().xgroupCreate(streamOffset, LettuceConverters.toBytes(groupName), XGroupCreateArgs.Builder.mkstream(mkSteam)); } catch (Exception ex) { throw convertLettuceAccessException(ex); } 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 95bae722f..a5ba8f4e0 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java @@ -55,6 +55,7 @@ import org.springframework.util.ClassUtils; * * @author Mark Paluch * @author Christoph Strobl + * @author Tugdual Grall * @since 2.2 */ class DefaultReactiveStreamOperations implements ReactiveStreamOperations { @@ -162,12 +163,17 @@ class DefaultReactiveStreamOperations implements ReactiveStreamOperat @Override public Mono createGroup(K key, ReadOffset readOffset, String group) { + return createMono(connection -> connection.xGroupCreate(rawKey(key), group, readOffset, false)); + } + + @Override + public Mono createGroup(K key, ReadOffset readOffset, String group, boolean mkStream) { Assert.notNull(key, "Key must not be null!"); Assert.notNull(readOffset, "ReadOffset must not be null!"); Assert.notNull(group, "Group must not be null!"); - return createMono(connection -> connection.xGroupCreate(rawKey(key), group, readOffset)); + return createMono(connection -> connection.xGroupCreate(rawKey(key), group, readOffset, mkStream)); } @Override 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 aa6f422b1..f8ea0aa26 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java @@ -50,6 +50,7 @@ import org.springframework.util.ClassUtils; * * @author Mark Paluch * @author Christoph Strobl + * @autor Tugdual Grall * @since 2.2 */ class DefaultStreamOperations extends AbstractOperations implements StreamOperations { @@ -162,6 +163,20 @@ class DefaultStreamOperations extends AbstractOperations i return execute(connection -> connection.xGroupCreate(rawKey, group, readOffset), true); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.StreamOperations#createGroup(java.lang.Object, org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset, java.lang.String, boolean) + */ + @Override + public String createGroup(K key, ReadOffset readOffset, String group, boolean mkStream) { + byte[] rawKey = rawKey(key); + if (!mkStream) { + return createGroup(key, readOffset, group); + } else { + return execute(connection -> connection.xGroupCreate(rawKey, group, readOffset, true), true); + } + } + /* * (non-Javadoc) * @see org.springframework.data.redis.core.StreamOperations#deleteConsumer(java.lang.Object, org.springframework.data.redis.connection.RedisStreamCommands.Consumer) 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 447d73f9f..e0db987b2 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java @@ -37,6 +37,7 @@ import org.springframework.util.Assert; * * @author Mark Paluch * @author Christoph Strobl + * @author Tugdual Grall * @since 2.2 */ public interface ReactiveStreamOperations extends HashMapperProvider { @@ -169,6 +170,19 @@ public interface ReactiveStreamOperations extends HashMapperProvider< return createGroup(key, ReadOffset.latest(), group); } + /** + * Create a consumer group at the {@link ReadOffset#latest() latest offset}. + * + * @param key the {@literal key} the stream is stored at. + * @param group name of the consumer group. + * @param mkStream if true the group will create the stream if not already present (MKSTREAM) + * @return the {@link Mono} emitting {@literal OK} if successful.. {@literal null} when used in pipeline / + * transaction. + */ + default Mono createGroup(K key, String group, boolean mkStream) { + return createGroup(key, ReadOffset.latest(), group, mkStream); + } + /** * Create a consumer group. * @@ -179,6 +193,16 @@ public interface ReactiveStreamOperations extends HashMapperProvider< */ Mono createGroup(K key, ReadOffset readOffset, String group); + /** + * Create a consumer group. + * + * @param key the {@literal key} the stream is stored at. + * @param readOffset the {@link ReadOffset} to apply. + * @param group name of the consumer group. + * @param mkStream if true the group will create the stream if needed (MKSTREAM) + * @return the {@link Mono} emitting {@literal OK} if successful. + */ + Mono createGroup(K key, ReadOffset readOffset, String group, boolean mkStream); /** * Delete a consumer from a consumer group. * 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 5096f560b..41ffaac0a 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/StreamOperations.java @@ -36,6 +36,7 @@ import org.springframework.util.Assert; * * @author Mark Paluch * @author Christoph Strobl + * @author Tugdual Grall * @since 2.2 */ public interface StreamOperations extends HashMapperProvider { @@ -176,6 +177,18 @@ public interface StreamOperations extends HashMapperProvider @Nullable String createGroup(K key, ReadOffset readOffset, String group); + /** + * Create a consumer group. + * + * @param key the {@literal key} the stream is stored at. + * @param readOffset the {@link ReadOffset} to apply. + * @param group name of the consumer group. + * @param mkStream if true the group will create the stream if needed (MKSTREAM) + * @return {@literal OK} if successful. {@literal null} when used in pipeline / transaction. + */ + @Nullable + String createGroup(K key, ReadOffset readOffset, String group, boolean mkStream); + /** * Delete a consumer from a consumer group. * diff --git a/src/main/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensions.kt b/src/main/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensions.kt index e98eae12a..7a069f5c7 100644 --- a/src/main/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensions.kt +++ b/src/main/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensions.kt @@ -113,6 +113,15 @@ suspend fun ReactiveStreamOperations.de suspend fun ReactiveStreamOperations.createGroupAndAwait(key: K, group: String): String = createGroup(key, group).awaitSingle() +/** + * Coroutines variant of [ReactiveStreamOperations.createGroup]. + * + * @author Mark Paluch + * @since 2.2 + */ +suspend fun ReactiveStreamOperations.createGroupAndAwait(key: K, group: String, mkStream: Boolean): String = + createGroup(key, group, mkStream).awaitSingle() + /** * Coroutines variant of [ReactiveStreamOperations.createGroup]. * @@ -122,6 +131,15 @@ suspend fun ReactiveStreamOperations.cr suspend fun ReactiveStreamOperations.createGroupAndAwait(key: K, readOffset: ReadOffset, group: String): String = createGroup(key, readOffset, group).awaitSingle() +/** + * Coroutines variant of [ReactiveStreamOperations.createGroup]. + * + * @author Tugdual Grall + * @since + */ +suspend fun ReactiveStreamOperations.createGroupAndAwait(key: K, readOffset: ReadOffset, group: String, mkStream: Boolean): String = + createGroup(key, readOffset, group, mkStream).awaitSingle() + /** * Coroutines variant of [ReactiveStreamOperations.deleteConsumer]. * 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 ce42cf4a8..17d144ef3 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -98,6 +98,7 @@ import org.springframework.test.annotation.ProfileValueSourceConfiguration; * @author Christoph Strobl * @author Thomas Darimont * @author Mark Paluch + * @author Tugdual Grall */ @ProfileValueSourceConfiguration(RedisTestProfileValueSource.class) public abstract class AbstractConnectionIntegrationTests { @@ -3058,6 +3059,29 @@ public abstract class AbstractConnectionIntegrationTests { assertThat((List) results.get(3)).isEmpty(); } + @Test // DATAREDIS-864 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xGroupCreateShouldWorkWithAndWithoutExistingStream() { + + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group",true)); + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + + actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1, ReadOffset.lastConsumed()))); + actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1, ReadOffset.lastConsumed()))); + + List results = getResults(); + + List> messages = (List) results.get(2); + + assertThat(messages.get(0).getStream()).isEqualTo(KEY_1); + assertThat(messages.get(0).getValue()).isEqualTo(Collections.singletonMap(KEY_2, VALUE_2)); + + assertThat((List) results.get(3)).isEmpty(); + } + @Test // DATAREDIS-864 @IfProfileValue(name = "redisVersion", value = "5.0") @WithRedisDriver({ RedisDriver.LETTUCE }) @@ -3252,6 +3276,7 @@ public abstract class AbstractConnectionIntegrationTests { @WithRedisDriver({ RedisDriver.LETTUCE }) public void xinfo() { + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group-without-stream",true)); 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")); @@ -3261,15 +3286,15 @@ public abstract class AbstractConnectionIntegrationTests { 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(results).hasSize(6); + RecordId firstRecord = (RecordId) results.get(1); + RecordId lastRecord = (RecordId) results.get(2); + XInfoStream info = (XInfoStream) results.get(5); assertThat(info.streamLength()).isEqualTo(2L); assertThat(info.radixTreeKeySize()).isOne(); assertThat(info.radixTreeNodesSize()).isEqualTo(2L); - assertThat(info.groupCount()).isOne(); + assertThat(info.groupCount()).isEqualTo(2L); assertThat(info.lastGeneratedId()).isEqualTo(lastRecord.getValue()); assertThat(info.firstEntryId()).isEqualTo(firstRecord.getValue()); assertThat(info.lastEntryId()).isEqualTo(lastRecord.getValue()); 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 837c9c635..f354dac82 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 @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.*; import static org.junit.Assume.*; import io.lettuce.core.XReadArgs; +import org.springframework.data.redis.RedisSystemException; import reactor.test.StepVerifier; import java.time.Duration; @@ -42,6 +43,7 @@ import org.springframework.data.redis.connection.stream.StreamOffset; * * @author Mark Paluch * @author Christoph Strobl + * @author Tugdual Grall */ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsTestsBase { @@ -188,6 +190,19 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT .verifyComplete(); } + @Test // DATAREDIS-864 + public void xGroupCreateShouldCreateGroupBeforeStream() { + connection.streamCommands().xGroupCreate(KEY_1_BBUFFER, "group-1", ReadOffset.latest(), false) + .as(StepVerifier::create) // + .expectError(RedisSystemException.class) // + .verify(); + + connection.streamCommands().xGroupCreate(KEY_1_BBUFFER, "group-1", ReadOffset.latest(), true) // + .as(StepVerifier::create) // + .expectNext("OK") // + .verifyComplete(); + } + @Test // DATAREDIS-864 @Ignore("commands sent correctly - however lettuce returns false") public void xGroupDelConsumerShouldRemoveConsumer() { diff --git a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java index 354af8d65..571fc251a 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java @@ -55,6 +55,7 @@ import org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOption * Integration tests for {@link StreamReceiver}. * * @author Mark Paluch + * @author Tugdual Grall */ public class StreamReceiverIntegrationTests { @@ -256,6 +257,31 @@ public class StreamReceiverIntegrationTests { .verify(Duration.ofSeconds(5)); } + @Test // DATAREDIS-864 + public void shouldCreateGroupWithOrWithoutExistingStream() { + StreamReceiver> receiver = StreamReceiver.create(connectionFactory); + + Flux> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"), + StreamOffset.create("my-stream", ReadOffset.lastConsumed())); + + // required to initialize stream + redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from("0-0"), "my-group", true); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value")); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key2", "value2")); + + messages.as(StepVerifier::create) // + .consumeNextWith(it -> { + assertThat(it.getStream()).isEqualTo("my-stream"); + assertThat(it.getValue()).containsValue("value"); + }).consumeNextWith(it -> { + + assertThat(it.getStream()).isEqualTo("my-stream"); + assertThat(it.getValue()).containsValue("value2"); + }) // + .thenCancel() // + .verify(Duration.ofSeconds(5)); + } + @Data @AllArgsConstructor static class LoginEvent { diff --git a/src/test/java/org/springframework/data/redis/test/util/RedisClientRule.java b/src/test/java/org/springframework/data/redis/test/util/RedisClientRule.java index 25db1f31c..ddd3b3efa 100644 --- a/src/test/java/org/springframework/data/redis/test/util/RedisClientRule.java +++ b/src/test/java/org/springframework/data/redis/test/util/RedisClientRule.java @@ -27,6 +27,7 @@ import org.springframework.data.redis.connection.RedisConnectionFactory; * * @author Christoph Strobl * @author Thomas Darimont + * @author Tugdual Grall */ public abstract class RedisClientRule implements TestRule { @@ -50,7 +51,7 @@ public abstract class RedisClientRule implements TestRule { return; } } - throw new AssumptionViolatedException("not a vaild redis connection for driver: " + redisConnectionFactory); + throw new AssumptionViolatedException("not a valid redis connection for driver: " + redisConnectionFactory); } base.evaluate(); diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt index a8bf2bb8d..7398195f9 100644 --- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt +++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt @@ -35,6 +35,7 @@ import reactor.core.publisher.Mono * * @author Mark Paluch * @author Sebastien Deleuze + * @author Tugdual Grall */ class ReactiveStreamOperationsExtensionsUnitTests {