diff --git a/src/main/java/org/springframework/data/redis/connection/stream/StreamReadOptions.java b/src/main/java/org/springframework/data/redis/connection/stream/StreamReadOptions.java index 814109ecf..fbbbc8faf 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/StreamReadOptions.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/StreamReadOptions.java @@ -26,7 +26,7 @@ import org.springframework.util.Assert; /** * Options for reading messages from a Redis Stream. - * + * * @author Mark Paluch * @author Christoph Strobl * @see 2.2 @@ -58,18 +58,21 @@ public class StreamReadOptions { } /** - * Enable auto-acknowledgement by setting the {@code NOACK} flag when reading in the context of a consumer group. + * Enable auto-acknowledgement by setting the {@code NOACK} flag when reading in the context of a consumer group. This + * method is an alias for {@link #autoAcknowledge()} for readability reasons. * * @return {@link StreamReadOptions} with {@code noack} applied. */ public StreamReadOptions noack() { - return new StreamReadOptions(block, count, true); + return autoAcknowledge(); } /** - * Enable auto-acknowledgement by setting the {@code NOACK} flag when reading in the context of a consumer group. + * Enable auto-acknowledgement by setting the {@code NOACK} flag when reading in the context of a consumer group. This + * method is an alias for {@link #noack()} for readability reasons. * * @return new instance of {@link StreamReadOptions} with {@code noack} applied. + * @since 2.3 */ public StreamReadOptions autoAcknowledge() { return new StreamReadOptions(block, count, true); @@ -80,7 +83,7 @@ public class StreamReadOptions { * read. * * @param timeout the timeout for the blocking read, must not be {@literal null} or negative. - * @return {@link StreamReadOptions} with {@code block} applied. + * @return new instance of {@link StreamReadOptions} with {@code block} applied. */ public StreamReadOptions block(Duration timeout) { diff --git a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java index 1faf3033f..61da8b692 100644 --- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java @@ -224,7 +224,8 @@ class DefaultStreamMessageListenerContainer> implement ConsumerStreamReadRequest consumerStreamRequest = (ConsumerStreamReadRequest) streamRequest; - StreamReadOptions readOptions = consumerStreamRequest.isAutoAck() ? this.readOptions.autoAcknowledge() : this.readOptions; + StreamReadOptions readOptions = consumerStreamRequest.isAutoAcknowledge() ? this.readOptions.autoAcknowledge() + : this.readOptions; Consumer consumer = consumerStreamRequest.getConsumer(); if (this.containerOptions.getHashMapper() != null) { diff --git a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java index 05d9ec846..94266456c 100644 --- a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java @@ -145,8 +145,7 @@ public interface StreamMessageListenerContainer> exten * @return the new {@link StreamMessageListenerContainer}. */ static > StreamMessageListenerContainer create( - RedisConnectionFactory connectionFactory, - StreamMessageListenerContainerOptions options) { + RedisConnectionFactory connectionFactory, StreamMessageListenerContainerOptions options) { Assert.notNull(connectionFactory, "RedisConnectionFactory must not be null!"); Assert.notNull(options, "StreamMessageListenerContainerOptions must not be null!"); @@ -183,8 +182,7 @@ public interface StreamMessageListenerContainer> exten * {@link org.springframework.data.redis.core.StreamOperations#acknowledge(Object, String, String...)} after * processing. *

- * Errors during {@link org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage} retrieval lead to - * {@link Subscription#cancel() cancellation} of the underlying task. + * Errors during {@link Record} retrieval lead to {@link Subscription#cancel() cancellation} of the underlying task. *

* On {@link StreamMessageListenerContainer#stop()} all {@link Subscription subscriptions} are cancelled prior to * shutting down the container itself. @@ -197,7 +195,8 @@ public interface StreamMessageListenerContainer> exten * @see ReadOffset#lastConsumed() */ default Subscription receive(Consumer consumer, StreamOffset streamOffset, StreamListener listener) { - return register(StreamReadRequest.builder(streamOffset).consumer(consumer).autoAck(false).build(), listener); + return register(StreamReadRequest.builder(streamOffset).consumer(consumer).autoAcknowledge(false).build(), + listener); } /** @@ -207,8 +206,7 @@ public interface StreamMessageListenerContainer> exten *

* Every message is acknowledged when received. *

- * Errors during {@link org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage} retrieval lead to - * {@link Subscription#cancel() cancellation} of the underlying task. + * Errors during {@link Record} retrieval lead to {@link Subscription#cancel() cancellation} of the underlying task. *

* On {@link StreamMessageListenerContainer#stop()} all {@link Subscription subscriptions} are cancelled prior to * shutting down the container itself. @@ -221,7 +219,7 @@ public interface StreamMessageListenerContainer> exten * @see ReadOffset#lastConsumed() */ default Subscription receiveAutoAck(Consumer consumer, StreamOffset streamOffset, StreamListener listener) { - return register(StreamReadRequest.builder(streamOffset).consumer(consumer).autoAck(true).build(), listener); + return register(StreamReadRequest.builder(streamOffset).consumer(consumer).autoAcknowledge(true).build(), listener); } /** @@ -229,15 +227,13 @@ public interface StreamMessageListenerContainer> exten * running} the {@link Subscription} will be added and run immediately, otherwise it'll be scheduled and started once * the container is actually {@link StreamMessageListenerContainer#start() started}. *

- * Errors during {@link org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage} are tested - * against test {@link StreamReadRequest#getCancelSubscriptionOnError() cancellation predicate} whether to cancel the - * underlying task. + * Errors during {@link Record} are tested against test {@link StreamReadRequest#getCancelSubscriptionOnError() + * cancellation predicate} whether to cancel the underlying task. *

* On {@link StreamMessageListenerContainer#stop()} all {@link Subscription subscriptions} are cancelled prior to * shutting down the container itself. *

- * Errors during {@link org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage} retrieval are - * delegated to the given {@link StreamReadRequest#getErrorHandler()}. + * Errors during {@link Record} retrieval are delegated to the given {@link StreamReadRequest#getErrorHandler()}. * * @param streamRequest must not be {@literal null}. * @param listener must not be {@literal null}. @@ -320,7 +316,20 @@ public interface StreamMessageListenerContainer> exten return consumer; } + /** + * @return + * @deprecated since 2.3, use {@link #isAutoAcknowledge()} for improved readability instead. + */ + @Deprecated public boolean isAutoAck() { + return isAutoAcknowledge(); + } + + /** + * @return + * @since 2.3 + */ + public boolean isAutoAcknowledge() { return autoAck; } } @@ -449,8 +458,23 @@ public interface StreamMessageListenerContainer> exten * @param autoAck {@literal true} (default) to auto-acknowledge received messages or {@literal false} for external * acknowledgement. * @return {@code this} {@link ConsumerStreamReadRequestBuilder}. + * @deprecated since 2.3, use {@link #autoAcknowledge(boolean)} instead. */ + @Deprecated public ConsumerStreamReadRequestBuilder autoAck(boolean autoAck) { + return autoAcknowledge(autoAck); + } + + /** + * Configure auto-acknowledgement for stream message consumption. This method is an alias for + * {@link #autoAck(boolean)} for improved readability. + * + * @param autoAck {@literal true} (default) to auto-acknowledge received messages or {@literal false} for external + * acknowledgement. + * @return {@code this} {@link ConsumerStreamReadRequestBuilder}. + * @since 2.3 + */ + public ConsumerStreamReadRequestBuilder autoAcknowledge(boolean autoAck) { this.autoAck = autoAck; return this; @@ -487,10 +511,9 @@ public interface StreamMessageListenerContainer> exten @SuppressWarnings("unchecked") private StreamMessageListenerContainerOptions(Duration pollTimeout, @Nullable Integer batchSize, - RedisSerializer keySerializer, - RedisSerializer hashKeySerializer, RedisSerializer hashValueSerializer, - @Nullable Class targetType, @Nullable HashMapper hashMapper, ErrorHandler errorHandler, - Executor executor) { + RedisSerializer keySerializer, RedisSerializer hashKeySerializer, + RedisSerializer hashValueSerializer, @Nullable Class targetType, + @Nullable HashMapper hashMapper, ErrorHandler errorHandler, Executor executor) { this.pollTimeout = pollTimeout; this.batchSize = batchSize; this.keySerializer = keySerializer; @@ -754,8 +777,7 @@ public interface StreamMessageListenerContainer> exten */ public StreamMessageListenerContainerOptions build() { return new StreamMessageListenerContainerOptions<>(pollTimeout, batchSize, keySerializer, hashKeySerializer, - hashValueSerializer, targetType, hashMapper, - errorHandler, executor); + hashValueSerializer, targetType, hashMapper, errorHandler, executor); } } } diff --git a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java index be0c43723..518f8142b 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java @@ -372,11 +372,11 @@ public class StreamMessageListenerContainerIntegrationTests { } } - private Integer getNumberOfPending(String stream, String group) { + private int getNumberOfPending(String stream, String group) { String value = ((List) ((LettuceConnection) connectionFactory.getConnection()).execute("XPENDING", - new NestedMultiOutput(StringCodec.UTF8), new byte[][] { stream.getBytes(), group.getBytes() })).iterator() - .next().toString(); + new NestedMultiOutput<>(StringCodec.UTF8), new byte[][] { stream.getBytes(), group.getBytes() })).get(0) + .toString(); return NumberUtils.parseNumber(value, Integer.class); }