diff --git a/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java b/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java index c46783872..4a0154978 100644 --- a/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java +++ b/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java @@ -89,6 +89,7 @@ import org.springframework.util.Assert; * * * @author Mark Paluch + * @author Eddie McDaniel * @param Stream key and Stream field type. * @param Stream value type. * @since 2.2 @@ -397,7 +398,7 @@ public interface StreamReceiver> { */ @SuppressWarnings("unchecked") public StreamReceiverOptionsBuilder> hashValueSerializer( - SerializationPair pair) { + SerializationPair pair) { Assert.notNull(pair, "SerializationPair must not be null"); diff --git a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java index e884dbc09..95f599724 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java @@ -17,12 +17,14 @@ package org.springframework.data.redis.stream; import static org.assertj.core.api.Assertions.*; import static org.junit.Assume.*; +import static org.mockito.ArgumentMatchers.*; import lombok.AllArgsConstructor; import lombok.Data; import reactor.core.publisher.Flux; import reactor.test.StepVerifier; +import java.nio.ByteBuffer; import java.time.Duration; import java.util.Collections; @@ -30,6 +32,7 @@ import org.junit.AfterClass; import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; +import org.mockito.Mockito; import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.RedisVersionUtils; @@ -55,6 +58,7 @@ import org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOption * Integration tests for {@link StreamReceiver}. * * @author Mark Paluch + * @author Eddie McDaniel */ public class StreamReceiverIntegrationTests { @@ -168,6 +172,31 @@ public class StreamReceiverIntegrationTests { .verify(Duration.ofSeconds(5)); } + @Test // DATAREDIS-1172 + public void shouldReceiveCustomHashValueRecords() { + + SerializationPair serializationPair = Mockito.mock(SerializationPair.class); + Mockito.when(serializationPair.read(any(ByteBuffer.class))).thenReturn(345920); + + StreamReceiverOptions> receiverOptions = StreamReceiverOptions.builder() + .hashValueSerializer(serializationPair).build(); + + StreamReceiver> receiver = StreamReceiver.create(connectionFactory, + receiverOptions); + + Flux> messages = receiver.receive(StreamOffset.fromStart("my-stream")); + + messages.as(StepVerifier::create) + .then(() -> reactiveRedisTemplate.opsForStream() + .add("my-stream", Collections.singletonMap("Jesse", "Pinkman")).subscribe()) + .consumeNextWith(it -> { + assertThat(it.getStream()).isEqualTo("my-stream"); + assertThat(it.getValue()).contains(entry("Jesse", 345920)); + }) + .thenCancel() + .verify(Duration.ofSeconds(5)); + } + @Test // DATAREDIS-864 public void latestModeLosesMessages() {