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 0c19b5832..fd6869d6b 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java @@ -1185,7 +1185,7 @@ public interface ReactiveStreamCommands { return new GroupCommand(getKey(), action, groupName, consumerName, offset); } - public boolean getMkStream() { + public boolean isMkStream() { return this.mkStream; } @@ -1235,6 +1235,7 @@ public interface ReactiveStreamCommands { * @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. + * @since 2.3 */ 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() 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 c45f25b74..6376c663f 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java @@ -473,6 +473,7 @@ public interface RedisStreamCommands { * @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. + * @since 2.3 */ @Nullable String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset, boolean mkStream); 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 a856a34a6..8f8c44460 100644 --- a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java @@ -2101,7 +2101,7 @@ public interface StringRedisConnection extends RedisConnection { /** * Create a consumer group. * - * @param key + * @param key the stream key. * @param readOffset * @param group name of the consumer group. * @since 2.2 @@ -2113,12 +2113,13 @@ public interface StringRedisConnection extends RedisConnection { /** * Create a consumer group. * - * @param key + * @param key the stream 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. + * @since 2.3 */ @Nullable String xGroupCreate(String key, ReadOffset readOffset, String group, boolean mkStream); 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 45898294a..2cb940c0a 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 @@ -192,7 +192,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { return cmd.xgroupCreate(offset, ByteUtils.getByteBuffer(command.getGroupName()), - XGroupCreateArgs.Builder.mkstream( command.getMkStream())) + XGroupCreateArgs.Builder.mkstream(command.isMkStream())) .map(it -> new CommandResponse<>(command, it) ); 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 835d0eceb..90f015ee2 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 @@ -236,18 +236,17 @@ class LettuceStreamCommands implements RedisStreamCommands { XReadArgs.StreamOffset streamOffset = XReadArgs.StreamOffset.from(key, readOffset.getOffset()); if (isPipelined()) { - pipeline(connection - .newLettuceResult(getAsyncConnection().xgroupCreate(streamOffset, LettuceConverters.toBytes(groupName), - XGroupCreateArgs.Builder.mkstream(mkSteam)))); + pipeline(connection.newLettuceResult(getAsyncConnection().xgroupCreate(streamOffset, + LettuceConverters.toBytes(groupName), XGroupCreateArgs.Builder.mkstream(mkSteam)))); return null; } if (isQueueing()) { - transaction(connection - .newLettuceResult(getAsyncConnection().xgroupCreate(streamOffset, LettuceConverters.toBytes(groupName), - XGroupCreateArgs.Builder.mkstream(mkSteam)))); + transaction(connection.newLettuceResult(getAsyncConnection().xgroupCreate(streamOffset, + LettuceConverters.toBytes(groupName), XGroupCreateArgs.Builder.mkstream(mkSteam)))); return null; } - return getConnection().xgroupCreate(streamOffset, LettuceConverters.toBytes(groupName), XGroupCreateArgs.Builder.mkstream(mkSteam)); + return getConnection().xgroupCreate(streamOffset, LettuceConverters.toBytes(groupName), + XGroupCreateArgs.Builder.mkstream(mkSteam)); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -712,7 +711,6 @@ class LettuceStreamCommands implements RedisStreamCommands { return connection.convertLettuceAccessException(ex); } - @SuppressWarnings("unchecked") private static XReadArgs.StreamOffset[] toStreamOffsets(StreamOffset[] streams) { 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 a5ba8f4e0..f628f4a6e 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java @@ -55,7 +55,6 @@ import org.springframework.util.ClassUtils; * * @author Mark Paluch * @author Christoph Strobl - * @author Tugdual Grall * @since 2.2 */ class DefaultReactiveStreamOperations implements ReactiveStreamOperations { @@ -163,17 +162,12 @@ 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, mkStream)); + return createMono(connection -> connection.xGroupCreate(rawKey(key), group, readOffset, true)); } @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 f8ea0aa26..21e3a401f 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java @@ -50,7 +50,6 @@ import org.springframework.util.ClassUtils; * * @author Mark Paluch * @author Christoph Strobl - * @autor Tugdual Grall * @since 2.2 */ class DefaultStreamOperations extends AbstractOperations implements StreamOperations { @@ -160,21 +159,7 @@ class DefaultStreamOperations extends AbstractOperations i public String createGroup(K key, ReadOffset readOffset, String group) { byte[] rawKey = rawKey(key); - 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); - } + return execute(connection -> connection.xGroupCreate(rawKey, group, readOffset, true), true); } /* 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 e0db987b2..63bc4ba50 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java @@ -37,7 +37,6 @@ import org.springframework.util.Assert; * * @author Mark Paluch * @author Christoph Strobl - * @author Tugdual Grall * @since 2.2 */ public interface ReactiveStreamOperations extends HashMapperProvider { @@ -159,7 +158,8 @@ public interface ReactiveStreamOperations extends HashMapperProvider< Mono delete(K key, RecordId... recordIds); /** - * Create a consumer group at the {@link ReadOffset#latest() latest offset}. + * Create a consumer group at the {@link ReadOffset#latest() latest offset}. This command creates the stream if it + * does not already exist. * * @param key the {@literal key} the stream is stored at. * @param group name of the consumer group. @@ -171,20 +171,7 @@ public interface ReactiveStreamOperations extends HashMapperProvider< } /** - * 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. + * Create a consumer group. This command creates the stream if it does not already exist. * * @param key the {@literal key} the stream is stored at. * @param readOffset the {@link ReadOffset} to apply. @@ -193,16 +180,6 @@ 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 41ffaac0a..003c5a456 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/StreamOperations.java @@ -36,7 +36,6 @@ import org.springframework.util.Assert; * * @author Mark Paluch * @author Christoph Strobl - * @author Tugdual Grall * @since 2.2 */ public interface StreamOperations extends HashMapperProvider { @@ -156,7 +155,8 @@ public interface StreamOperations extends HashMapperProvider Long delete(K key, RecordId... recordIds); /** - * Create a consumer group at the {@link ReadOffset#latest() latest offset}. + * Create a consumer group at the {@link ReadOffset#latest() latest offset}. This command creates the stream if it + * does not already exist. * * @param key the {@literal key} the stream is stored at. * @param group name of the consumer group. @@ -167,7 +167,7 @@ public interface StreamOperations extends HashMapperProvider } /** - * Create a consumer group. + * Create a consumer group. This command creates the stream if it does not already exist. * * @param key the {@literal key} the stream is stored at. * @param readOffset the {@link ReadOffset} to apply. @@ -177,18 +177,6 @@ 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 7a069f5c7..e98eae12a 100644 --- a/src/main/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensions.kt +++ b/src/main/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensions.kt @@ -113,15 +113,6 @@ 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]. * @@ -131,15 +122,6 @@ 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 17d144ef3..a5eb7fc82 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -3064,7 +3064,7 @@ public abstract class AbstractConnectionIntegrationTests { @WithRedisDriver({ RedisDriver.LETTUCE }) public void xGroupCreateShouldWorkWithAndWithoutExistingStream() { - actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group",true)); + 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"), @@ -3276,7 +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.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")); 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 f354dac82..57b6ec78c 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,7 +19,6 @@ 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; @@ -29,7 +28,9 @@ import org.assertj.core.data.Offset; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; + import org.springframework.data.domain.Range; +import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.RedisTestProfileValueSource; import org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; 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 571fc251a..e884dbc09 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java @@ -55,7 +55,6 @@ import org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOption * Integration tests for {@link StreamReceiver}. * * @author Mark Paluch - * @author Tugdual Grall */ public class StreamReceiverIntegrationTests { @@ -213,9 +212,8 @@ public class StreamReceiverIntegrationTests { Flux> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"), StreamOffset.create("my-stream", ReadOffset.lastConsumed())); - // required to initialize stream - redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value")); redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from("0-0"), "my-group"); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value")); redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key2", "value2")); messages.as(StepVerifier::create) // @@ -246,9 +244,8 @@ public class StreamReceiverIntegrationTests { Flux> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"), StreamOffset.create("my-stream", ReadOffset.lastConsumed())); - // required to initialize stream - redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value")); redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from("0-0"), "my-group"); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value")); messages.as(StepVerifier::create) // .expectNextCount(1) // @@ -257,31 +254,6 @@ 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/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt index 7398195f9..a8bf2bb8d 100644 --- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt +++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveStreamOperationsExtensionsUnitTests.kt @@ -35,7 +35,6 @@ import reactor.core.publisher.Mono * * @author Mark Paluch * @author Sebastien Deleuze - * @author Tugdual Grall */ class ReactiveStreamOperationsExtensionsUnitTests {