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 b0f5937c0..89c613622 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java @@ -20,7 +20,6 @@ import reactor.core.publisher.Mono; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; -import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.Map; @@ -59,6 +58,8 @@ import org.springframework.util.ClassUtils; * * @author Mark Paluch * @author Christoph Strobl + * @author Marcin Zielinski + * @author John Blum * @since 2.2 */ class DefaultReactiveStreamOperations implements ReactiveStreamOperations { @@ -144,20 +145,12 @@ class DefaultReactiveStreamOperations implements ReactiveStreamOperat } @Override - public Flux> claim(K key, String group, String newOwner, Duration minIdleTime, - RecordId... recordIds) { + public Flux> claim(K key, String consumerGroup, String newOwner, XClaimOptions xClaimOptions) { - return createFlux(connection -> connection.xClaim(rawKey(key), group, newOwner, minIdleTime, recordIds) + return createFlux(connection -> connection.xClaim(rawKey(key), consumerGroup, newOwner, xClaimOptions) .map(this::deserializeRecord)); } - @Override - public Flux> claim(K key, String group, String newOwner, XClaimOptions xClaimOptions) { - - return createFlux( - connection -> connection.xClaim(rawKey(key), group, newOwner, xClaimOptions).map(this::deserializeRecord)); - } - @Override public Mono delete(K key, RecordId... recordIds) { 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 a209276d3..724460801 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java @@ -16,7 +16,6 @@ package org.springframework.data.redis.core; import java.nio.charset.StandardCharsets; -import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -43,6 +42,7 @@ import org.springframework.data.redis.connection.stream.StreamOffset; import org.springframework.data.redis.connection.stream.StreamReadOptions; import org.springframework.data.redis.hash.HashMapper; import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.data.redis.support.collections.CollectionUtils; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -52,6 +52,8 @@ import org.springframework.util.ClassUtils; * * @author Mark Paluch * @author Christoph Strobl + * @author Marcin Zielinski + * @author John Blum * @since 2.2 */ class DefaultStreamOperations extends AbstractOperations implements StreamOperations { @@ -135,32 +137,16 @@ class DefaultStreamOperations extends AbstractOperations i } @Override - public List> claim(K key, String group, String newOwner, Duration minIdleTime, - RecordId... recordIds) { - byte[] rawKey = rawKey(key); + public List> claim(K key, String consumerGroup, String newOwner, XClaimOptions xClaimOptions) { - return execute(new RecordDeserializingRedisCallback() { + return CollectionUtils.nullSafeList(execute(new RecordDeserializingRedisCallback() { @Nullable @Override List inRedis(RedisConnection connection) { - return connection.streamCommands().xClaim(rawKey, group, newOwner, minIdleTime, recordIds); + return connection.streamCommands().xClaim(rawKey(key), consumerGroup, newOwner, xClaimOptions); } - }); - } - - @Override - public List> claim(K key, String group, String newOwner, XClaimOptions xClaimOptions) { - byte[] rawKey = rawKey(key); - - return execute(new RecordDeserializingRedisCallback() { - - @Nullable - @Override - List inRedis(RedisConnection connection) { - return connection.streamCommands().xClaim(rawKey, group, newOwner, xClaimOptions); - } - }); + })); } @Override 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 2c920c64a..1002a660b 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java @@ -27,11 +27,22 @@ import org.reactivestreams.Publisher; import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.Limit; import org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions; -import org.springframework.data.redis.connection.stream.*; +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.ObjectRecord; +import org.springframework.data.redis.connection.stream.PendingMessage; +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.connection.stream.StreamRecords; import org.springframework.data.redis.hash.HashMapper; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -42,6 +53,8 @@ import org.springframework.util.Assert; * @author Mark Paluch * @author Christoph Strobl * @author Dengliming + * @author Marcin Zielinski + * @author John Blum * @since 2.2 */ public interface ReactiveStreamOperations extends HashMapperProvider { @@ -129,33 +142,48 @@ public interface ReactiveStreamOperations extends HashMapperProvider< Mono add(Record record); /** - * Changes the ownership of a pending message, so that the new owner is the consumer specified as the command argument. - * The message is claimed only if its idle time is greater the minimum idle time specified when calling XCLAIM + * Changes the ownership of a pending message so that the new owner is the consumer specified as + * the command argument. * - * @param key the stream key. - * @param group name of the consumer group. - * @param newOwner name of the consumer claiming the message. - * @param minIdleTime idle time required for a message to be claimed. - * @param recordIds record IDs to be claimed + * The message is claimed only if its idle time (ms) is greater than the {@link Duration minimum idle time} + * specified when calling {@literal XCLAIM}. * - * @return the {@link Flux} of claimed MapRecords. + * @param key {@link K key} to the steam. + * @param consumerGroup {@link String name} of the consumer group. + * @param newOwner {@link String name} of the consumer claiming the message. + * @param minIdleTime {@link Duration minimum idle time} required for a message to be claimed. + * @param recordIds {@link RecordId record IDs} to be claimed. + * @return {@link Flux} of claimed {@link MapRecord MapRecords}. * @see Redis Documentation: XCLAIM + * @see org.springframework.data.redis.connection.stream.MapRecord + * @see org.springframework.data.redis.connection.stream.RecordId + * @see #claim(Object, String, String, XClaimOptions) + * @see reactor.core.publisher.Flux */ - Flux> claim(K key, String group, String newOwner, Duration minIdleTime, RecordId... recordIds); + default Flux> claim(K key, String consumerGroup, String newOwner, Duration minIdleTime, + RecordId... recordIds) { + + return claim(key, consumerGroup, newOwner, XClaimOptions.minIdle(minIdleTime).ids(recordIds)); + } /** - * Changes the ownership of a pending message, so that the new owner is the consumer specified as the command argument. - * The message is claimed only if its idle time is greater the minimum idle time specified when calling XCLAIM + * Changes the ownership of a pending message so that the new owner is the consumer specified as + * the command argument. + + * The message is claimed only if its idle time (ms) is greater than the given {@link Duration minimum idle time} + * specified when calling {@literal XCLAIM}. * - * @param key the stream key. - * @param group name of the consumer group. - * @param newOwner name of the consumer claiming the message. - * @param xClaimOptions additional parameters for the CLAIM call. - * - * @return the {@link Flux} of claimed MapRecords. + * @param key {@link K key} to the steam. + * @param consumerGroup {@link String name} of the consumer group. + * @param newOwner {@link String name} of the consumer claiming the message. + * @param xClaimOptions additional parameters for the {@literal CLAIM} call. + * @return a {@link Flux} of claimed {@link MapRecord MapRecords}. * @see Redis Documentation: XCLAIM + * @see org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions + * @see org.springframework.data.redis.connection.stream.MapRecord + * @see reactor.core.publisher.Flux */ - Flux> claim(K key, String group, String newOwner, XClaimOptions xClaimOptions); + Flux> claim(K key, String consumerGroup, String newOwner, XClaimOptions xClaimOptions); /** * Removes the specified records from the stream. Returns the number of records deleted, that may be different from 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 823afeb1c..0a258aa8a 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/StreamOperations.java @@ -25,11 +25,22 @@ import java.util.Map; import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.Limit; import org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions; -import org.springframework.data.redis.connection.stream.*; +import org.springframework.data.redis.connection.stream.ByteRecord; +import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.ObjectRecord; +import org.springframework.data.redis.connection.stream.PendingMessage; +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.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; import org.springframework.data.redis.hash.HashMapper; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -40,6 +51,8 @@ import org.springframework.util.Assert; * @author Mark Paluch * @author Christoph Strobl * @author Dengliming + * @author Marcin Zielinski + * @author John Blum * @since 2.2 */ public interface StreamOperations extends HashMapperProvider { @@ -122,33 +135,46 @@ public interface StreamOperations extends HashMapperProvider RecordId add(Record record); /** - * Changes the ownership of a pending message, so that the new owner is the consumer specified as the command argument. - * The message is claimed only if its idle time is greater the minimum idle time specified when calling XCLAIM + * Changes the ownership of a pending message so that the new owner is the consumer specified as + * the command argument. * - * @param key the stream key. - * @param group name of the consumer group. - * @param newOwner name of the consumer claiming the message. - * @param minIdleTime idle time required for a message to be claimed. - * @param recordIds record IDs to be claimed + * The message is claimed only if its idle time (ms) is greater than the given {@link Duration minimum idle time} + * specified when calling {@literal XCLAIM}. * - * @return list of claimed MapRecords. + * @param key {@link K key} to the steam. + * @param consumerGroup {@link String name} of the consumer group. + * @param newOwner {@link String name} of the consumer claiming the message. + * @param minIdleTime {@link Duration minimum idle time} required for a message to be claimed. + * @param recordIds {@link RecordId record IDs} to be claimed. + * @return {@link List} of claimed {@link MapRecord MapRecords}. * @see Redis Documentation: XCLAIM + * @see org.springframework.data.redis.connection.stream.MapRecord + * @see org.springframework.data.redis.connection.stream.RecordId + * @see #claim(Object, String, String, XClaimOptions) */ - List> claim(K key, String group, String newOwner, Duration minIdleTime, RecordId... recordIds); + default List> claim(K key, String consumerGroup, String newOwner, Duration minIdleTime, + RecordId... recordIds) { + + return claim(key, consumerGroup, newOwner, XClaimOptions.minIdle(minIdleTime).ids(recordIds)); + } /** - * Changes the ownership of a pending message, so that the new owner is the consumer specified as the command argument. - * The message is claimed only if its idle time is greater the minimum idle time specified when calling XCLAIM + * Changes the ownership of a pending message so that the new owner is the consumer specified as + * the command argument. * - * @param key the stream key. - * @param group name of the consumer group. - * @param newOwner name of the consumer claiming the message. - * @param xClaimOptions additional parameters for the CLAIM call. + * The message is claimed only if its idle time (ms) is greater than the given {@link Duration minimum idle time} + * specified when calling {@literal XCLAIM}. * - * @return list of claimed MapRecords. + * @param key {@link K key} to the steam. + * @param consumerGroup {@link String name} of the consumer group. + * @param newOwner {@link String name} of the consumer claiming the message. + * @param xClaimOptions additional parameters for the {@literal CLAIM} call. + * @return {@link List} of claimed {@link MapRecord MapRecords}. * @see Redis Documentation: XCLAIM + * @see org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions + * @see org.springframework.data.redis.connection.stream.MapRecord */ - List> claim(K key, String group, String newOwner, XClaimOptions xClaimOptions); + List> claim(K key, String consumerGroup, String newOwner, XClaimOptions xClaimOptions); /** * Removes the specified records from the stream. Returns the number of records deleted, that may be different from diff --git a/src/main/java/org/springframework/data/redis/support/collections/CollectionUtils.java b/src/main/java/org/springframework/data/redis/support/collections/CollectionUtils.java index bf6909422..d1951ba00 100644 --- a/src/main/java/org/springframework/data/redis/support/collections/CollectionUtils.java +++ b/src/main/java/org/springframework/data/redis/support/collections/CollectionUtils.java @@ -18,18 +18,22 @@ package org.springframework.data.redis.support.collections; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.List; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.core.RedisOperations; import org.springframework.data.redis.core.SessionCallback; +import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; /** * Utility class used mainly for type conversion by the default collection implementations. Meant for internal use. * * @author Costin Leau + * @author John Blum */ -abstract class CollectionUtils { +public abstract class CollectionUtils { @SuppressWarnings("unchecked") static Collection reverse(Collection c) { @@ -70,4 +74,9 @@ abstract class CollectionUtils { } }); } + + @NonNull + public static List nullSafeList(@Nullable List list) { + return list != null ? list : Collections.emptyList(); + } } diff --git a/src/test/java/org/springframework/data/redis/core/DefaultReactiveStreamOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultReactiveStreamOperationsIntegrationTests.java index 5c758dd2a..9b7ec7d07 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultReactiveStreamOperationsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultReactiveStreamOperationsIntegrationTests.java @@ -59,7 +59,8 @@ import org.springframework.data.redis.test.extension.parametrized.ParameterizedR * Integration tests for {@link DefaultReactiveStreamOperations}. * * @author Mark Paluch - * @auhtor Christoph Strobl + * @author Christoph Strobl + * @author Marcin Zielinski */ @MethodSource("testParams") @SuppressWarnings("unchecked") @@ -361,7 +362,7 @@ public class DefaultReactiveStreamOperationsIntegrationTests { } - @ParameterizedRedisTest // https://github.com/spring-projects/spring-data-redis/issues/2465 + @ParameterizedRedisTest // GH-2465 void claimShouldReadMessageDetails() { K key = keyFactory.instance(); diff --git a/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsIntegrationTests.java index e51e0c056..58e3ee3c3 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsIntegrationTests.java @@ -50,6 +50,7 @@ import org.springframework.data.redis.test.extension.parametrized.ParameterizedR * * @author Mark Paluch * @author Christoph Strobl + * @author Marcin Zielinski */ @MethodSource("testParams") @EnabledOnCommand("XADD") @@ -413,7 +414,7 @@ public class DefaultStreamOperationsIntegrationTests { assertThat(pending.get(0).getTotalDeliveryCount()).isOne(); } - @ParameterizedRedisTest // https://github.com/spring-projects/spring-data-redis/issues/2465 + @ParameterizedRedisTest // GH-2465 void claimShouldReadMessageDetails() { K key = keyFactory.instance();