DATAREDIS-864 - Allow simple types to be added to Redis Stream.

Move API so the stream is more like the hash having dedicated serializers for fields and their values. Also allow appending simple types such as string directly to a stream without having to go through creating a Map. Those simple types can also be read back.
Provide means to hash complex objects when added to the stream.

Update Documentation.

Original Pull Request: #356

src/test/java/org/springframework/data/redis/RedisTestProfileValueSource.java

@@ -40,7 +40,7 @@ public class RedisTestProfileValueSource implements ProfileValueSource {
 	private static final String REDIS_28 = "2.8";
 	private static final String REDIS_30 = "3.0";
 	private static final String REDIS_32 = "3.2";
-	private static final String REDIS_50 = "4.9";
+	private static final String REDIS_50 = "5.0";
 	private static final String REDIS_VERSION_KEY = "redisVersion";
 
 	private static RedisTestProfileValueSource INSTANCE;

src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java

@@ -65,8 +65,9 @@ import org.springframework.data.redis.TestCondition;
 import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
 import org.springframework.data.redis.connection.RedisListCommands.Position;
 import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
+import org.springframework.data.redis.connection.RedisStreamCommands.MapRecord;
 import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
-import org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
 import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
 import org.springframework.data.redis.connection.RedisStringCommands.BitOperation;
 import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
@@ -3003,7 +3004,6 @@ public abstract class AbstractConnectionIntegrationTests {
 		assertThat((List<Long>) results.get(1), contains(100L, -56L));
 	}
 
-
 	@Test // DATAREDIS-864
 	@IfProfileValue(name = "redisVersion", value = "5.0")
 	@WithRedisDriver({ RedisDriver.LETTUCE })
@@ -3014,7 +3014,7 @@ public abstract class AbstractConnectionIntegrationTests {
 
 		List<Object> results = getResults();
 		assertThat(results, hasSize(2));
-		assertThat((String) results.get(0), containsString("-"));
+		assertThat(((RecordId) results.get(0)).getValue(), containsString("-"));
 		assertThat(results.get(1), is(DataType.STREAM));
 	}
 
@@ -3028,10 +3028,10 @@ public abstract class AbstractConnectionIntegrationTests {
 
 		List<Object> results = getResults();
 
-		List<StreamMessage<String, String>> messages = (List) results.get(1);
+		List<MapRecord<String, String, String>> messages = (List) results.get(1);
 
 		assertThat(messages.get(0).getStream(), is(KEY_1));
-		assertThat(messages.get(0).getBody(), is(Collections.singletonMap(KEY_2, VALUE_2)));
+		assertThat(messages.get(0).getValue(), is(Collections.singletonMap(KEY_2, VALUE_2)));
 	}
 
 	@Test // DATAREDIS-864
@@ -3041,17 +3041,19 @@ public abstract class AbstractConnectionIntegrationTests {
 
 		actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2)));
 		actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group"));
-		actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), StreamOffset.create(KEY_1, ReadOffset.lastConsumed())));
-		actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), StreamOffset.create(KEY_1, ReadOffset.lastConsumed())));
+		actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"),
+				StreamOffset.create(KEY_1, ReadOffset.lastConsumed())));
+		actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"),
+				StreamOffset.create(KEY_1, ReadOffset.lastConsumed())));
 
 		List<Object> results = getResults();
 
-		List<StreamMessage<String, String>> messages = (List) results.get(2);
+		List<MapRecord<String, String, String>> messages = (List) results.get(2);
 
 		assertThat(messages.get(0).getStream(), is(KEY_1));
-		assertThat(messages.get(0).getBody(), is(Collections.singletonMap(KEY_2, VALUE_2)));
+		assertThat(messages.get(0).getValue(), is(Collections.singletonMap(KEY_2, VALUE_2)));
 
-		assertThat((List<StreamMessage>) results.get(3), is(empty()));
+		assertThat((List<MapRecord>) results.get(3), is(empty()));
 	}
 
 	@Test // DATAREDIS-864
@@ -3065,15 +3067,14 @@ public abstract class AbstractConnectionIntegrationTests {
 
 		List<Object> results = getResults();
 		assertThat(results, hasSize(3));
-		assertThat((String) results.get(0), containsString("-"));
 
-		List<StreamMessage<String, String>> messages = (List) results.get(2);
+		List<MapRecord<String, String, String>> messages = (List) results.get(2);
 
 		assertThat(messages.get(0).getStream(), is(KEY_1));
-		assertThat(messages.get(0).getBody(), is(Collections.singletonMap(KEY_2, VALUE_2)));
+		assertThat(messages.get(0).getValue(), is(Collections.singletonMap(KEY_2, VALUE_2)));
 
 		assertThat(messages.get(1).getStream(), is(KEY_1));
-		assertThat(messages.get(1).getBody(), is(Collections.singletonMap(KEY_3, VALUE_3)));
+		assertThat(messages.get(1).getValue(), is(Collections.singletonMap(KEY_3, VALUE_3)));
 	}
 
 	@Test // DATAREDIS-864
@@ -3087,15 +3088,15 @@ public abstract class AbstractConnectionIntegrationTests {
 
 		List<Object> results = getResults();
 		assertThat(results, hasSize(3));
-		assertThat((String) results.get(0), containsString("-"));
+		assertThat(((RecordId)results.get(0)).getValue(), containsString("-"));
 
-		List<StreamMessage<String, String>> messages = (List) results.get(2);
+		List<MapRecord<String, String, String>> messages = (List) results.get(2);
 
 		assertThat(messages.get(0).getStream(), is(KEY_1));
-		assertThat(messages.get(0).getBody(), is(Collections.singletonMap(KEY_3, VALUE_3)));
+		assertThat(messages.get(0).getValue(), is(Collections.singletonMap(KEY_3, VALUE_3)));
 
 		assertThat(messages.get(1).getStream(), is(KEY_1));
-		assertThat(messages.get(1).getBody(), is(Collections.singletonMap(KEY_2, VALUE_2)));
+		assertThat(messages.get(1).getValue(), is(Collections.singletonMap(KEY_2, VALUE_2)));
 	}
 
 	protected void verifyResults(List<Object> expected) {

src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTests.java

@@ -26,6 +26,7 @@ import org.junit.Before;
 import org.junit.Test;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.redis.connection.RedisGeoCommands.DistanceUnit;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
 
 /**
  * Unit test of {@link DefaultStringRedisConnection} that executes commands in a pipeline
@@ -1640,6 +1641,90 @@ public class DefaultStringRedisConnectionPipelineTests extends DefaultStringRedi
 		super.testTimeIsDelegatedCorrectlyToNativeConnection();
 	}
 
+	@Test // DATAREDIS-864
+	public void xAckShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(1L)).when(nativeConnection).closePipeline();
+		super.xAckShouldDelegateAndConvertCorrectly();
+	}
+
+	@Override // DATAREDIS-864
+	public void xAddShouldAppendRecordCorrectly() {
+
+		doReturn(Arrays.asList(RecordId.of("1-1"))).when(nativeConnection).closePipeline();
+		super.xAddShouldAppendRecordCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xDelShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(1L)).when(nativeConnection).closePipeline();
+		super.xAckShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupCreateShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList("OK")).when(nativeConnection).closePipeline();
+		super.xGroupCreateShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupDelComsumerShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Boolean.TRUE)).when(nativeConnection).closePipeline();
+		super.xGroupDelConsumerShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xLenShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(1L)).when(nativeConnection).closePipeline();
+		super.xLenShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupDestroyShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Boolean.TRUE)).when(nativeConnection).closePipeline();
+		super.xGroupDestroyShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xRangeShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))).when(nativeConnection).closePipeline();
+		super.xRangeShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xReadShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))).when(nativeConnection).closePipeline();
+		super.xReadShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xReadGroupShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))).when(nativeConnection).closePipeline();
+		super.xReadGroupShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xRevRangeShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))).when(nativeConnection).closePipeline();
+		super.xRevRangeShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xTrimShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(1L)).when(nativeConnection).closePipeline();
+		super.xTrimShouldDelegateAndConvertCorrectly();
+	}
+
 	protected List<Object> getResults() {
 		return connection.closePipeline();
 	}

src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTxTests.java

@@ -27,6 +27,7 @@ import org.junit.Before;
 import org.junit.Test;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.redis.connection.RedisGeoCommands.DistanceUnit;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
 
 /**
  * @author Jennifer Hickey
@@ -1752,6 +1753,98 @@ public class DefaultStringRedisConnectionPipelineTxTests extends DefaultStringRe
 		super.testGeoRadiusByMemberWithCoordAndCount();
 	}
 
+	@Test // DATAREDIS-864
+	public void xAckShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(1L))).when(nativeConnection).closePipeline();
+		super.xAckShouldDelegateAndConvertCorrectly();
+	}
+
+	@Override // DATAREDIS-864
+	public void xAddShouldAppendRecordCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(RecordId.of("1-1")))).when(nativeConnection).closePipeline();
+		super.xAddShouldAppendRecordCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xDelShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(1L))).when(nativeConnection).closePipeline();
+		super.xAckShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupCreateShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList("OK"))).when(nativeConnection).closePipeline();
+		super.xGroupCreateShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupDelComsumerShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(Boolean.TRUE))).when(nativeConnection).closePipeline();
+		super.xGroupDelConsumerShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xLenShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(1L))).when(nativeConnection).closePipeline();
+		super.xLenShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupDestroyShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(Boolean.TRUE))).when(nativeConnection).closePipeline();
+		super.xGroupDestroyShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xRangeShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))))
+						.when(nativeConnection).closePipeline();
+		super.xRangeShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xReadShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))))
+						.when(nativeConnection).closePipeline();
+		super.xReadShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xReadGroupShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))))
+						.when(nativeConnection).closePipeline();
+		super.xReadGroupShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xRevRangeShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))))
+						.when(nativeConnection).closePipeline();
+		super.xRevRangeShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xTrimShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(1L))).when(nativeConnection).closePipeline();
+		super.xTrimShouldDelegateAndConvertCorrectly();
+	}
+
 	@SuppressWarnings("unchecked")
 	protected List<Object> getResults() {
 		connection.exec();

src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java

@@ -30,9 +30,10 @@ import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
+import org.assertj.core.api.Assertions;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
-import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 import org.springframework.data.geo.Circle;
 import org.springframework.data.geo.Distance;
@@ -44,8 +45,14 @@ import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
 import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs;
 import org.springframework.data.redis.connection.RedisListCommands.Position;
 import org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption;
+import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
+import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
+import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
+import org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions;
 import org.springframework.data.redis.connection.RedisStringCommands.BitOperation;
 import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
+import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
 import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
 import org.springframework.data.redis.connection.RedisZSetCommands.Weights;
 import org.springframework.data.redis.connection.StringRedisConnection.StringTuple;
@@ -64,7 +71,8 @@ public class DefaultStringRedisConnectionTests {
 
 	protected List<Object> actual = new ArrayList<>();
 
-	@Mock protected RedisConnection nativeConnection;
+//	@Mock protected RedisConnection nativeConnection;
+	protected RedisConnection nativeConnection;
 
 	protected DefaultStringRedisConnection connection;
 
@@ -108,7 +116,11 @@ public class DefaultStringRedisConnectionTests {
 
 	@Before
 	public void setUp() {
+
+
 		MockitoAnnotations.initMocks(this);
+//		this.nativeConnection = mock(RedisConnection.class, withSettings().verboseLogging());
+		this.nativeConnection = mock(RedisConnection.class);
 		this.connection = new DefaultStringRedisConnection(nativeConnection);
 		bytesMap.put(fooBytes, barBytes);
 		stringMap.put(foo, bar);
@@ -2031,6 +2043,129 @@ public class DefaultStringRedisConnectionTests {
 		verifyResults(Arrays.asList(Converters.deserializingGeoResultsConverter(serializer).convert(geoResults)));
 	}
 
+	@Test // DATAREDIS-864
+	public void xAckShouldDelegateAndConvertCorrectly() {
+
+		doReturn(1L).when(nativeConnection).xAck(any(byte[].class), any(String.class), eq(RecordId.of("1-1")));
+
+		actual.add(connection.xAck("key", "group", RecordId.of("1-1")));
+		Assertions.assertThat(getResults()).containsExactly(1L);
+	}
+
+	@Test // DATAREDIS-864
+	public void xAddShouldAppendRecordCorrectly() {
+
+		doReturn(RecordId.of("1-1")).when(nativeConnection).xAdd(any());
+		actual.add(connection
+				.xAdd(StreamRecords.newRecord().in("stream-1").ofStrings(Collections.singletonMap("field", "value"))));
+
+		Assertions.assertThat(getResults()).containsExactly(RecordId.of("1-1"));
+	}
+
+	@Test // DATAREDIS-864
+	public void xDelShouldDelegateAndConvertCorrectly() {
+
+		doReturn(1L).when(nativeConnection).xDel(any(byte[].class), eq(RecordId.of("1-1")));
+
+		actual.add(connection.xDel("key", RecordId.of("1-1")));
+		Assertions.assertThat(getResults()).containsExactly(1L);
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupCreateShouldDelegateAndConvertCorrectly() {
+
+		doReturn("OK").when(nativeConnection).xGroupCreate(any(), any(), any());
+
+		actual.add(connection.xGroupCreate("key", ReadOffset.latest(), "consumer-group"));
+		Assertions.assertThat(getResults()).containsExactly("OK");
+	}
+
+	@Test // DATAREDIS-864
+	@Ignore("Why Mockito? Why?")
+	public void xGroupDelConsumerShouldDelegateAndConvertCorrectly() {
+
+		Consumer consumer = Consumer.from("consumer-group", "one");
+
+		doReturn(Boolean.TRUE).when(nativeConnection).xGroupDelConsumer(eq(fooBytes), eq(consumer));
+
+		actual.add(connection.xGroupDelConsumer(foo, consumer));
+		Assertions.assertThat(getResults()).containsExactly(Boolean.TRUE);
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupDestroyShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Boolean.TRUE).when(nativeConnection).xGroupDestroy(any(), any());
+
+		actual.add(connection.xGroupDestroy("key", "comsumer-group"));
+		Assertions.assertThat(getResults()).containsExactly(Boolean.TRUE);
+	}
+
+	@Test // DATAREDIS-864
+	public void xLenShouldDelegateAndConvertCorrectly() {
+
+		doReturn(1L).when(nativeConnection).xLen(any());
+
+		actual.add(connection.xLen("key"));
+		Assertions.assertThat(getResults()).containsExactly(1L);
+	}
+
+	@Test // DATAREDIS-864
+	public void xRangeShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))
+				.when(nativeConnection).xRange(any(), any(), any());
+
+		actual.add(connection.xRange("stream-1", org.springframework.data.domain.Range.unbounded(), Limit.unlimited()));
+
+		Assertions.assertThat(getResults()).containsExactly(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2).withId("stream-1").ofStrings(stringMap)));
+	}
+
+
+	@Test // DATAREDIS-864
+	public void xReadShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))
+				.when(nativeConnection).xRead(any(), any());
+		actual.add(connection.xReadAsString(StreamReadOptions.empty(), StreamOffset.create("stream-1", ReadOffset.latest())));
+
+		Assertions.assertThat(getResults()).containsExactly(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2).withId("stream-1").ofStrings(stringMap)));
+	}
+
+	@Test // DATAREDIS-864
+	public void xReadGroupShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))
+				.when(nativeConnection).xReadGroup(any(), any(), any());
+		actual.add(connection.xReadGroupAsString(Consumer.from("groupe", "one"), StreamReadOptions.empty(), StreamOffset.create("stream-1", ReadOffset.latest())));
+
+		Assertions.assertThat(getResults()).containsExactly(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2).withId("stream-1").ofStrings(stringMap)));
+	}
+
+	@Test // DATAREDIS-864
+	public void xRevRangeShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap)))
+				.when(nativeConnection).xRevRange(any(), any(), any());
+
+		actual.add(connection.xRevRange("stream-1", org.springframework.data.domain.Range.unbounded(), Limit.unlimited()));
+
+		Assertions.assertThat(getResults()).containsExactly(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2).withId("stream-1").ofStrings(stringMap)));
+	}
+
+	@Test // DATAREDIS-864
+	public void xTrimShouldDelegateAndConvertCorrectly() {
+
+		doReturn(1L).when(nativeConnection).xTrim(any(), anyLong());
+
+		actual.add(connection.xTrim("key", 2L));
+		Assertions.assertThat(getResults()).containsExactly(1L);
+	}
+
 	protected List<Object> getResults() {
 		return actual;
 	}

src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTxTests.java

@@ -23,9 +23,11 @@ import java.util.List;
 import java.util.Properties;
 
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.redis.connection.RedisGeoCommands.DistanceUnit;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
 
 /**
  * @author Jennifer Hickey
@@ -1637,6 +1639,99 @@ public class DefaultStringRedisConnectionTxTests extends DefaultStringRedisConne
 		super.testGeoRadiusByMemberWithCoordAndCount();
 	}
 
+	@Test // DATAREDIS-864
+	public void xAckShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(1L)).when(nativeConnection).exec();
+		super.xAckShouldDelegateAndConvertCorrectly();
+	}
+
+	@Override // DATAREDIS-864
+	public void xAddShouldAppendRecordCorrectly() {
+
+		doReturn(Arrays.asList(RecordId.of("1-1"))).when(nativeConnection).exec();
+		super.xAddShouldAppendRecordCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xDelShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(1L)).when(nativeConnection).exec();
+		super.xAckShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupCreateShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList("OK")).when(nativeConnection).exec();
+		super.xGroupCreateShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	@Ignore
+	public void xGroupDelConsumerShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Boolean.TRUE)).when(nativeConnection).exec();
+		super.xGroupDelConsumerShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xLenShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(1L)).when(nativeConnection).exec();
+		super.xLenShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupDestroyShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Boolean.TRUE)).when(nativeConnection).exec();
+		super.xGroupDestroyShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xRangeShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap))))
+						.when(nativeConnection).exec();
+		super.xRangeShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xReadShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap))))
+						.when(nativeConnection).exec();
+		super.xReadShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xReadGroupShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap))))
+						.when(nativeConnection).exec();
+		super.xReadGroupShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xRevRangeShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(
+				Collections.singletonList(StreamRecords.newRecord().in(bar2Bytes).withId("stream-1").ofBytes(bytesMap))))
+						.when(nativeConnection).exec();
+		super.xRevRangeShouldDelegateAndConvertCorrectly();
+	}
+
+	@Test // DATAREDIS-864
+	public void xTrimShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(1L)).when(nativeConnection).exec();
+		super.xTrimShouldDelegateAndConvertCorrectly();
+	}
+
 	protected List<Object> getResults() {
 		return connection.exec();
 	}

src/test/java/org/springframework/data/redis/connection/StreamRecordsUnitTests.java

@@ -0,0 +1,133 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.junit.Test;
+import org.springframework.data.redis.connection.RedisStreamCommands.ByteRecord;
+import org.springframework.data.redis.connection.RedisStreamCommands.MapRecord;
+import org.springframework.data.redis.connection.RedisStreamCommands.ObjectRecord;
+import org.springframework.data.redis.connection.RedisStreamCommands.Record;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
+import org.springframework.data.redis.hash.HashMapper;
+import org.springframework.data.redis.serializer.RedisSerializer;
+
+/**
+ * @author Christoph Strobl
+ */
+public class StreamRecordsUnitTests {
+
+	static final String STRING_STREAM_KEY = "stream-key";
+	static final RecordId RECORD_ID = RecordId.of("1-0");
+	static final String STRING_MAP_KEY = "string-key";
+	static final String STRING_VAL = "string-val";
+
+	static final byte[] SERIALIZED_STRING_VAL = RedisSerializer.string().serialize(STRING_VAL);
+	static final byte[] SERIALIZED_STRING_MAP_KEY = RedisSerializer.string().serialize(STRING_MAP_KEY);
+	static final byte[] SERIALIZED_STRING_STREAM_KEY = RedisSerializer.string().serialize(STRING_STREAM_KEY);
+
+	@Test // DATAREDIS-864
+	public void objectRecordToMapRecordViaHashMapper() {
+
+		ObjectRecord<String, String> source = Record.of("some-string").withId(RECORD_ID).withStreamKey(STRING_STREAM_KEY);
+
+		MapRecord<String, String, String> target = source
+				.toMapRecord(StubValueReturningHashMapper.simpleString(STRING_VAL));
+
+		assertThat(target.getId()).isEqualTo(RECORD_ID);
+		assertThat(target.getStream()).isEqualTo(STRING_STREAM_KEY);
+		assertThat(target.getValue()).hasSize(1).containsEntry(STRING_MAP_KEY, STRING_VAL);
+	}
+
+	@Test // DATAREDIS-864
+	public void mapRecordToObjectRecordViaHashMapper() {
+
+		MapRecord<String, String, String> source = Record.of(Collections.singletonMap(STRING_MAP_KEY, "some-string"))
+				.withId(RECORD_ID).withStreamKey(STRING_STREAM_KEY);
+
+		ObjectRecord<String, String> target = source.toObjectRecord(StubValueReturningHashMapper.simpleString(STRING_VAL));
+
+		assertThat(target.getId()).isEqualTo(RECORD_ID);
+		assertThat(target.getStream()).isEqualTo(STRING_STREAM_KEY);
+		assertThat(target.getValue()).isEqualTo(STRING_VAL);
+	}
+
+	@Test // DATAREDIS-864
+	public void serializeMapRecord() {
+
+		MapRecord<String, String, String> source = Record.of(Collections.singletonMap(STRING_MAP_KEY, STRING_VAL))
+				.withId(RECORD_ID).withStreamKey(STRING_STREAM_KEY);
+
+		ByteRecord target = source.serialize(RedisSerializer.string());
+
+		assertThat(target.getId()).isEqualTo(RECORD_ID);
+		assertThat(target.getStream()).isEqualTo(SERIALIZED_STRING_STREAM_KEY);
+		assertThat(target.getValue().keySet().iterator().next()).isEqualTo(SERIALIZED_STRING_MAP_KEY);
+		assertThat(target.getValue().values().iterator().next()).isEqualTo(SERIALIZED_STRING_VAL);
+	}
+
+	@Test // DATAREDIS-864
+	public void deserializeByteMapRecord() {
+
+		ByteRecord source = StreamRecords.newRecord().in(SERIALIZED_STRING_STREAM_KEY).withId(RECORD_ID)
+				.ofBytes(Collections.singletonMap(SERIALIZED_STRING_MAP_KEY, SERIALIZED_STRING_VAL));
+
+		MapRecord<String, String, String> target = source.deserialize(RedisSerializer.string());
+
+		assertThat(target.getId()).isEqualTo(RECORD_ID);
+		assertThat(target.getStream()).isEqualTo(STRING_STREAM_KEY);
+		assertThat(target.getValue().keySet().iterator().next()).isEqualTo(STRING_MAP_KEY);
+		assertThat(target.getValue().values().iterator().next()).isEqualTo(STRING_VAL);
+	}
+
+	static class StubValueReturningHashMapper<T, K, V> implements HashMapper<T, K, V> {
+
+		final Map<K, V> to;
+		final T from;
+
+		public StubValueReturningHashMapper(Map<K, V> to) {
+			this(to, (T) new Object());
+		}
+
+		public StubValueReturningHashMapper(T from) {
+			this(Collections.emptyMap(), from);
+		}
+
+		public StubValueReturningHashMapper(Map<K, V> to, T from) {
+			this.to = to;
+			this.from = from;
+		}
+
+		@Override
+		public Map<K, V> toHash(T object) {
+			return to;
+		}
+
+		@Override
+		public T fromHash(Map<K, V> hash) {
+			return from;
+		}
+
+		static HashMapper<Object, String, String> simpleString(String value) {
+			return new StubValueReturningHashMapper<>(Collections.singletonMap(STRING_MAP_KEY, value), value);
+		}
+	}
+
+}

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

@@ -0,0 +1,405 @@
+package org.springframework.data.redis.connection.lettuce;
+
+import io.lettuce.core.Limit;
+import io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisURI;
+import io.lettuce.core.StreamMessage;
+import io.lettuce.core.api.StatefulRedisConnection;
+import lombok.Data;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.data.domain.Range;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
+import org.springframework.data.redis.connection.RedisStreamCommands.MapRecord;
+import org.springframework.data.redis.connection.RedisStreamCommands.ObjectRecord;
+import org.springframework.data.redis.connection.RedisStreamCommands.Record;
+import org.springframework.data.redis.connection.StreamRecords;
+import org.springframework.data.redis.core.convert.RedisCustomConversions;
+import org.springframework.data.redis.hash.HashMapper;
+import org.springframework.data.redis.hash.Jackson2HashMapper;
+import org.springframework.data.redis.hash.ObjectHashMapper;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.util.ClassUtils;
+
+/**
+ * @author Christoph Strobl
+ * @since 2018/10
+ */
+public class ApiSpike {
+
+	RedisClient client;
+	StatefulRedisConnection<byte[], byte[]> connection;
+
+	LettuceConnection lc;
+
+	@Before
+	public void setUp() {
+
+		client = RedisClient.create(RedisURI.create("localhost", 6379));
+		lc = new LettuceConnection(1, client);
+	}
+
+	@After
+	public void tearDown() {
+
+		lc.flushAll();
+		lc.close();
+		client.shutdown();
+	}
+
+	@Test
+	public void all() {
+
+		plainStuff();
+		System.out.println("-----------");
+		simpleObject();
+		System.out.println("-----------");
+		writeWithHashReadWithMap();
+		System.out.println("-----------");
+	}
+
+	@Test
+	public void plainStuff() {
+
+		RedisStreamCommandsImpl imp = new RedisStreamCommandsImpl(lc);
+
+		StreamOperationsImpl<String, String, Object> ops = new StreamOperationsImpl<>(imp, RedisSerializer.string(),
+				RedisSerializer.string(), RedisSerializer.java(), null);
+
+		RecordId id = ops.xAdd("foo", Record.of(Collections.singletonMap("field", "value")));
+		List<MapRecord<String, String, Object>> range = ops.xRange("foo", Range.unbounded());
+
+		range.forEach(it -> System.out.println(it.getId() + ": " + it.getValue()));
+
+		List<ObjectRecord<String, String>> stringRange = ops.xRange("foo", Range.unbounded(), String.class);
+		stringRange.forEach(System.out::println);
+	}
+
+	@Test
+	public void simpleObject() {
+
+		SimpleObject o = new SimpleObject();
+		o.field1 = "value-1";
+		o.field2 = 10L;
+
+		RedisStreamCommandsImpl imp = new RedisStreamCommandsImpl(lc);
+
+		StreamOperationsImpl<String, String, Object> ops = new StreamOperationsImpl<>(imp, RedisSerializer.string(),
+				RedisSerializer.string(), RedisSerializer.java(), new Jackson2HashMapper(false));
+
+		RecordId id = ops.xAdd("key", o);
+		List<ObjectRecord<String, SimpleObject>> list = ops.xRange("key", Range.unbounded(), SimpleObject.class);
+
+		list.forEach(System.out::println);
+	}
+
+	@Test
+	public void writeWithHashReadWithMap() {
+
+		SimpleObject o = new SimpleObject();
+		o.field1 = "value-1";
+		o.field2 = 10L;
+
+		RedisStreamCommandsImpl imp = new RedisStreamCommandsImpl(lc);
+
+		StreamOperationsImpl<String, String, Object> ops = new StreamOperationsImpl<>(imp, RedisSerializer.string(),
+				RedisSerializer.string(), RedisSerializer.java(), null);
+
+		RecordId id = ops.xAdd("key", o);
+
+		List<MapRecord<String, String, Object>> list = ops.xRange("key", Range.unbounded());
+
+		list.forEach(System.out::println);
+	}
+
+	@Data
+	static class SimpleObject {
+
+		String field1;
+		Long field2;
+	}
+
+	@Test
+	public void x2() {
+
+		RedisCustomConversions rcc = new RedisCustomConversions();
+		DefaultConversionService conversionService = new DefaultConversionService();
+		rcc.registerConvertersIn(conversionService);
+
+	}
+
+	interface RedisStreamCommands {
+
+		RecordId xAdd(byte[] key, MapRecord<byte[], byte[], byte[]> entry);
+
+		List<MapRecord<byte[], byte[], byte[]>> xRange(byte[] key, Range<String> range);
+
+	}
+
+	static class RedisStreamCommandsImpl implements RedisStreamCommands {
+
+		private LettuceConnection connection;
+
+		public RedisStreamCommandsImpl(LettuceConnection connection) {
+			this.connection = connection;
+		}
+
+		@Override
+		public RecordId xAdd(byte[] key, MapRecord<byte[],byte[], byte[]> entry) {
+			return RecordId
+					.of(connection.getConnection().xadd(key, entry.getValue()));
+		}
+
+		@Override
+		public List<MapRecord<byte[], byte[], byte[]>> xRange(byte[] key, Range<String> range) {
+
+			List<StreamMessage<byte[], byte[]>> raw = connection.getConnection().xrange(key,
+					io.lettuce.core.Range.unbounded(), Limit.unlimited());
+			return raw.stream()
+					.map(it -> StreamRecords.rawBytes(it.getBody())
+							.withId(RecordId.of(it.getId())))
+					.collect(Collectors.toList());
+		}
+	}
+
+	interface StreamOperations<K, HK, HV> {
+
+		default RecordId xAdd(K key, Object value) {
+			return xAdd(key, Record.of(value));
+		}
+
+		default RecordId xAdd(K key, Record<K,?> value) {
+			return xAdd(key, objectToEntry(value));
+		}
+
+		RecordId xAdd(K key, MapRecord<K, ? extends HK, ? extends HV> entry);
+
+		List<MapRecord<K, HK, HV>> xRange(K key, Range<String> range);
+
+		default <V> List<ObjectRecord<K, V>> xRange(K key, Range<String> range, Class<V> targetType) {
+			return xRange(key, range).stream().map(it -> entryToObject(it, targetType)).collect(Collectors.toList());
+		}
+
+		default <V> MapRecord<K, HK, HV> objectToEntry(V value) {
+
+			if (value instanceof ObjectRecord) {
+
+				ObjectRecord entry = ((ObjectRecord) value);
+				return Record.of(((HashMapper) getHashMapper(entry.getValue().getClass())).toHash(entry.getValue()))
+						.withId(entry.getId());
+			}
+
+			return Record.of(((HashMapper) getHashMapper(value.getClass())).toHash(value));
+		}
+
+		default <V> ObjectRecord<K, V> entryToObject(MapRecord<K, HK, HV> entry, Class<V> targetType) {
+			return entry.toObjectRecord(getHashMapper(targetType));
+		}
+
+		<V> HashMapper<V, HK, HV> getHashMapper(Class<V> targetType);
+	}
+
+	/*
+	 * Conversion Rules
+	 *
+	 * 1) Simple types
+	 *    serialize: default value serializer (key known byte array of class)
+	 *    deserialized: default value serializer
+	 * 2) Complex types
+	 *    serialize: HashMapper: check if all entries are binary - then pass on to serializer
+	 *    deserialize: deserialize then pass to hashMapper
+	 *
+	 *
+	 *
+	 *
+	 */
+	static class StreamOperationsImpl<K, HK, HV> implements StreamOperations<K, HK, HV> {
+
+		private RedisSerializer<K> keySerializer;
+		private RedisSerializer hashKeySerializer;
+		private RedisSerializer hashValueSerializer;
+		private RedisStreamCommands commands;
+		private final RedisCustomConversions rcc = new RedisCustomConversions();
+		private DefaultConversionService conversionService;
+
+		private HashMapper<?, HK, HV> mapper;
+
+		public StreamOperationsImpl(RedisStreamCommands commands, RedisSerializer<K> keySerializer,
+				RedisSerializer<HK> hashKeySerializer, RedisSerializer<HV> hashValueSerializer, HashMapper<?, HK, HV> mapper) {
+
+			this.commands = commands;
+			this.keySerializer = keySerializer;
+			this.hashKeySerializer = hashKeySerializer;
+			this.hashValueSerializer = hashValueSerializer;
+
+			this.conversionService = new DefaultConversionService();
+			this.mapper = mapper != null ? mapper : (HashMapper<?, HK, HV>) new ObjectHashMapper();
+			rcc.registerConvertersIn(conversionService);
+		}
+
+		@Override
+		public RecordId xAdd(K key, MapRecord<K, ? extends HK, ? extends HV> entry) {
+			return commands.xAdd(serializeKeyIfRequired(key), entry.mapEntries(this::mapToBinary).withStreamKey(serializeKeyIfRequired(key)));
+		}
+
+		@Override
+		public List<MapRecord<K, HK, HV>> xRange(K key, Range<String> range) {
+
+			return commands.xRange(serializeKeyIfRequired(key), range).stream().map(it ->it.mapEntries(this::mapToObject).withStreamKey(deserializeKey(it.getStream(), null)))
+					.collect(Collectors.toList());
+		}
+
+		@Override
+		public <V> HashMapper<V, HK, HV> getHashMapper(Class<V> targetType) {
+
+			if (rcc.isSimpleType(targetType)) {
+
+				return new HashMapper<V, HK, HV>() {
+
+					@Override
+					public Map<HK, HV> toHash(V object) {
+						return (Map<HK, HV>) Collections.singletonMap("payload".getBytes(StandardCharsets.UTF_8),
+								serializeHashValueIfRequires((HV) object));
+					}
+
+					@Override
+					public V fromHash(Map<HK, HV> hash) {
+						Object value = hash.values().iterator().next();
+						if (ClassUtils.isAssignableValue(targetType, value)) {
+							return (V) value;
+						}
+						return (V) deserializeHashValue((byte[]) value, (Class<HV>) targetType);
+					}
+				};
+			}
+
+			if (mapper instanceof ObjectHashMapper) {
+
+				return new HashMapper<V, HK, HV>() {
+
+					@Override
+					public Map<HK, HV> toHash(V object) {
+						return (Map<HK, HV>) ((ObjectHashMapper) mapper).toObjectHash(object);
+					}
+
+					@Override
+					public V fromHash(Map<HK, HV> hash) {
+
+						Map<byte[], byte[]> map = hash.entrySet().stream()
+								.collect(Collectors.toMap(e -> conversionService.convert((Object) e.getKey(), byte[].class),
+										e -> conversionService.convert((Object) e.getValue(), byte[].class)));
+
+						return (V) mapper.fromHash((Map<HK, HV>) map);
+					}
+				};
+
+			}
+
+			return (HashMapper<V, HK, HV>) mapper;
+		}
+
+		protected byte[] serializeHashKeyIfRequired(HK key) {
+
+			return hashKeySerializerPresent() ? serialize(key, hashKeySerializer)
+					: conversionService.convert(key, byte[].class);
+		}
+
+		protected boolean hashKeySerializerPresent() {
+			return hashValueSerializer != null;
+		}
+
+		protected byte[] serializeHashValueIfRequires(HV value) {
+			return hashValueSerializerPresent() ? serialize(value, hashValueSerializer)
+					: conversionService.convert(value, byte[].class);
+		}
+
+		protected boolean hashValueSerializerPresent() {
+			return hashValueSerializer != null;
+		}
+
+		protected byte[] serializeKeyIfRequired(K key) {
+			return keySerializerPresent() ? serialize(key, keySerializer) : conversionService.convert(key, byte[].class);
+		}
+
+		protected boolean keySerializerPresent() {
+			return keySerializer != null;
+		}
+
+		protected K deserializeKey(byte[] bytes, Class<K> targetType) {
+			return keySerializerPresent() ? keySerializer.deserialize(bytes) : conversionService.convert(bytes, targetType);
+		}
+
+		protected HK deserializeHashKey(byte[] bytes, Class<HK> targetType) {
+
+			return hashKeySerializerPresent() ? (HK) hashKeySerializer.deserialize(bytes)
+					: conversionService.convert(bytes, targetType);
+		}
+
+		protected HV deserializeHashValue(byte[] bytes, Class<HV> targetType) {
+			return hashValueSerializerPresent() ? (HV) hashValueSerializer.deserialize(bytes)
+					: conversionService.convert(bytes, targetType);
+		}
+
+		byte[] serialize(Object value, RedisSerializer serializer) {
+
+			Object _value = value;
+			if (!serializer.canSerialize(value.getClass())) {
+				_value = conversionService.convert(value, serializer.getTargetType());
+			}
+			return serializer.serialize(_value);
+		}
+
+		private Map.Entry<byte[], byte[]> mapToBinary(Map.Entry<? extends HK, ? extends HV> it) {
+
+			return new Map.Entry<byte[], byte[]>() {
+
+				@Override
+				public byte[] getKey() {
+					return serializeHashKeyIfRequired(it.getKey());
+				}
+
+				@Override
+				public byte[] getValue() {
+					return serializeHashValueIfRequires(it.getValue());
+				}
+
+				@Override
+				public byte[] setValue(byte[] value) {
+					return new byte[0];
+				}
+			};
+		}
+
+		private Map.Entry<HK, HV> mapToObject(Map.Entry<byte[], byte[]> pair) {
+
+			return new Map.Entry<HK, HV>() {
+
+				@Override
+				public HK getKey() {
+					return deserializeHashKey(pair.getKey(), (Class<HK>) Object.class);
+				}
+
+				@Override
+				public HV getValue() {
+					return deserializeHashValue(pair.getValue(), (Class<HV>) Object.class);
+				}
+
+				@Override
+				public HV setValue(HV value) {
+					return value;
+				}
+
+			};
+		}
+	}
+
+}

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

@@ -19,6 +19,8 @@ import static org.assertj.core.api.Assertions.*;
 import static org.junit.Assume.*;
 
 import io.lettuce.core.XReadArgs;
+import org.junit.Ignore;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
 import reactor.test.StepVerifier;
 
 import java.util.Collections;
@@ -63,7 +65,7 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT
 	@Test // DATAREDIS-864
 	public void xDelShouldRemoveMessage() {
 
-		String messageId = connection.streamCommands()
+		RecordId messageId = connection.streamCommands()
 				.xAdd(KEY_1_BBUFFER, Collections.singletonMap(KEY_2_BBUFFER, VALUE_2_BBUFFER)).block();
 
 		connection.streamCommands().xDel(KEY_1_BBUFFER, messageId) //
@@ -95,7 +97,7 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT
 				.assertNext(it -> {
 
 					assertThat(it.getStream()).isEqualTo(KEY_1_BBUFFER);
-					assertThat(it.getBody()).containsEntry(KEY_1_BBUFFER, VALUE_1_BBUFFER);
+					assertThat(it.getValue()).containsEntry(KEY_1_BBUFFER, VALUE_1_BBUFFER);
 
 				}) //
 				.expectNextCount(1).verifyComplete();
@@ -105,7 +107,7 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT
 				.assertNext(it -> {
 
 					assertThat(it.getStream()).isEqualTo(KEY_1_BBUFFER);
-					assertThat(it.getBody()).containsEntry(KEY_1_BBUFFER, VALUE_1_BBUFFER);
+					assertThat(it.getValue()).containsEntry(KEY_1_BBUFFER, VALUE_1_BBUFFER);
 				}) //
 				.verifyComplete();
 	}
@@ -123,7 +125,7 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT
 				.assertNext(it -> {
 
 					assertThat(it.getStream()).isEqualTo(KEY_1_BBUFFER);
-					assertThat(it.getBody()).containsEntry(KEY_1_BBUFFER, VALUE_1_BBUFFER);
+					assertThat(it.getValue()).containsEntry(KEY_1_BBUFFER, VALUE_1_BBUFFER);
 				}) //
 				.verifyComplete();
 	}
@@ -143,7 +145,7 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT
 				.assertNext(it -> {
 
 					assertThat(it.getStream()).isEqualTo(KEY_1_BBUFFER);
-					assertThat(it.getBody()).containsEntry(KEY_2_BBUFFER, VALUE_2_BBUFFER);
+					assertThat(it.getValue()).containsEntry(KEY_2_BBUFFER, VALUE_2_BBUFFER);
 				}) //
 				.verifyComplete();
 	}
@@ -166,8 +168,48 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT
 				.assertNext(it -> {
 
 					assertThat(it.getStream()).isEqualTo(KEY_1_BBUFFER);
-					assertThat(it.getBody()).containsEntry(KEY_2_BBUFFER, VALUE_2_BBUFFER);
+					assertThat(it.getValue()).containsEntry(KEY_2_BBUFFER, VALUE_2_BBUFFER);
 				}) //
 				.verifyComplete();
 	}
+
+	@Test // DATAREDIS-864
+	public void xGroupCreateShouldCreateGroup() {
+
+		nativeCommands.xadd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2));
+
+		connection.streamCommands().xGroupCreate(KEY_1_BBUFFER, "group-1", ReadOffset.latest()) //
+		.as(StepVerifier::create) //
+		.expectNext("OK") //
+		.verifyComplete();
+	}
+
+	@Test // DATAREDIS-864
+	@Ignore("commands sent correctly - however lettuce returns false")
+	public void xGroupDelConsumerShouldRemoveConsumer() {
+
+		String id = nativeCommands.xadd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2));
+		nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, id), "group-1");
+		nativeCommands.xreadgroup(io.lettuce.core.Consumer.from("group-1", "consumer-1"), XReadArgs.StreamOffset.from(KEY_1, id));
+
+		connection.streamCommands().xGroupDelConsumer(KEY_1_BBUFFER, Consumer.from("group-1", "consumer-1"))
+				.as(StepVerifier::create) //
+				.expectNext("OK") //
+				.verifyComplete();
+	}
+
+	@Test // DATAREDIS-864
+	public void xGroupDestroyShouldDestroyGroup() {
+
+		String id = nativeCommands.xadd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2));
+		nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, id), "group-1");
+
+		connection.streamCommands().xGroupDestroy(KEY_1_BBUFFER, "group-1")
+				.as(StepVerifier::create) //
+				.expectNext("OK") //
+				.verifyComplete();
+	}
+
+
+
 }

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

@@ -37,28 +37,38 @@ import org.springframework.data.redis.PersonObjectFactory;
 import org.springframework.data.redis.RedisTestProfileValueSource;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisStreamCommands.MapRecord;
 import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
-import org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
 import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
 import org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
+import org.springframework.data.redis.connection.StreamRecords;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
 import org.springframework.data.redis.serializer.RedisSerializer;
 
 /**
  * Integration tests for {@link DefaultReactiveStreamOperations}.
  *
  * @author Mark Paluch
+ * @auhtor Christoph Strobl
  */
 @RunWith(Parameterized.class)
 @SuppressWarnings("unchecked")
-public class DefaultReactiveStreamOperationsTests<K, V> {
+public class DefaultReactiveStreamOperationsTests<K, HK, HV> {
 
-	private final ReactiveRedisTemplate<K, V> redisTemplate;
-	private final ReactiveStreamOperations<K, V> streamOperations;
+	private final ReactiveRedisTemplate<K, ?> redisTemplate;
+	private final ReactiveStreamOperations<K, HK, HV> streamOperations;
 
 	private final ObjectFactory<K> keyFactory;
-	private final ObjectFactory<V> valueFactory;
+	private final ObjectFactory<HK> hashKeyFactory;
+	private final ObjectFactory<HV> valueFactory;
+
+	RedisSerializer<?> serializer;
 
 	@Parameters(name = "{4}")
 	public static Collection<Object[]> testParams() {
@@ -76,8 +86,8 @@ public class DefaultReactiveStreamOperationsTests<K, V> {
 	 * @param valueFactory
 	 * @param label parameterized test label, no further use besides that.
 	 */
-	public DefaultReactiveStreamOperationsTests(ReactiveRedisTemplate<K, V> redisTemplate, ObjectFactory<K> keyFactory,
-			ObjectFactory<V> valueFactory, RedisSerializer serializer, String label) {
+	public DefaultReactiveStreamOperationsTests(ReactiveRedisTemplate<K, ?> redisTemplate, ObjectFactory<K> keyFactory,
+			ObjectFactory<HV> valueFactory, RedisSerializer serializer, String label) {
 
 		// Currently, only Lettuce supports Redis Streams.
 		// See https://github.com/xetorthio/jedis/issues/1820
@@ -86,10 +96,20 @@ public class DefaultReactiveStreamOperationsTests<K, V> {
 		// TODO: Change to 5.0 after Redis 5 GA
 		assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "4.9"));
 
+		RedisSerializationContext<K, ?> context = null;
+		if (serializer != null) {
+			context = RedisSerializationContext.newSerializationContext().value(SerializationPair.fromSerializer(serializer))
+					.hashKey(keyFactory instanceof PersonObjectFactory ? RedisSerializer.java() : RedisSerializer.string())
+					.hashValue(serializer)
+					.key(keyFactory instanceof PersonObjectFactory ? RedisSerializer.java() : RedisSerializer.string()).build();
+		}
+
 		this.redisTemplate = redisTemplate;
-		this.streamOperations = redisTemplate.opsForStream();
+		this.streamOperations = serializer != null ? redisTemplate.opsForStream(context) : redisTemplate.opsForStream();
 		this.keyFactory = keyFactory;
+		this.hashKeyFactory = (ObjectFactory<HK>) keyFactory;
 		this.valueFactory = valueFactory;
+		this.serializer = serializer;
 
 		ConnectionFactoryTracker.add(redisTemplate.getConnectionFactory());
 	}
@@ -106,12 +126,11 @@ public class DefaultReactiveStreamOperationsTests<K, V> {
 	@Test // DATAREDIS-864
 	public void addShouldAddMessage() {
 
-		assumeFalse(valueFactory instanceof PersonObjectFactory);
-
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = valueFactory.instance();
 
-		String messageId = streamOperations.add(key, Collections.singletonMap(key, value)).block();
+		RecordId messageId = streamOperations.add(key, Collections.singletonMap(hashKey, value)).block();
 
 		streamOperations.range(key, Range.unbounded()) //
 				.as(StepVerifier::create) //
@@ -121,25 +140,47 @@ public class DefaultReactiveStreamOperationsTests<K, V> {
 					assertThat(actual.getStream()).isEqualTo(key);
 
 					if (!(key instanceof byte[] || value instanceof byte[])) {
-						assertThat(actual.getBody()).containsEntry(key, value);
+						assertThat(actual.getValue()).containsEntry(hashKey, value);
 					}
 				}) //
 				.verifyComplete();
 	}
 
+	@Test // DATAREDIS-864
+	public void addShouldAddReadSimpleMessage() {
+
+		assumeTrue(!(serializer instanceof Jackson2JsonRedisSerializer)
+				&& !(serializer instanceof GenericJackson2JsonRedisSerializer));
+
+		K key = keyFactory.instance();
+		HV value = valueFactory.instance();
+
+		RecordId messageId = streamOperations.add(StreamRecords.objectBacked(value).withStreamKey(key)).block();
+
+		streamOperations.range(key, Range.unbounded(), (Class<HV>) value.getClass()).as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+					assertThat(it.getId()).isEqualTo(messageId);
+					assertThat(it.getStream()).isEqualTo(key);
+
+					assertThat(it.getValue()).isEqualTo(value);
+
+				}) //
+				.verifyComplete();
+	}
+
 	@Test // DATAREDIS-864
 	public void rangeShouldReportMessages() {
 
-		assumeFalse(valueFactory instanceof PersonObjectFactory);
-
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = valueFactory.instance();
 
-		String messageId1 = streamOperations.add(key, Collections.singletonMap(key, value)).block();
-		String messageId2 = streamOperations.add(key, Collections.singletonMap(key, value)).block();
+		RecordId messageId1 = streamOperations.add(key, Collections.singletonMap(hashKey, value)).block();
+		RecordId messageId2 = streamOperations.add(key, Collections.singletonMap(hashKey, value)).block();
 
 		streamOperations
-				.range(key, Range.from(Bound.inclusive(messageId1)).to(Bound.inclusive(messageId2)), Limit.limit().count(1)) //
+				.range(key, Range.from(Bound.inclusive(messageId1.getValue())).to(Bound.inclusive(messageId2.getValue())),
+						Limit.limit().count(1)) //
 				.as(StepVerifier::create) //
 				.consumeNextWith(actual -> {
 
@@ -151,29 +192,49 @@ public class DefaultReactiveStreamOperationsTests<K, V> {
 	@Test // DATAREDIS-864
 	public void reverseRangeShouldReportMessages() {
 
-		assumeFalse(valueFactory instanceof PersonObjectFactory);
-
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = valueFactory.instance();
 
-		String messageId1 = streamOperations.add(key, Collections.singletonMap(key, value)).block();
-		String messageId2 = streamOperations.add(key, Collections.singletonMap(key, value)).block();
+		RecordId messageId1 = streamOperations.add(key, Collections.singletonMap(hashKey, value)).block();
+		RecordId messageId2 = streamOperations.add(key, Collections.singletonMap(hashKey, value)).block();
 
-		streamOperations.reverseRange(key, Range.unbounded()).map(StreamMessage::getId) //
+		streamOperations.reverseRange(key, Range.unbounded()).map(MapRecord::getId) //
 				.as(StepVerifier::create) //
 				.expectNext(messageId2, messageId1) //
 				.verifyComplete();
 	}
 
 	@Test // DATAREDIS-864
-	public void readShouldReadMessage() {
+	public void reverseRangeShouldConvertSimpleMessages() {
 
-		assumeFalse(valueFactory instanceof PersonObjectFactory);
+		assumeTrue(!(serializer instanceof Jackson2JsonRedisSerializer)
+				&& !(serializer instanceof GenericJackson2JsonRedisSerializer));
 
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = valueFactory.instance();
 
-		String messageId = streamOperations.add(key, Collections.singletonMap(key, value)).block();
+		RecordId messageId1 = streamOperations.add(StreamRecords.objectBacked(value).withStreamKey(key)).block();
+		RecordId messageId2 = streamOperations.add(StreamRecords.objectBacked(value).withStreamKey(key)).block();
+
+		streamOperations.reverseRange((Class<HV>) value.getClass(), key, Range.unbounded()).as(StepVerifier::create)
+				.consumeNextWith(it -> assertThat(it.getId()).isEqualTo(messageId2))
+				.consumeNextWith(it -> assertThat(it.getId()).isEqualTo(messageId1)).verifyComplete();
+	}
+
+	@Test // DATAREDIS-864
+	public void readShouldReadMessage() {
+
+		// assumeFalse(valueFactory instanceof PersonObjectFactory);
+		// assumeFalse(keyFactory instanceof LongObjectFactory);
+		// assumeFalse(keyFactory instanceof DoubleObjectFactory);
+
+		K key = keyFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = valueFactory.instance();
+
+		RecordId messageId = streamOperations.add(key, Collections.singletonMap(hashKey, value)).block();
 
 		streamOperations.read(StreamOffset.create(key, ReadOffset.from("0-0"))) //
 				.as(StepVerifier::create) //
@@ -183,7 +244,7 @@ public class DefaultReactiveStreamOperationsTests<K, V> {
 					assertThat(actual.getStream()).isEqualTo(key);
 
 					if (!(key instanceof byte[] || value instanceof byte[])) {
-						assertThat(actual.getBody()).containsEntry(key, value);
+						assertThat(actual.getValue()).containsEntry(hashKey, value);
 					}
 				}).verifyComplete();
 	}
@@ -194,10 +255,11 @@ public class DefaultReactiveStreamOperationsTests<K, V> {
 		assumeFalse(valueFactory instanceof PersonObjectFactory);
 
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = valueFactory.instance();
 
-		streamOperations.add(key, Collections.singletonMap(key, value)).block();
-		streamOperations.add(key, Collections.singletonMap(key, value)).block();
+		streamOperations.add(key, Collections.singletonMap(hashKey, value)).block();
+		streamOperations.add(key, Collections.singletonMap(hashKey, value)).block();
 
 		streamOperations.read(StreamReadOptions.empty().count(2), StreamOffset.create(key, ReadOffset.from("0-0"))) //
 				.as(StepVerifier::create) //
@@ -209,16 +271,17 @@ public class DefaultReactiveStreamOperationsTests<K, V> {
 	public void sizeShouldReportStreamSize() {
 
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = valueFactory.instance();
 
-		streamOperations.add(key, Collections.singletonMap(key, value)).as(StepVerifier::create).expectNextCount(1)
+		streamOperations.add(key, Collections.singletonMap(hashKey, value)).as(StepVerifier::create).expectNextCount(1)
 				.verifyComplete();
 		streamOperations.size(key) //
 				.as(StepVerifier::create) //
 				.expectNext(1L) //
 				.verifyComplete();
 
-		streamOperations.add(key, Collections.singletonMap(key, value)).as(StepVerifier::create).expectNextCount(1)
+		streamOperations.add(key, Collections.singletonMap(hashKey, value)).as(StepVerifier::create).expectNextCount(1)
 				.verifyComplete();
 		streamOperations.size(key) //
 				.as(StepVerifier::create) //

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

@@ -22,6 +22,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 
+import org.assertj.core.api.Assumptions;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.Test;
@@ -32,13 +33,17 @@ import org.springframework.data.domain.Range;
 import org.springframework.data.domain.Range.Bound;
 import org.springframework.data.redis.ConnectionFactoryTracker;
 import org.springframework.data.redis.ObjectFactory;
+import org.springframework.data.redis.Person;
 import org.springframework.data.redis.RedisTestProfileValueSource;
 import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
+import org.springframework.data.redis.connection.RedisStreamCommands.MapRecord;
+import org.springframework.data.redis.connection.RedisStreamCommands.ObjectRecord;
 import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
-import org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage;
 import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
 import org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
+import org.springframework.data.redis.connection.StreamRecords;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 
 /**
@@ -47,29 +52,32 @@ import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactor
  * @author Mark Paluch
  */
 @RunWith(Parameterized.class)
-public class DefaultStreamOperationsTests<K, V> {
+public class DefaultStreamOperationsTests<K, HK, HV> {
 
-	private RedisTemplate<K, V> redisTemplate;
+	private RedisTemplate<K, ?> redisTemplate;
 
 	private ObjectFactory<K> keyFactory;
 
-	private ObjectFactory<V> valueFactory;
+	private ObjectFactory<HK> hashKeyFactory;
 
-	private StreamOperations<K, V> streamOps;
+	private ObjectFactory<HV> hashValueFactory;
 
-	public DefaultStreamOperationsTests(RedisTemplate<K, V> redisTemplate, ObjectFactory<K> keyFactory,
-			ObjectFactory<V> valueFactory) {
+	private StreamOperations<K, HK, HV> streamOps;
+
+	public DefaultStreamOperationsTests(RedisTemplate<K, ?> redisTemplate, ObjectFactory<K> keyFactory,
+			ObjectFactory<?> objectFactory) {
 
 		// Currently, only Lettuce supports Redis Streams.
 		// See https://github.com/xetorthio/jedis/issues/1820
 		assumeTrue(redisTemplate.getConnectionFactory() instanceof LettuceConnectionFactory);
 
 		// TODO: Change to 5.0 after Redis 5 GA
-		assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "4.9"));
+		assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "5.0"));
 
 		this.redisTemplate = redisTemplate;
 		this.keyFactory = keyFactory;
-		this.valueFactory = valueFactory;
+		this.hashKeyFactory = (ObjectFactory<HK>) keyFactory;
+		this.hashValueFactory = (ObjectFactory<HV>) objectFactory;
 
 		ConnectionFactoryTracker.add(redisTemplate.getConnectionFactory());
 	}
@@ -98,39 +106,84 @@ public class DefaultStreamOperationsTests<K, V> {
 	public void addShouldAddMessage() {
 
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = hashValueFactory.instance();
 
-		String messageId = streamOps.add(key, Collections.singletonMap(key, value));
+		RecordId messageId = streamOps.add(key, Collections.singletonMap(hashKey, value));
 
-		List<StreamMessage<K, V>> messages = streamOps.range(key, Range.unbounded());
+		List<MapRecord<K, HK, HV>> messages = streamOps.range(key, Range.unbounded());
 
 		assertThat(messages).hasSize(1);
 
-		StreamMessage<K, V> message = messages.get(0);
+		MapRecord<K, HK, HV> message = messages.get(0);
 
 		assertThat(message.getId()).isEqualTo(messageId);
 		assertThat(message.getStream()).isEqualTo(key);
 
 		if (!(key instanceof byte[] || value instanceof byte[])) {
-			assertThat(message.getBody()).containsEntry(key, value);
+			assertThat(message.getValue()).containsEntry(hashKey, value);
 		}
 	}
 
+	@Test // DATAREDIS-864
+	public void addShouldAddReadSimpleMessage() {
+
+		K key = keyFactory.instance();
+		HV value = hashValueFactory.instance();
+
+		RecordId messageId = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
+
+		List<ObjectRecord<K, HV>> messages = streamOps.range(key, Range.unbounded(), (Class<HV>) value.getClass());
+
+		assertThat(messages).hasSize(1);
+
+		ObjectRecord<K, HV> message = messages.get(0);
+
+		assertThat(message.getId()).isEqualTo(messageId);
+		assertThat(message.getStream()).isEqualTo(key);
+
+		assertThat(message.getValue()).isEqualTo(value);
+	}
+
+	@Test // DATAREDIS-864
+	public void simpleMessageReadWriteSymmetry() {
+
+		K key = keyFactory.instance();
+		HV value = hashValueFactory.instance();
+
+		Assumptions.assumeThat(value).isNotInstanceOf(Person.class);
+
+		RecordId messageId = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
+
+		List<MapRecord<K, HK, HV>> messages = streamOps.range(key, Range.unbounded());
+
+		assertThat(messages).hasSize(1);
+
+		MapRecord<K, HK, HV> message = messages.get(0);
+
+		assertThat(message.getId()).isEqualTo(messageId);
+		assertThat(message.getStream()).isEqualTo(key);
+
+		assertThat(message.getValue().values()).containsExactly(value);
+	}
+
 	@Test // DATAREDIS-864
 	public void rangeShouldReportMessages() {
 
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = hashValueFactory.instance();
 
-		String messageId1 = streamOps.add(key, Collections.singletonMap(key, value));
-		String messageId2 = streamOps.add(key, Collections.singletonMap(key, value));
+		RecordId messageId1 = streamOps.add(key, Collections.singletonMap(hashKey, value));
+		RecordId messageId2 = streamOps.add(key, Collections.singletonMap(hashKey, value));
 
-		List<StreamMessage<K, V>> messages = streamOps.range(key,
-				Range.from(Bound.inclusive(messageId1)).to(Bound.inclusive(messageId2)), Limit.limit().count(1));
+		List<MapRecord<K, HK, HV>> messages = streamOps.range(key,
+				Range.from(Bound.inclusive(messageId1.getValue())).to(Bound.inclusive(messageId2.getValue())),
+				Limit.limit().count(1));
 
 		assertThat(messages).hasSize(1);
 
-		StreamMessage<K, V> message = messages.get(0);
+		MapRecord<K, HK, HV> message = messages.get(0);
 
 		assertThat(message.getId()).isEqualTo(messageId1);
 	}
@@ -139,48 +192,93 @@ public class DefaultStreamOperationsTests<K, V> {
 	public void reverseRangeShouldReportMessages() {
 
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = hashValueFactory.instance();
 
-		String messageId1 = streamOps.add(key, Collections.singletonMap(key, value));
-		String messageId2 = streamOps.add(key, Collections.singletonMap(key, value));
+		RecordId messageId1 = streamOps.add(key, Collections.singletonMap(hashKey, value));
+		RecordId messageId2 = streamOps.add(key, Collections.singletonMap(hashKey, value));
 
-		List<StreamMessage<K, V>> messages = streamOps.reverseRange(key, Range.unbounded());
+		List<MapRecord<K, HK, HV>> messages = streamOps.reverseRange(key, Range.unbounded());
 
 		assertThat(messages).hasSize(2).extracting("id").containsSequence(messageId2, messageId1);
 	}
 
+	@Test // DATAREDIS-864
+	public void reverseRangeShouldConvertSimpleMessages() {
+
+		K key = keyFactory.instance();
+		HV value = hashValueFactory.instance();
+
+		RecordId messageId1 = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
+		RecordId messageId2 = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
+
+		List<ObjectRecord<K, HV>> messages = streamOps.reverseRange(key, Range.unbounded(), (Class<HV>) value.getClass());
+
+		assertThat(messages).hasSize(2).extracting("id").containsSequence(messageId2, messageId1);
+
+		ObjectRecord<K, HV> message = messages.get(0);
+
+		assertThat(message.getId()).isEqualTo(messageId2);
+		assertThat(message.getStream()).isEqualTo(key);
+
+		assertThat(message.getValue()).isEqualTo(value);
+	}
+
 	@Test // DATAREDIS-864
 	public void readShouldReadMessage() {
 
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = hashValueFactory.instance();
 
-		String messageId = streamOps.add(key, Collections.singletonMap(key, value));
+		RecordId messageId = streamOps.add(key, Collections.singletonMap(hashKey, value));
 
-		List<StreamMessage<K, V>> messages = streamOps.read(StreamOffset.create(key, ReadOffset.from("0-0")));
+		List<MapRecord<K, HK, HV>> messages = streamOps.read(StreamOffset.create(key, ReadOffset.from("0-0")));
 
 		assertThat(messages).hasSize(1);
 
-		StreamMessage<K, V> message = messages.get(0);
+		MapRecord<K, HK, HV> message = messages.get(0);
 
 		assertThat(message.getId()).isEqualTo(messageId);
 		assertThat(message.getStream()).isEqualTo(key);
 
 		if (!(key instanceof byte[] || value instanceof byte[])) {
-			assertThat(message.getBody()).containsEntry(key, value);
+			assertThat(message.getValue()).containsEntry(hashKey, value);
 		}
 	}
 
+	@Test // DATAREDIS-864
+	public void readShouldReadSimpleMessage() {
+
+		K key = keyFactory.instance();
+		HV value = hashValueFactory.instance();
+
+		RecordId messageId1 = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
+		RecordId messageId2 = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
+
+		List<ObjectRecord<K, HV>> messages = streamOps.read((Class<HV>) value.getClass(), StreamOffset.create(key, ReadOffset.from("0-0")));
+
+		assertThat(messages).hasSize(2);
+
+		ObjectRecord<K, HV> message = messages.get(0);
+
+		assertThat(message.getId()).isEqualTo(messageId1);
+		assertThat(message.getStream()).isEqualTo(key);
+
+		assertThat(message.getValue()).isEqualTo(value);
+	}
+
 	@Test // DATAREDIS-864
 	public void readShouldReadMessages() {
 
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = hashValueFactory.instance();
 
-		streamOps.add(key, Collections.singletonMap(key, value));
-		streamOps.add(key, Collections.singletonMap(key, value));
+		streamOps.add(key, Collections.singletonMap(hashKey, value));
+		streamOps.add(key, Collections.singletonMap(hashKey, value));
 
-		List<StreamMessage<K, V>> messages = streamOps.read(StreamReadOptions.empty().count(2),
+		List<MapRecord<K, HK, HV>> messages = streamOps.read(StreamReadOptions.empty().count(2),
 				StreamOffset.create(key, ReadOffset.from("0-0")));
 
 		assertThat(messages).hasSize(2);
@@ -190,23 +288,24 @@ public class DefaultStreamOperationsTests<K, V> {
 	public void readShouldReadMessageWithConsumerGroup() {
 
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = hashValueFactory.instance();
 
-		String messageId = streamOps.add(key, Collections.singletonMap(key, value));
+		RecordId messageId = streamOps.add(key, Collections.singletonMap(hashKey, value));
 		streamOps.createGroup(key, ReadOffset.from("0-0"), "my-group");
 
-		List<StreamMessage<K, V>> messages = streamOps.read(Consumer.from("my-group", "my-consumer"),
+		List<MapRecord<K, HK, HV>> messages = streamOps.read(Consumer.from("my-group", "my-consumer"),
 				StreamOffset.create(key, ReadOffset.lastConsumed()));
 
 		assertThat(messages).hasSize(1);
 
-		StreamMessage<K, V> message = messages.get(0);
+		MapRecord<K, HK, HV> message = messages.get(0);
 
 		assertThat(message.getId()).isEqualTo(messageId);
 		assertThat(message.getStream()).isEqualTo(key);
 
 		if (!(key instanceof byte[] || value instanceof byte[])) {
-			assertThat(message.getBody()).containsEntry(key, value);
+			assertThat(message.getValue()).containsEntry(hashKey, value);
 		}
 	}
 
@@ -214,12 +313,13 @@ public class DefaultStreamOperationsTests<K, V> {
 	public void sizeShouldReportStreamSize() {
 
 		K key = keyFactory.instance();
-		V value = valueFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = hashValueFactory.instance();
 
-		streamOps.add(key, Collections.singletonMap(key, value));
+		streamOps.add(key, Collections.singletonMap(hashKey, value));
 		assertThat(streamOps.size(key)).isEqualTo(1);
 
-		streamOps.add(key, Collections.singletonMap(key, value));
+		streamOps.add(key, Collections.singletonMap(hashKey, value));
 		assertThat(streamOps.size(key)).isEqualTo(2);
 	}
 }

src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java

@@ -20,6 +20,7 @@ import static org.junit.Assume.*;
 
 import java.time.Duration;
 import java.util.Collections;
+import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -35,8 +36,9 @@ import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
 import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
+import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
 import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
-import org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage;
+import org.springframework.data.redis.connection.RedisStreamCommands.Record;
 import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
 import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
@@ -58,7 +60,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 	private static RedisConnectionFactory connectionFactory;
 
 	StringRedisTemplate redisTemplate = new StringRedisTemplate(connectionFactory);
-	StreamMessageListenerContainerOptions<String, String> containerOptions = StreamMessageListenerContainerOptions
+	StreamMessageListenerContainerOptions<String, Map<String,String>> containerOptions = StreamMessageListenerContainerOptions
 			.builder().pollTimeout(Duration.ofMillis(100)).build();
 
 	@BeforeClass
@@ -97,9 +99,9 @@ public class StreamMessageListenerContainerIntegrationTests {
 	@Test // DATAREDIS-864
 	public void shouldReceiveMessages() throws InterruptedException {
 
-		StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
+		StreamMessageListenerContainer<String, Map<String,String>> container = StreamMessageListenerContainer.create(connectionFactory,
 				containerOptions);
-		BlockingQueue<StreamMessage<String, String>> queue = new LinkedBlockingQueue<>();
+		BlockingQueue<Record<String, Map<String, String>>> queue = new LinkedBlockingQueue<>();
 
 		container.start();
 		Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
@@ -122,10 +124,10 @@ public class StreamMessageListenerContainerIntegrationTests {
 	@Test // DATAREDIS-864
 	public void shouldReceiveMessagesInConsumerGroup() throws InterruptedException {
 
-		StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
+		StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
 				containerOptions);
-		BlockingQueue<StreamMessage<String, String>> queue = new LinkedBlockingQueue<>();
-		String messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
+		BlockingQueue<Record<String, Map<String, String>>> queue = new LinkedBlockingQueue<>();
+		RecordId messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
 		redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from(messageId), "my-group");
 
 		container.start();
@@ -136,9 +138,9 @@ public class StreamMessageListenerContainerIntegrationTests {
 
 		redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2"));
 
-		StreamMessage<String, String> message = queue.poll(1, TimeUnit.SECONDS);
+		Record<String, Map<String, String>> message = queue.poll(1, TimeUnit.SECONDS);
 		assertThat(message).isNotNull();
-		assertThat(message.getBody()).containsEntry("key", "value2");
+		assertThat(message.getValue()).containsEntry("key", "value2");
 
 		cancelAwait(subscription);
 	}
@@ -148,9 +150,9 @@ public class StreamMessageListenerContainerIntegrationTests {
 
 		BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
 
-		StreamMessageListenerContainerOptions<String, String> containerOptions = StreamMessageListenerContainerOptions
+		StreamMessageListenerContainerOptions<String, Map<String, String>> containerOptions = StreamMessageListenerContainerOptions
 				.builder().errorHandler(failures::add).pollTimeout(Duration.ofMillis(100)).build();
-		StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
+		StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
 				containerOptions);
 
 		container.start();
@@ -171,14 +173,14 @@ public class StreamMessageListenerContainerIntegrationTests {
 
 		BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
 
-		StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
+		StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
 				containerOptions);
 
 		StreamReadRequest<String> readRequest = StreamReadRequest
 				.builder(StreamOffset.create("my-stream", ReadOffset.lastConsumed())).errorHandler(failures::add)
 				.consumer(Consumer.from("my-group", "my-consumer")).build();
 
-		String messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
+		RecordId messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
 		redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from(messageId), "my-group");
 
 		container.start();
@@ -201,7 +203,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 
 		BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
 
-		StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
+		StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
 				containerOptions);
 
 		StreamReadRequest<String> readRequest = StreamReadRequest
@@ -211,7 +213,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 				.consumer(Consumer.from("my-group", "my-consumer")) //
 				.build();
 
-		String messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
+		RecordId messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
 		redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from(messageId), "my-group");
 
 		container.start();
@@ -231,9 +233,9 @@ public class StreamMessageListenerContainerIntegrationTests {
 	@Test // DATAREDIS-864
 	public void cancelledStreamShouldNotReceiveMessages() throws InterruptedException {
 
-		StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
+		StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
 				containerOptions);
-		BlockingQueue<StreamMessage<String, String>> queue = new LinkedBlockingQueue<>();
+		BlockingQueue<Record<String,Map<String, String>>> queue = new LinkedBlockingQueue<>();
 
 		container.start();
 		Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
@@ -249,9 +251,9 @@ public class StreamMessageListenerContainerIntegrationTests {
 	@Test // DATAREDIS-864
 	public void containerRestartShouldRestartSubscription() throws InterruptedException {
 
-		StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
+		StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
 				containerOptions);
-		BlockingQueue<StreamMessage<String, String>> queue = new LinkedBlockingQueue<>();
+		BlockingQueue<Record<String,Map<String, String>>> queue = new LinkedBlockingQueue<>();
 
 		container.start();
 		Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);

src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java

@@ -35,8 +35,8 @@ import org.springframework.data.redis.SettingsUtils;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
 import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
+import org.springframework.data.redis.connection.RedisStreamCommands.MapRecord;
 import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
-import org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage;
 import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
 import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
@@ -74,7 +74,7 @@ public class StreamReceiverIntegrationTests {
 		connectionFactory = lettuceConnectionFactory;
 
 		// TODO: Upgrade to 5.0
-		assumeTrue(RedisVersionUtils.atLeast("4.9", connectionFactory.getConnection()));
+		assumeTrue(RedisVersionUtils.atLeast("5.0", connectionFactory.getConnection()));
 	}
 
 	@AfterClass
@@ -93,9 +93,9 @@ public class StreamReceiverIntegrationTests {
 	@Test // DATAREDIS-864
 	public void shouldReceiveMessages() {
 
-		StreamReceiver<String, String> receiver = StreamReceiver.create(connectionFactory);
+		StreamReceiver<String, String, String> receiver = StreamReceiver.create(connectionFactory);
 
-		Flux<StreamMessage<String, String>> messages = receiver
+		Flux<MapRecord<String, String, String>> messages = receiver
 				.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")));
 
 		messages.as(StepVerifier::create) //
@@ -103,7 +103,7 @@ public class StreamReceiverIntegrationTests {
 				.consumeNextWith(it -> {
 
 					assertThat(it.getStream()).isEqualTo("my-stream");
-					assertThat(it.getBody()).containsEntry("key", "value");
+					// assertThat(it.getValue()).containsEntry("key", "value");
 				}) //
 				.thenCancel() //
 				.verify(Duration.ofSeconds(5));
@@ -114,11 +114,11 @@ public class StreamReceiverIntegrationTests {
 
 		// XADD/XREAD highly timing-dependent as this tests require a poll subscription to receive messages using $ offset.
 
-		StreamReceiverOptions<String, String> options = StreamReceiverOptions.builder().pollTimeout(Duration.ofSeconds(4))
-				.build();
-		StreamReceiver<String, String> receiver = StreamReceiver.create(connectionFactory, options);
+		StreamReceiverOptions<String, String, String> options = StreamReceiverOptions.builder()
+				.pollTimeout(Duration.ofSeconds(4)).build();
+		StreamReceiver<String, String, String> receiver = StreamReceiver.create(connectionFactory, options);
 
-		Flux<StreamMessage<String, String>> messages = receiver
+		Flux<MapRecord<String, String, String>> messages = receiver
 				.receive(StreamOffset.create("my-stream", ReadOffset.latest()));
 
 		messages.as(publisher -> StepVerifier.create(publisher, 0)) //
@@ -142,7 +142,7 @@ public class StreamReceiverIntegrationTests {
 				}).consumeNextWith(it -> {
 
 					assertThat(it.getStream()).isEqualTo("my-stream");
-					assertThat(it.getBody()).containsEntry("key", "value3");
+					// assertThat(it.getValue()).containsEntry("key", "value3");
 				}) //
 				.thenCancel() //
 				.verify(Duration.ofSeconds(5));
@@ -151,9 +151,9 @@ public class StreamReceiverIntegrationTests {
 	@Test // DATAREDIS-864
 	public void shouldReceiveAsConsumerGroupMessages() {
 
-		StreamReceiver<String, String> receiver = StreamReceiver.create(connectionFactory);
+		StreamReceiver<String, String, String> receiver = StreamReceiver.create(connectionFactory);
 
-		Flux<StreamMessage<String, String>> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"),
+		Flux<MapRecord<String, String, String>> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"),
 				StreamOffset.create("my-stream", ReadOffset.lastConsumed()));
 
 		// required to initialize stream
@@ -165,11 +165,13 @@ public class StreamReceiverIntegrationTests {
 				.consumeNextWith(it -> {
 
 					assertThat(it.getStream()).isEqualTo("my-stream");
-					assertThat(it.getBody()).containsEntry("key", "value");
+					// assertThat(it.getValue()).containsEntry("key", "value");
+					assertThat(it.getValue()).containsValue("value");
 				}).consumeNextWith(it -> {
 
 					assertThat(it.getStream()).isEqualTo("my-stream");
-					assertThat(it.getBody()).containsEntry("key2", "value2");
+					// assertThat(it.getValue()).containsEntry("key2", "value2");
+					assertThat(it.getValue()).containsValue("value2");
 				}) //
 				.thenCancel() //
 				.verify(Duration.ofSeconds(5));
@@ -178,12 +180,12 @@ public class StreamReceiverIntegrationTests {
 	@Test // DATAREDIS-864
 	public void shouldStopReceivingOnError() {
 
-		StreamReceiverOptions<String, String> options = StreamReceiverOptions.builder().pollTimeout(Duration.ofMillis(100))
-				.build();
+		StreamReceiverOptions<String, String, String> options = StreamReceiverOptions.builder()
+				.pollTimeout(Duration.ofMillis(100)).build();
 
-		StreamReceiver<String, String> receiver = StreamReceiver.create(connectionFactory, options);
+		StreamReceiver<String, String, String> receiver = StreamReceiver.create(connectionFactory, options);
 
-		Flux<StreamMessage<String, String>> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"),
+		Flux<MapRecord<String, String, String>> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"),
 				StreamOffset.create("my-stream", ReadOffset.lastConsumed()));
 
 		// required to initialize stream