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: spring-projects#249

Author: mp911de

src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java

@@ -245,6 +245,14 @@ class CommandResponse<I, O> {
 
 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 ByteBufferResponse(I input, ByteBuffer output) {
 }
 }
 
+/**
+ * {@link CommandResponse} implementation for {@link ByteBuffer} responses for absent keys.
+ */
+class AbsentByteBufferResponse<I> extends ByteBufferResponse<I> {
+
+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.
  */

src/main/java/org/springframework/data/redis/connection/ReactiveStringCommands.java

@@ -199,7 +199,8 @@ default Mono<ByteBuffer> get(ByteBuffer key) {
 
 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 @@ default Mono<ByteBuffer> getSet(ByteBuffer key, ByteBuffer value) {
 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);
 }
 
 /**

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java

@@ -24,6 +24,7 @@
 
 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 @@ public Flux<ByteBufferResponse<SetCommand>> getSet(Publisher<SetCommand> command
 }
 
 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 @@ public Flux<ByteBufferResponse<KeyCommand>> get(Publisher<KeyCommand> commands)
 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));
 }));
 }
 

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsTests.java

@@ -20,7 +20,11 @@
 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.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 void getSetShouldReturnPreviousValueCorrectly() {
 assertThat(nativeCommands.get(KEY_1), is(equalTo(VALUE_2)));
 }
 
-@Test // DATAREDIS-525
+@Test // DATAREDIS-525, DATAREDIS-645
 public void getSetShouldReturnPreviousValueCorrectlyWhenNoExists() {
 
 Mono<ByteBuffer> 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 void getShouldRetriveValueCorrectly() {
 assertThat(result.block(), is(equalTo(VALUE_1_BBUFFER)));
 }
 
-@Test // DATAREDIS-525
+@Test // DATAREDIS-525, DATAREDIS-645
 public void getShouldRetriveNullValueCorrectly() {
 
 Mono<ByteBuffer> result = connection.stringCommands().get(KEY_1_BBUFFER);
-assertThat(result.block(), is(equalTo(ByteBuffer.allocate(0))));
+assertThat(result.block(), is(nullValue()));
 }
 
 @Test // DATAREDIS-525

src/test/java/org/springframework/data/redis/core/DefaultReactiveValueOperationsIntegrationTests.java

@@ -195,6 +195,8 @@ public void get() {
 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 void getAndSet() {
 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();