From bdf15de653d59f7d4501f027ef4ea03157ee1923 Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Mon, 28 Jun 2021 12:42:05 +0200 Subject: [PATCH] Support precise sub second timeout for BZPOPMIN, BZPOPMAX. In cases where the timeout is defined in msec we calculate a precise timeout for the BZPOP* operations. #2102 will bring this behaviour also to List commands. Original Pull Request: #2088 --- .../connection/ReactiveZSetCommands.java | 50 +++++++++++++------ .../redis/connection/RedisZSetCommands.java | 8 +-- .../lettuce/LettuceReactiveZSetCommands.java | 19 ++++++- .../lettuce/LettuceZSetCommands.java | 17 +++++++ .../data/redis/core/ZSetOperations.java | 16 +++--- .../AbstractConnectionIntegrationTests.java | 10 +++- ...ultStringRedisConnectionPipelineTests.java | 2 + ...tStringRedisConnectionPipelineTxTests.java | 1 + .../DefaultStringRedisConnectionTests.java | 1 + .../DefaultStringRedisConnectionTxTests.java | 1 + .../AbstractRedisZSetTestIntegration.java | 4 +- 11 files changed, 96 insertions(+), 33 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java index 7be989992..bb3f01ced 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java @@ -25,10 +25,10 @@ import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Optional; +import java.util.concurrent.TimeUnit; import java.util.function.Function; import org.reactivestreams.Publisher; - import org.springframework.data.domain.Range; import org.springframework.data.domain.Sort.Direction; import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; @@ -163,8 +163,7 @@ public interface ReactiveZSetCommands { } /** - * Applies {@literal GT} mode. Constructs a new command - * instance with all previously configured properties. + * Applies {@literal GT} mode. Constructs a new command instance with all previously configured properties. * * @return a new {@link ZAddCommand} with {@literal incr} applied. * @since 2.5 @@ -174,8 +173,7 @@ public interface ReactiveZSetCommands { } /** - * Applies {@literal LT} mode. Constructs a new command - * instance with all previously configured properties. + * Applies {@literal LT} mode. Constructs a new command instance with all previously configured properties. * * @return a new {@link ZAddCommand} with {@literal incr} applied. * @since 2.5 @@ -206,7 +204,6 @@ public interface ReactiveZSetCommands { } /** - * * @return {@literal true} if {@literal GT} is set. * @since 2.5 */ @@ -1337,15 +1334,18 @@ public interface ReactiveZSetCommands { private final PopDirection direction; - private final Duration timeout; + private final @Nullable TimeUnit timeUnit; + private final @Nullable Long timeout; private final long count; - private BZPopCommand(@Nullable ByteBuffer key, Duration timeout, long count, PopDirection direction) { + private BZPopCommand(@Nullable ByteBuffer key, @Nullable Long timeout, @Nullable TimeUnit timeUnit, long count, + PopDirection direction) { super(key); this.count = count; this.timeout = timeout; + this.timeUnit = timeUnit; this.direction = direction; } @@ -1355,7 +1355,7 @@ public interface ReactiveZSetCommands { * @return a new {@link BZPopCommand} for min pop ({@literal ZPOPMIN}). */ public static BZPopCommand min() { - return new BZPopCommand(null, null, 0, PopDirection.MIN); + return new BZPopCommand(null, null, null, 0, PopDirection.MIN); } /** @@ -1364,7 +1364,7 @@ public interface ReactiveZSetCommands { * @return a new {@link BZPopCommand} for max pop ({@literal ZPOPMAX}). */ public static BZPopCommand max() { - return new BZPopCommand(null, null, 0, PopDirection.MAX); + return new BZPopCommand(null, null, null, 0, PopDirection.MAX); } /** @@ -1377,7 +1377,7 @@ public interface ReactiveZSetCommands { Assert.notNull(key, "Key must not be null!"); - return new BZPopCommand(key, timeout, count, direction); + return new BZPopCommand(key, timeout, timeUnit, count, direction); } /** @@ -1387,7 +1387,7 @@ public interface ReactiveZSetCommands { * @return a new {@link BZPopCommand} with {@literal value} applied. */ public BZPopCommand count(long count) { - return new BZPopCommand(getKey(), timeout, count, direction); + return new BZPopCommand(getKey(), timeout, timeUnit, count, direction); } /** @@ -1400,7 +1400,21 @@ public interface ReactiveZSetCommands { Assert.notNull(timeout, "Timeout must not be null!"); - return new BZPopCommand(getKey(), timeout, count, direction); + return blockingFor(timeout.toMillis(), TimeUnit.MILLISECONDS); + } + + /** + * Applies a {@link Duration timeout}. Constructs a new command instance with all previously configured properties. + * + * @param timeout value. + * @param timeout must not be {@literal null}. + * @return a new {@link BZPopCommand} with {@link Duration timeout} applied. + */ + public BZPopCommand blockingFor(long timeout, TimeUnit timeUnit) { + + Assert.notNull(timeUnit, "TimeUnit must not be null!"); + + return new BZPopCommand(getKey(), timeout, timeUnit, count, direction); } /** @@ -1410,10 +1424,16 @@ public interface ReactiveZSetCommands { return direction; } - public Duration getTimeout() { + @Nullable + public Long getTimeout() { return timeout; } + @Nullable + public TimeUnit getTimeUnit() { + return timeUnit; + } + public long getCount() { return count; } @@ -1664,7 +1684,7 @@ public interface ReactiveZSetCommands { /** * Creates a new {@link ZMScoreCommand} given a {@link List members}. * - * @param member must not be {@literal null}. + * @param members must not be {@literal null}. * @return a new {@link ZMScoreCommand} for {@link List} of members. */ public static ZMScoreCommand scoreOf(Collection members) { diff --git a/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java index ddb208a89..9e575e8b7 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java @@ -997,8 +997,8 @@ public interface RedisZSetCommands { Set zPopMin(byte[] key, long count); /** - * Remove and return the value with its score having the lowest score from sorted set at {@code key}. Blocks - * connection until element available or {@code timeout} reached. + * Remove and return the value with its score having the lowest score from sorted set at {@code key}.
+ * Blocks connection until element available or {@code timeout} reached. * * @param key must not be {@literal null}. * @param timeout @@ -1034,8 +1034,8 @@ public interface RedisZSetCommands { Set zPopMax(byte[] key, long count); /** - * Remove and return the value with its score having the highest score from sorted set at {@code key}. Blocks - * connection until element available or {@code timeout} reached. + * Remove and return the value with its score having the highest score from sorted set at {@code key}.
+ * Blocks connection until element available or {@code timeout} reached. * * @param key must not be {@literal null}. * @param timeout diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java index 6ffb24897..e9f07fee7 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java @@ -25,8 +25,8 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.nio.ByteBuffer; -import java.time.temporal.ChronoUnit; import java.util.List; +import java.util.concurrent.TimeUnit; import org.reactivestreams.Publisher; @@ -390,7 +390,18 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { Assert.notNull(command.getKey(), "Key must not be null!"); Assert.notNull(command.getTimeout(), "Timeout must not be null!"); - long timeout = command.getTimeout().get(ChronoUnit.SECONDS); + if(command.getTimeUnit() == TimeUnit.MILLISECONDS) { + + double timeout = preciseTimeout(command.getTimeout(), command.getTimeUnit()); + + Mono> result = (command.getDirection() == PopDirection.MIN + ? cmd.bzpopmin(timeout, command.getKey()) + : cmd.bzpopmax(timeout, command.getKey())).filter(Value::hasValue).map(Value::getValue); + + return new CommandResponse<>(command, result.filter(Value::hasValue).map(this::toTuple).flux()); + } + + long timeout = command.getTimeUnit().toSeconds(command.getTimeout()); Mono> result = (command.getDirection() == PopDirection.MIN ? cmd.bzpopmin(timeout, command.getKey()) @@ -624,6 +635,10 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { return new DefaultTuple(ByteUtils.getBytes(value), score); } + static double preciseTimeout(long val, TimeUnit unit) { + return (double) unit.toMillis(val) / 1000.0D; + } + protected LettuceReactiveRedisConnection getConnection() { return connection; } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java index 287a81b25..e245ccb20 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java @@ -316,6 +316,13 @@ class LettuceZSetCommands implements RedisZSetCommands { Assert.notNull(key, "Key must not be null!"); Assert.notNull(unit, "TimeUnit must not be null!"); + if(TimeUnit.MILLISECONDS == unit) { + + return connection.invoke(connection.getAsyncDedicatedConnection()) + .from(RedisSortedSetAsyncCommands::bzpopmin, preciseTimeout(timeout, unit), key) + .get(it -> it.map(LettuceConverters::toTuple).getValueOrElse(null)); + } + return connection.invoke(connection.getAsyncDedicatedConnection()) .from(RedisSortedSetAsyncCommands::bzpopmin, unit.toSeconds(timeout), key) .get(it -> it.map(LettuceConverters::toTuple).getValueOrElse(null)); @@ -359,6 +366,13 @@ class LettuceZSetCommands implements RedisZSetCommands { Assert.notNull(key, "Key must not be null!"); Assert.notNull(unit, "TimeUnit must not be null!"); + if(TimeUnit.MILLISECONDS == unit) { + + return connection.invoke(connection.getAsyncDedicatedConnection()) + .from(RedisSortedSetAsyncCommands::bzpopmax, preciseTimeout(timeout, unit), key) + .get(it -> it.map(LettuceConverters::toTuple).getValueOrElse(null)); + } + return connection.invoke(connection.getAsyncDedicatedConnection()) .from(RedisSortedSetAsyncCommands::bzpopmax, unit.toSeconds(timeout), key) .get(it -> it.map(LettuceConverters::toTuple).getValueOrElse(null)); @@ -703,4 +717,7 @@ class LettuceZSetCommands implements RedisZSetCommands { return target; } + static double preciseTimeout(long val, TimeUnit unit) { + return (double) unit.toMillis(val) / 1000.0D; + } } diff --git a/src/main/java/org/springframework/data/redis/core/ZSetOperations.java b/src/main/java/org/springframework/data/redis/core/ZSetOperations.java index 1414ea78c..b862c45f6 100644 --- a/src/main/java/org/springframework/data/redis/core/ZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ZSetOperations.java @@ -365,8 +365,8 @@ public interface ZSetOperations { Set> popMin(K key, long count); /** - * Remove and return the value with its score having the lowest score from sorted set at {@code key}. Blocks - * connection until element available or {@code timeout} reached. + * Remove and return the value with its score having the lowest score from sorted set at {@code key}.
+ * Blocks connection until element available or {@code timeout} reached. * * @param key must not be {@literal null}. * @param timeout @@ -379,8 +379,8 @@ public interface ZSetOperations { TypedTuple popMin(K key, long timeout, TimeUnit unit); /** - * Remove and return the value with its score having the lowest score from sorted set at {@code key}. Blocks - * connection until element available or {@code timeout} reached. + * Remove and return the value with its score having the lowest score from sorted set at {@code key}.
+ * Blocks connection until element available or {@code timeout} reached. * * @param key must not be {@literal null}. * @param timeout must not be {@literal null}. @@ -422,8 +422,8 @@ public interface ZSetOperations { Set> popMax(K key, long count); /** - * Remove and return the value with its score having the highest score from sorted set at {@code key}. Blocks - * connection until element available or {@code timeout} reached. + * Remove and return the value with its score having the highest score from sorted set at {@code key}.
+ * Blocks connection until element available or {@code timeout} reached. * * @param key must not be {@literal null}. * @param timeout @@ -436,8 +436,8 @@ public interface ZSetOperations { TypedTuple popMax(K key, long timeout, TimeUnit unit); /** - * Remove and return the value with its score having the highest score from sorted set at {@code key}. Blocks - * connection until element available or {@code timeout} reached. + * Remove and return the value with its score having the highest score from sorted set at {@code key}.
+ * Blocks connection until element available or {@code timeout} reached. * * @param key must not be {@literal null}. * @param timeout must not be {@literal null}. 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 94abb4a10..e45e3c33e 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -2074,6 +2074,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test void testZRemRangeByScore() { + actual.add(connection.zAdd("myset", 2, "Bob")); actual.add(connection.zAdd("myset", 1, "James")); actual.add(connection.zRemRangeByScore("myset", 0d, 1d)); @@ -2084,6 +2085,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test void testZRevRank() { + actual.add(connection.zAdd("myset", 2, "Bob")); actual.add(connection.zAdd("myset", 1, "James")); actual.add(connection.zAdd("myset", 3, "Joe")); @@ -2093,6 +2095,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test void testZScore() { + actual.add(connection.zAdd("myset", 2, "Bob")); actual.add(connection.zAdd("myset", 1, "James")); actual.add(connection.zAdd("myset", 3, "Joe")); @@ -2103,15 +2106,17 @@ public abstract class AbstractConnectionIntegrationTests { @Test @EnabledOnCommand("ZMSCORE") void testZMScore() { + actual.add(connection.zAdd("myset", 2, "Bob")); actual.add(connection.zAdd("myset", 1, "James")); actual.add(connection.zAdd("myset", 3, "Joe")); - actual.add(connection.zMScore("myset", "James", "Joe")); - verifyResults(Arrays.asList(new Object[] { true, true, true, Arrays.asList(1d, 3d) })); + actual.add(connection.zMScore("myset", "James", "Joe", "Dave")); + verifyResults(Arrays.asList(new Object[] { true, true, true, Arrays.asList(1d, 3d, null) })); } @Test void testZUnionStore() { + actual.add(connection.zAdd("myset", 2, "Bob")); actual.add(connection.zAdd("myset", 1, "James")); actual.add(connection.zAdd("myset", 5, "Joe")); @@ -2125,6 +2130,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test void testZUnionStoreAggWeights() { + actual.add(connection.zAdd("myset", 2, "Bob")); actual.add(connection.zAdd("myset", 1, "James")); actual.add(connection.zAdd("myset", 4, "Joe")); diff --git a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTests.java b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTests.java index e8ca76ce9..9d0e4d9dc 100644 --- a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTests.java +++ b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTests.java @@ -1332,12 +1332,14 @@ public class DefaultStringRedisConnectionPipelineTests extends DefaultStringRedi @Test public void testZMScore() { + doReturn(Collections.singletonList(Arrays.asList(1d, 3d))).when(nativeConnection).closePipeline(); super.testZMScore(); } @Test public void testZUnionStoreAggWeightsBytes() { + doReturn(Collections.singletonList(5L)).when(nativeConnection).closePipeline(); super.testZUnionStoreAggWeightsBytes(); } diff --git a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTxTests.java b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTxTests.java index 0b5a5d651..7e1badabc 100644 --- a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTxTests.java +++ b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTxTests.java @@ -1435,6 +1435,7 @@ public class DefaultStringRedisConnectionPipelineTxTests extends DefaultStringRe @Test public void testZMScore() { + doReturn(Collections.singletonList(Collections.singletonList(Arrays.asList(1d, 3d)))).when(nativeConnection) .closePipeline(); super.testZMScore(); diff --git a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java index b4ddc021e..53942841a 100644 --- a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java @@ -1626,6 +1626,7 @@ public class DefaultStringRedisConnectionTests { @Test public void testZMScore() { + doReturn(Arrays.asList(1d, 3d)).when(nativeConnection).zMScore(fooBytes, barBytes, bar2Bytes); actual.add(connection.zMScore(foo, bar, bar2)); verifyResults(Collections.singletonList(Arrays.asList(1d, 3d))); diff --git a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTxTests.java b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTxTests.java index 5c8e8519b..08c51943d 100644 --- a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTxTests.java +++ b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTxTests.java @@ -1318,6 +1318,7 @@ public class DefaultStringRedisConnectionTxTests extends DefaultStringRedisConne @Test public void testZMScore() { + doReturn(Collections.singletonList(Arrays.asList(1d, 3d))).when(nativeConnection).exec(); super.testZMScore(); } diff --git a/src/test/java/org/springframework/data/redis/support/collections/AbstractRedisZSetTestIntegration.java b/src/test/java/org/springframework/data/redis/support/collections/AbstractRedisZSetTestIntegration.java index 28b429baf..71355615c 100644 --- a/src/test/java/org/springframework/data/redis/support/collections/AbstractRedisZSetTestIntegration.java +++ b/src/test/java/org/springframework/data/redis/support/collections/AbstractRedisZSetTestIntegration.java @@ -121,7 +121,7 @@ public abstract class AbstractRedisZSetTestIntegration extends AbstractRedisC assertThat(zSet.first()).isEqualTo(t1); } - @ParameterizedRedisTest + @ParameterizedRedisTest // GH-2038 @EnabledOnCommand("ZPOPMIN") void testPopFirst() { @@ -137,7 +137,7 @@ public abstract class AbstractRedisZSetTestIntegration extends AbstractRedisC assertThat(zSet).hasSize(2); } - @ParameterizedRedisTest + @ParameterizedRedisTest // GH-2038 @EnabledOnCommand("ZPOPMIN") void testPopFirstWithTimeout() {