Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-1138 - Fix detection whether XREAD is blocking.

Browse Source 
We now correctly check if a BLOCK option is configured using a timeout of zero or higher. Previously we only checked if the configured value is greater than zero and didn't consider that a timeout of zero blocks indefinitely.

Original Pull Request: #528
This commit is contained in:
Mark Paluch
2020-05-05 14:26:41 +02:00
committed by Christoph Strobl
parent def86fcf57
commit f4676edbd2
4 changed files with 49 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java
View File

@@ -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)));
}

7
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java
View File

@@ -516,7 +516,7 @@ class LettuceStreamCommands implements RedisStreamCommands {
XReadArgs.StreamOffset<byte[]>[] 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<byte[]> 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<byte[]>[] toStreamOffsets(StreamOffset<byte[]>[] streams) {

8
src/main/java/org/springframework/data/redis/connection/stream/StreamReadOptions.java
View File

@@ -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;
}
}