From d6ea867e370ca9d410ca44731993f4b03596b053 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Thu, 11 May 2017 13:08:22 +0200 Subject: [PATCH] DATAREDIS-645 - Introduce absent values in ReactiveStringCommands. We now return AbsentByteBufferResponse to indicate absence of a Redis key when retrieving the key as top-level element. CommandResponse indicates via isPresent its presence or absence of a response value. Omission of absent values can cause a stream with a different response structure compared to the request. Retaining the stream sequence is required for response to request correlation. GET and GETSET commands use presence of CommandResponse to filter response emission. Both commands return a single element so absence does not interferes with the request structure. Previously, we just returned an empty byte-array that caused downstream errors deserializing and emitting null values that are prohibited in Reactive Streams. Original Pull Request: #249 --- .../connection/ReactiveRedisConnection.java | 28 +++++++++++++++++++ .../connection/ReactiveStringCommands.java | 6 ++-- .../LettuceReactiveStringCommands.java | 5 ++-- .../LettuceReactiveStringCommandsTests.java | 20 ++++++------- ...activeValueOperationsIntegrationTests.java | 4 +++ 5 files changed, 49 insertions(+), 14 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java index fb6d656a4..01f87a36b 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java @@ -245,6 +245,14 @@ public interface ReactiveRedisConnection extends Closeable { private final I input; private final O output; + + /** + * @return {@literal true} if the response is present. An absent {@link CommandResponse} maps to Redis + * {@literal (nil)}. + */ + public boolean isPresent() { + return true; + } } /** @@ -267,6 +275,26 @@ public interface ReactiveRedisConnection extends Closeable { } } + /** + * {@link CommandResponse} implementation for {@link ByteBuffer} responses for absent keys. + */ + class AbsentByteBufferResponse extends ByteBufferResponse { + + private final static ByteBuffer EMPTY_BYTE_BUFFER = ByteBuffer.wrap(new byte[0]); + + public AbsentByteBufferResponse(I input) { + super(input, EMPTY_BYTE_BUFFER); + } + + /* (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse#isPresent() + */ + @Override + public boolean isPresent() { + return false; + } + } + /** * {@link CommandResponse} implementation for {@link List} responses. */ diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveStringCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveStringCommands.java index e62ff6ac2..02f3bf753 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStringCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStringCommands.java @@ -199,7 +199,8 @@ public interface ReactiveStringCommands { Assert.notNull(key, "Key must not be null!"); - return get(Mono.just(new KeyCommand(key))).next().map(CommandResponse::getOutput); + return get(Mono.just(new KeyCommand(key))).next().filter(CommandResponse::isPresent) + .map(CommandResponse::getOutput); } /** @@ -225,7 +226,8 @@ public interface ReactiveStringCommands { Assert.notNull(key, "Key must not be null!"); Assert.notNull(value, "Value must not be null!"); - return getSet(Mono.just(SetCommand.set(key).value(value))).next().map(ByteBufferResponse::getOutput); + return getSet(Mono.just(SetCommand.set(key).value(value))).next().filter(CommandResponse::isPresent) + .map(ByteBufferResponse::getOutput); } /** diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java index 382311f6e..07cdeb9ed 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java @@ -24,6 +24,7 @@ import java.util.List; import org.reactivestreams.Publisher; import org.springframework.data.domain.Range; +import org.springframework.data.redis.connection.ReactiveRedisConnection.AbsentByteBufferResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.ByteBufferResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; @@ -113,7 +114,7 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands { } return cmd.getset(command.getKey(), command.getValue()).map((value) -> new ByteBufferResponse<>(command, value)) - .defaultIfEmpty(new ByteBufferResponse<>(command, EMPTY_BYTE_BUFFER)); + .defaultIfEmpty(new AbsentByteBufferResponse<>(command)); })); } @@ -129,7 +130,7 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands { Assert.notNull(command.getKey(), "Key must not be null!"); return cmd.get(command.getKey()).map((value) -> new ByteBufferResponse<>(command, value)) - .defaultIfEmpty(new ByteBufferResponse<>(command, EMPTY_BYTE_BUFFER)); + .defaultIfEmpty(new AbsentByteBufferResponse<>(command)); })); } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsTests.java index ca35ee46d..fe32bed6f 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsTests.java @@ -20,7 +20,11 @@ import static org.hamcrest.core.Is.*; import static org.hamcrest.core.IsEqual.*; import static org.hamcrest.core.IsNull.*; import static org.junit.Assert.*; -import static org.junit.Assume.assumeThat; +import static org.junit.Assume.*; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.TestSubscriber; import java.nio.ByteBuffer; import java.util.Arrays; @@ -37,14 +41,11 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiVa import org.springframework.data.redis.connection.ReactiveStringCommands.SetCommand; import org.springframework.data.redis.connection.RedisStringCommands.BitOperation; import org.springframework.data.redis.core.types.Expiration; - import org.springframework.data.redis.test.util.LettuceRedisClientProvider; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.TestSubscriber; /** * @author Christoph Strobl + * @author Mark Paluch */ public class LettuceReactiveStringCommandsTests extends LettuceReactiveCommandsTestsBase { @@ -59,14 +60,13 @@ public class LettuceReactiveStringCommandsTests extends LettuceReactiveCommandsT assertThat(nativeCommands.get(KEY_1), is(equalTo(VALUE_2))); } - @Test // DATAREDIS-525 + @Test // DATAREDIS-525, DATAREDIS-645 public void getSetShouldReturnPreviousValueCorrectlyWhenNoExists() { Mono result = connection.stringCommands().getSet(KEY_1_BBUFFER, VALUE_2_BBUFFER); ByteBuffer value = result.block(); - assertThat(value, is(notNullValue())); - assertThat(value, is(equalTo(ByteBuffer.allocate(0)))); + assertThat(value, is(nullValue())); assertThat(nativeCommands.get(KEY_1), is(equalTo(VALUE_2))); } @@ -104,11 +104,11 @@ public class LettuceReactiveStringCommandsTests extends LettuceReactiveCommandsT assertThat(result.block(), is(equalTo(VALUE_1_BBUFFER))); } - @Test // DATAREDIS-525 + @Test // DATAREDIS-525, DATAREDIS-645 public void getShouldRetriveNullValueCorrectly() { Mono result = connection.stringCommands().get(KEY_1_BBUFFER); - assertThat(result.block(), is(equalTo(ByteBuffer.allocate(0)))); + assertThat(result.block(), is(nullValue())); } @Test // DATAREDIS-525 diff --git a/src/test/java/org/springframework/data/redis/core/DefaultReactiveValueOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultReactiveValueOperationsIntegrationTests.java index b633dec1a..246d8e3f5 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultReactiveValueOperationsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultReactiveValueOperationsIntegrationTests.java @@ -195,6 +195,8 @@ public class DefaultReactiveValueOperationsIntegrationTests { K key = keyFactory.instance(); V value = valueFactory.instance(); + StepVerifier.create(valueOperations.get(key)).verifyComplete(); + StepVerifier.create(valueOperations.set(key, value)).expectNext(true).verifyComplete(); StepVerifier.create(valueOperations.get(key)).expectNext(value).verifyComplete(); @@ -207,6 +209,8 @@ public class DefaultReactiveValueOperationsIntegrationTests { V value = valueFactory.instance(); V nextValue = valueFactory.instance(); + StepVerifier.create(valueOperations.getAndSet(key, nextValue)).verifyComplete(); + StepVerifier.create(valueOperations.set(key, value)).expectNext(true).verifyComplete(); StepVerifier.create(valueOperations.getAndSet(key, nextValue)).expectNext(value).verifyComplete();