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 f5c59dd48..58c317a17 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 @@ -322,7 +322,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { StreamReadOptions readOptions = command.getReadOptions(); - if (readOptions.getBlock() != null && readOptions.getBlock() > 0) { + if (readOptions.isBlocking()) { return new CommandResponse<>(command, connection.executeDedicated(cmd -> doRead(command, readOptions, cmd))); } 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 4eb49cd80..fffc43eb7 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 @@ -516,7 +516,7 @@ class LettuceStreamCommands implements RedisStreamCommands { XReadArgs.StreamOffset[] streamOffsets = toStreamOffsets(streams); XReadArgs args = StreamConverters.toReadArgs(readOptions); - if (isBlocking(readOptions)) { + if (readOptions.isBlocking()) { try { if (isPipelined()) { @@ -568,7 +568,7 @@ class LettuceStreamCommands implements RedisStreamCommands { XReadArgs args = StreamConverters.toReadArgs(readOptions); io.lettuce.core.Consumer lettuceConsumer = toConsumer(consumer); - if (isBlocking(readOptions)) { + if (readOptions.isBlocking()) { try { if (isPipelined()) { @@ -699,9 +699,6 @@ class LettuceStreamCommands implements RedisStreamCommands { return connection.convertLettuceAccessException(ex); } - private static boolean isBlocking(StreamReadOptions readOptions) { - return readOptions.getBlock() != null && readOptions.getBlock() > 0; - } @SuppressWarnings("unchecked") private static XReadArgs.StreamOffset[] toStreamOffsets(StreamOffset[] streams) { 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 fbbbc8faf..f12c5157c 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 @@ -105,4 +105,12 @@ public class StreamReadOptions { return new StreamReadOptions(block, count, noack); } + + /** + * @return {@literal true} if the arguments indicate a blocking read. + * @since 2.3 + */ + public boolean isBlocking() { + return getBlock() != null && getBlock() >= 0; + } } diff --git a/src/test/java/org/springframework/data/redis/connection/stream/StreamReadOptionsUnitTests.java b/src/test/java/org/springframework/data/redis/connection/stream/StreamReadOptionsUnitTests.java new file mode 100644 index 000000000..ca0cfc766 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/stream/StreamReadOptionsUnitTests.java @@ -0,0 +1,38 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.stream; + +import static org.assertj.core.api.Assertions.*; + +import java.time.Duration; + +import org.junit.Test; + +/** + * Unit tests for {@link StreamReadOptions}. + * + * @author Mark Paluch + */ +public class StreamReadOptionsUnitTests { + + @Test // DATAREDIS-1138 + public void shouldConsiderBlocking() { + + assertThat(StreamReadOptions.empty().isBlocking()).isFalse(); + assertThat(StreamReadOptions.empty().block(Duration.ofSeconds(1)).isBlocking()).isTrue(); + assertThat(StreamReadOptions.empty().block(Duration.ZERO).isBlocking()).isTrue(); + } +}