DATAREDIS-1172 - Fix parameter type of hash value serializer in StreamReceiver.

Original pull request: #542.

src/main/java/org/springframework/data/redis/stream/StreamReceiver.java

@@ -89,6 +89,7 @@ import org.springframework.util.Assert;
  * </pre>
  *
  * @author Mark Paluch
+ * @author Eddie McDaniel
  * @param <K> Stream key and Stream field type.
  * @param <V> Stream value type.
  * @since 2.2
@@ -397,7 +398,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 		 */
 		@SuppressWarnings("unchecked")
 		public <HK, HV> StreamReceiverOptionsBuilder<K, MapRecord<K, HK, HV>> hashValueSerializer(
-				SerializationPair<HK> pair) {
+				SerializationPair<HV> pair) {
 
 			Assert.notNull(pair, "SerializationPair must not be null");
 

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<Integer> serializationPair = Mockito.mock(SerializationPair.class);
+		Mockito.when(serializationPair.read(any(ByteBuffer.class))).thenReturn(345920);
+
+		StreamReceiverOptions<String, MapRecord<String, String, Integer>> receiverOptions = StreamReceiverOptions.builder()
+				.<String, Integer>hashValueSerializer(serializationPair).build();
+
+		StreamReceiver<String, MapRecord<String, String, Integer>> receiver = StreamReceiver.create(connectionFactory,
+				receiverOptions);
+
+		Flux<MapRecord<String, String, Integer>> 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() {