Polishing.

Add support for Jedis Streams using JedisCluster. Add fromMany(…) for non-pipelined usage to JedisInvoker.

Reformat code, add author tags. Extract Jedis-specific stream type converters to StreamConverters. Properly convert StreamEntry and StreamEntryID into list/map. Update tests.

See #1711
Original pull request: #1977.

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java

@@ -303,6 +303,15 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 		return new JedisClusterSetCommands(this);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisConnection#streamCommands()
+	 */
+	@Override
+	public RedisStreamCommands streamCommands() {
+		return new JedisClusterStreamCommands(this);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisConnection#zSetCommands()

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStreamCommands.java

@@ -0,0 +1,416 @@
+/*
+ * Copyright 2021 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
+ *
+ *      https://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.jedis;
+
+import static org.springframework.data.redis.connection.jedis.StreamConverters.*;
+
+import redis.clients.jedis.BuilderFactory;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.domain.Range;
+import org.springframework.data.redis.connection.RedisStreamCommands;
+import org.springframework.data.redis.connection.RedisZSetCommands;
+import org.springframework.data.redis.connection.stream.ByteRecord;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.PendingMessages;
+import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.RecordId;
+import org.springframework.data.redis.connection.stream.StreamInfo;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.connection.stream.StreamReadOptions;
+import org.springframework.util.Assert;
+
+/**
+ * @author Dengliming
+ * @since 2.3
+ */
+class JedisClusterStreamCommands implements RedisStreamCommands {
+
+	private final JedisClusterConnection connection;
+
+	JedisClusterStreamCommands(JedisClusterConnection connection) {
+		this.connection = connection;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xAck(byte[], String, org.springframework.data.redis.connection.stream.RecordId[])
+	 */
+	@Override
+	public Long xAck(byte[] key, String group, RecordId... recordIds) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.hasText(group, "Group name must not be null or empty!");
+		Assert.notNull(recordIds, "recordIds must not be null!");
+
+		try {
+			return connection.getCluster().xack(key, JedisConverters.toBytes(group),
+					entryIdsToBytes(Arrays.asList(recordIds)));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xAdd(MapRecord, XAddOptions)
+	 */
+	@Override
+	public RecordId xAdd(MapRecord<byte[], byte[], byte[]> record, XAddOptions options) {
+
+		Assert.notNull(record, "Record must not be null!");
+		Assert.notNull(record.getStream(), "Stream must not be null!");
+
+		byte[] id = JedisConverters.toBytes(record.getId().getValue());
+		long maxLength = Long.MAX_VALUE;
+		if (options.hasMaxlen()) {
+			maxLength = options.getMaxlen();
+		}
+
+		try {
+			return RecordId.of(JedisConverters
+					.toString(connection.getCluster().xadd(record.getStream(), id, record.getValue(), maxLength, false)));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xClaimJustId(byte[], java.lang.String, java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions)
+	 */
+	@Override
+	public List<RecordId> xClaimJustId(byte[] key, String group, String newOwner, XClaimOptions options) {
+		throw new UnsupportedOperationException("JedisCluster does not support xClaimJustId.");
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xClaim(byte[], java.lang.String, java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions)
+	 */
+	@Override
+	public List<ByteRecord> xClaim(byte[] key, String group, String newOwner, XClaimOptions options) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(group, "Group must not be null!");
+		Assert.notNull(newOwner, "NewOwner must not be null!");
+
+		long minIdleTime = options.getMinIdleTime() == null ? -1L : options.getMinIdleTime().toMillis();
+		int retryCount = options.getRetryCount() == null ? -1 : options.getRetryCount().intValue();
+		long unixTime = options.getUnixTime() == null ? -1L : options.getUnixTime().toEpochMilli();
+
+		try {
+			return convertToByteRecord(key,
+					connection.getCluster().xclaim(key, JedisConverters.toBytes(group), JedisConverters.toBytes(newOwner),
+							minIdleTime, unixTime, retryCount, options.isForce(), entryIdsToBytes(options.getIds())));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xDel(byte[], java.lang.String[])
+	 */
+	@Override
+	public Long xDel(byte[] key, RecordId... recordIds) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(recordIds, "recordIds must not be null!");
+
+		try {
+			return connection.getCluster().xdel(key, entryIdsToBytes(Arrays.asList(recordIds)));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupCreate(byte[], org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset, java.lang.String)
+	 */
+	@Override
+	public String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset) {
+		return xGroupCreate(key, groupName, readOffset, false);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupCreate(byte[], org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset, java.lang.String, boolean)
+	 */
+	@Override
+	public String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset, boolean mkStream) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.hasText(groupName, "Group name must not be null or empty!");
+		Assert.notNull(readOffset, "ReadOffset must not be null!");
+
+		try {
+			return connection.getCluster().xgroupCreate(key, JedisConverters.toBytes(groupName),
+					JedisConverters.toBytes(readOffset.getOffset()), mkStream);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupDelConsumer(byte[], org.springframework.data.redis.connection.RedisStreamCommands.Consumer)
+	 */
+	@Override
+	public Boolean xGroupDelConsumer(byte[] key, Consumer consumer) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(consumer, "Consumer must not be null!");
+
+		try {
+			return connection.getCluster().xgroupDelConsumer(key, JedisConverters.toBytes(consumer.getGroup()),
+					JedisConverters.toBytes(consumer.getName())) != 0L;
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupDestroy(byte[], java.lang.String)
+	 */
+	@Override
+	public Boolean xGroupDestroy(byte[] key, String groupName) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.hasText(groupName, "Group name must not be null or empty!");
+
+		try {
+			return connection.getCluster().xgroupDestroy(key, JedisConverters.toBytes(groupName)) != 0L;
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfo(byte[])
+	 */
+	@Override
+	public StreamInfo.XInfoStream xInfo(byte[] key) {
+		throw new UnsupportedOperationException("JedisCluster does not support XINFO.");
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfoGroups(byte[])
+	 */
+	@Override
+	public StreamInfo.XInfoGroups xInfoGroups(byte[] key) {
+		throw new UnsupportedOperationException("JedisCluster does not support XINFO.");
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfoConsumers(byte[], java.lang.String)
+	 */
+	@Override
+	public StreamInfo.XInfoConsumers xInfoConsumers(byte[] key, String groupName) {
+		throw new UnsupportedOperationException("JedisCluster does not support XINFO.");
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xLen(byte[])
+	 */
+	@Override
+	public Long xLen(byte[] key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		try {
+			return connection.getCluster().xlen(key);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xPending(byte[], java.lang.String)
+	 */
+	@Override
+	public PendingMessagesSummary xPending(byte[] key, String groupName) {
+		throw new UnsupportedOperationException("Jedis does not support returning PendingMessagesSummary.");
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xPending(byte[], java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XPendingOptions)
+	 */
+	@Override
+	public PendingMessages xPending(byte[] key, String groupName, XPendingOptions options) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(groupName, "GroupName must not be null!");
+
+		Range<String> range = (Range<String>) options.getRange();
+		byte[] group = JedisConverters.toBytes(groupName);
+
+		try {
+
+			List<byte[]> response = connection.getCluster().xpending(key, group,
+					JedisConverters.toBytes(getLowerValue(range)), JedisConverters.toBytes(getUpperValue(range)),
+					options.getCount().intValue(), JedisConverters.toBytes(options.getConsumerName()));
+
+			return StreamConverters.toPendingMessages(groupName, range,
+					BuilderFactory.STREAM_PENDING_ENTRY_LIST.build(response));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xRange(byte[], org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
+	 */
+	@Override
+	public List<ByteRecord> xRange(byte[] key, Range<String> range, RedisZSetCommands.Limit limit) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(range, "Range must not be null!");
+		Assert.notNull(limit, "Limit must not be null!");
+
+		int count = limit.isUnlimited() ? Integer.MAX_VALUE : limit.getCount();
+
+		try {
+			return convertToByteRecord(key, connection.getCluster().xrange(key, JedisConverters.toBytes(getLowerValue(range)),
+					JedisConverters.toBytes(getUpperValue(range)), count));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xRead(org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions, org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset[])
+	 */
+	@Override
+	public List<ByteRecord> xRead(StreamReadOptions readOptions, StreamOffset<byte[]>... streams) {
+
+		Assert.notNull(readOptions, "StreamReadOptions must not be null!");
+		Assert.notNull(streams, "StreamOffsets must not be null!");
+
+		long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock();
+		int count = readOptions.getCount() != null ? readOptions.getCount().intValue() : Integer.MAX_VALUE;
+
+		try {
+
+			List<byte[]> xread = connection.getCluster().xread(count, block, toStreamOffsets(streams));
+
+			if (xread == null) {
+				return Collections.emptyList();
+			}
+
+			return StreamConverters.convertToByteRecords(xread);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xReadGroup(org.springframework.data.redis.connection.RedisStreamCommands.Consumer, org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions, org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset[])
+	 */
+	@Override
+	public List<ByteRecord> xReadGroup(Consumer consumer, StreamReadOptions readOptions,
+			StreamOffset<byte[]>... streams) {
+
+		Assert.notNull(consumer, "Consumer must not be null!");
+		Assert.notNull(readOptions, "StreamReadOptions must not be null!");
+		Assert.notNull(streams, "StreamOffsets must not be null!");
+
+		long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock();
+		int count = readOptions.getCount() == null ? -1 : readOptions.getCount().intValue();
+
+		try {
+
+			List<byte[]> xread = connection.getCluster().xreadGroup(JedisConverters.toBytes(consumer.getGroup()),
+					JedisConverters.toBytes(consumer.getName()), count, block, readOptions.isNoack(), toStreamOffsets(streams));
+
+			if (xread == null) {
+				return Collections.emptyList();
+			}
+
+			return StreamConverters.convertToByteRecords(xread);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xRevRange(byte[], org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
+	 */
+	@Override
+	public List<ByteRecord> xRevRange(byte[] key, Range<String> range, RedisZSetCommands.Limit limit) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(range, "Range must not be null!");
+		Assert.notNull(limit, "Limit must not be null!");
+
+		int count = limit.isUnlimited() ? Integer.MAX_VALUE : limit.getCount();
+
+		try {
+			return convertToByteRecord(key, connection.getCluster().xrevrange(key,
+					JedisConverters.toBytes(getUpperValue(range)), JedisConverters.toBytes(getLowerValue(range)), count));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xTrim(byte[], long)
+	 */
+	@Override
+	public Long xTrim(byte[] key, long count) {
+		return xTrim(key, count, false);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xTrim(byte[], long, boolean)
+	 */
+	@Override
+	public Long xTrim(byte[] key, long count, boolean approximateTrimming) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		try {
+			return connection.getCluster().xtrim(key, count, approximateTrimming);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	private DataAccessException convertJedisAccessException(Exception ex) {
+		return connection.convertJedisAccessException(ex);
+	}
+
+}

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -65,6 +65,7 @@ import org.springframework.util.StringUtils;
  * @author Mark Paluch
  * @author Ninad Divadkar
  * @author Guy Korland
+ * @author Dengliming
  */
 public class JedisConnection extends AbstractRedisConnection {
 

src/main/java/org/springframework/data/redis/connection/jedis/JedisInvoker.java

@@ -363,6 +363,21 @@ class JedisInvoker {
 				it -> pipelineFunction.apply(it, t1, t2, t3, t4, t5, t6));
 	}
 
+	/**
+	 * Compose a invocation pipeline from the {@link ConnectionFunction0} that returns a {@link Collection}-like result
+	 * and return a {@link ManyInvocationSpec} for further composition.
+	 *
+	 * @param function must not be {@literal null}.
+	 */
+	<R extends Collection<E>, E> ManyInvocationSpec<E> fromMany(ConnectionFunction0<R> function) {
+
+		Assert.notNull(function, "ConnectionFunction must not be null!");
+
+		return fromMany(function, connection -> {
+			throw new UnsupportedOperationException("Operation not supported in pipelining/transaction mode");
+		});
+	}
+
 	/**
 	 * Compose a invocation pipeline from the {@link ConnectionFunction0} that returns a {@link Collection}-like result
 	 * and return a {@link ManyInvocationSpec} for further composition.

src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java

@@ -15,13 +15,16 @@
  */
 package org.springframework.data.redis.connection.jedis;
 
-import static org.springframework.data.redis.connection.jedis.StreamConverters.convertToByteRecord;
+import redis.clients.jedis.BinaryJedis;
+import redis.clients.jedis.BuilderFactory;
+import redis.clients.jedis.MultiKeyPipelineBase;
+import redis.clients.jedis.StreamConsumersInfo;
+import redis.clients.jedis.StreamGroupInfo;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
 
 import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.RedisStreamCommands;
@@ -38,12 +41,6 @@ import org.springframework.data.redis.connection.stream.StreamOffset;
 import org.springframework.data.redis.connection.stream.StreamReadOptions;
 import org.springframework.util.Assert;
 
-import redis.clients.jedis.BinaryJedis;
-import redis.clients.jedis.BuilderFactory;
-import redis.clients.jedis.MultiKeyPipelineBase;
-import redis.clients.jedis.StreamConsumersInfo;
-import redis.clients.jedis.StreamGroupInfo;
-
 /**
  * @author Dengliming
  * @since 2.3
@@ -62,12 +59,13 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public Long xAck(byte[] key, String group, RecordId... recordIds) {
+
 		Assert.notNull(key, "Key must not be null!");
 		Assert.hasText(group, "Group name must not be null or empty!");
 		Assert.notNull(recordIds, "recordIds must not be null!");
 
 		return connection.invoke().just(BinaryJedis::xack, MultiKeyPipelineBase::xack, key, JedisConverters.toBytes(group),
-				entryIdsToBytes(recordIds));
+				StreamConverters.entryIdsToBytes(Arrays.asList(recordIds)));
 	}
 
 	/*
@@ -76,11 +74,12 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public RecordId xAdd(MapRecord<byte[], byte[], byte[]> record, XAddOptions options) {
+
 		Assert.notNull(record, "Record must not be null!");
 		Assert.notNull(record.getStream(), "Stream must not be null!");
 
 		byte[] id = JedisConverters.toBytes(record.getId().getValue());
-		Long maxLength = Long.MAX_VALUE;
+		long maxLength = Long.MAX_VALUE;
 		if (options.hasMaxlen()) {
 			maxLength = options.getMaxlen();
 		}
@@ -104,21 +103,22 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public List<ByteRecord> xClaim(byte[] key, String group, String newOwner, XClaimOptions options) {
+
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(group, "Group must not be null!");
 		Assert.notNull(newOwner, "NewOwner must not be null!");
 
-		final long minIdleTime = options.getMinIdleTime() == null ? -1L : options.getMinIdleTime().toMillis();
-		final int retryCount = options.getRetryCount() == null ? -1 : options.getRetryCount().intValue();
-		final long unixTime = options.getUnixTime() == null ? -1L : options.getUnixTime().toEpochMilli();
+		long minIdleTime = options.getMinIdleTime() == null ? -1L : options.getMinIdleTime().toMillis();
+		int retryCount = options.getRetryCount() == null ? -1 : options.getRetryCount().intValue();
+		long unixTime = options.getUnixTime() == null ? -1L : options.getUnixTime().toEpochMilli();
 
 		return connection.invoke()
 				.from(
 						it -> it.xclaim(key, JedisConverters.toBytes(group), JedisConverters.toBytes(newOwner), minIdleTime,
-								unixTime, retryCount, options.isForce(), entryIdsToBytes(options.getIds())),
+								unixTime, retryCount, options.isForce(), StreamConverters.entryIdsToBytes(options.getIds())),
 						it -> it.xclaim(key, JedisConverters.toBytes(group), JedisConverters.toBytes(newOwner), minIdleTime,
-								unixTime, retryCount, options.isForce(), entryIdsToBytes(options.getIds())))
-				.get(r -> convertToByteRecord(key, r));
+								unixTime, retryCount, options.isForce(), StreamConverters.entryIdsToBytes(options.getIds())))
+				.get(r -> StreamConverters.convertToByteRecord(key, r));
 	}
 
 	/*
@@ -127,10 +127,12 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public Long xDel(byte[] key, RecordId... recordIds) {
+
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(recordIds, "recordIds must not be null!");
 
-		return connection.invoke().just(BinaryJedis::xdel, MultiKeyPipelineBase::xdel, key, entryIdsToBytes(recordIds));
+		return connection.invoke().just(BinaryJedis::xdel, MultiKeyPipelineBase::xdel, key,
+				StreamConverters.entryIdsToBytes(Arrays.asList(recordIds)));
 	}
 
 	/*
@@ -148,6 +150,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset, boolean mkStream) {
+
 		Assert.notNull(key, "Key must not be null!");
 		Assert.hasText(groupName, "Group name must not be null or empty!");
 		Assert.notNull(readOffset, "ReadOffset must not be null!");
@@ -162,6 +165,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public Boolean xGroupDelConsumer(byte[] key, Consumer consumer) {
+
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(consumer, "Consumer must not be null!");
 
@@ -175,6 +179,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public Boolean xGroupDestroy(byte[] key, String groupName) {
+
 		Assert.notNull(key, "Key must not be null!");
 		Assert.hasText(groupName, "Group name must not be null or empty!");
 
@@ -189,6 +194,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public StreamInfo.XInfoStream xInfo(byte[] key) {
+
 		Assert.notNull(key, "Key must not be null!");
 
 		if (isQueueing() || isPipelined()) {
@@ -197,7 +203,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 
 		return connection.invoke().just(it -> {
 			redis.clients.jedis.StreamInfo streamInfo = it.xinfoStream(key);
-			return StreamInfo.XInfoStream.fromList(mapToList(streamInfo.getStreamInfo()));
+			return StreamInfo.XInfoStream.fromList(StreamConverters.mapToList(streamInfo.getStreamInfo()));
 		});
 	}
 
@@ -207,6 +213,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public StreamInfo.XInfoGroups xInfoGroups(byte[] key) {
+
 		Assert.notNull(key, "Key must not be null!");
 
 		if (isQueueing() || isPipelined()) {
@@ -216,7 +223,8 @@ class JedisStreamCommands implements RedisStreamCommands {
 		return connection.invoke().just(it -> {
 			List<StreamGroupInfo> streamGroupInfos = it.xinfoGroup(key);
 			List<Object> sources = new ArrayList<>();
-			streamGroupInfos.forEach(streamGroupInfo -> sources.add(mapToList(streamGroupInfo.getGroupInfo())));
+			streamGroupInfos
+					.forEach(streamGroupInfo -> sources.add(StreamConverters.mapToList(streamGroupInfo.getGroupInfo())));
 			return StreamInfo.XInfoGroups.fromList(sources);
 		});
 	}
@@ -227,6 +235,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public StreamInfo.XInfoConsumers xInfoConsumers(byte[] key, String groupName) {
+
 		Assert.notNull(key, "Key must not be null!");
 		Assert.hasText(groupName, "Group name must not be null or empty!");
 
@@ -238,7 +247,8 @@ class JedisStreamCommands implements RedisStreamCommands {
 			List<StreamConsumersInfo> streamConsumersInfos = it.xinfoConsumers(key, JedisConverters.toBytes(groupName));
 			List<Object> sources = new ArrayList<>();
 			streamConsumersInfos
-					.forEach(streamConsumersInfo -> sources.add(mapToList(streamConsumersInfo.getConsumerInfo())));
+					.forEach(
+							streamConsumersInfo -> sources.add(StreamConverters.mapToList(streamConsumersInfo.getConsumerInfo())));
 			return StreamInfo.XInfoConsumers.fromList(groupName, sources);
 		});
 	}
@@ -249,6 +259,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public Long xLen(byte[] key) {
+
 		Assert.notNull(key, "Key must not be null!");
 
 		return connection.invoke().just(BinaryJedis::xlen, MultiKeyPipelineBase::xlen, key);
@@ -269,6 +280,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public PendingMessages xPending(byte[] key, String groupName, XPendingOptions options) {
+
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(groupName, "GroupName must not be null!");
 
@@ -277,9 +289,10 @@ class JedisStreamCommands implements RedisStreamCommands {
 
 		return connection.invoke().from((it, t1, t2, t3, t4, t5, t6) -> {
 			Object r = it.xpending(t1, t2, t3, t4, t5, t6);
+
 			return BuilderFactory.STREAM_PENDING_ENTRY_LIST.build(r);
-		}, MultiKeyPipelineBase::xpending, key, group, JedisConverters.toBytes(getLowerValue(range)),
-				JedisConverters.toBytes(getUpperValue(range)), options.getCount().intValue(),
+		}, MultiKeyPipelineBase::xpending, key, group, JedisConverters.toBytes(StreamConverters.getLowerValue(range)),
+				JedisConverters.toBytes(StreamConverters.getUpperValue(range)), options.getCount().intValue(),
 				JedisConverters.toBytes(options.getConsumerName()))
 				.get(r -> StreamConverters.toPendingMessages(groupName, range, r));
 	}
@@ -290,6 +303,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public List<ByteRecord> xRange(byte[] key, Range<String> range, RedisZSetCommands.Limit limit) {
+
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 		Assert.notNull(limit, "Limit must not be null!");
@@ -298,11 +312,11 @@ class JedisStreamCommands implements RedisStreamCommands {
 
 		return connection.invoke()
 				.from(
-						it -> it.xrange(key, JedisConverters.toBytes(getLowerValue(range)),
-								JedisConverters.toBytes(getUpperValue(range)), count),
-						it -> it.xrange(key, JedisConverters.toBytes(getLowerValue(range)),
-								JedisConverters.toBytes(getUpperValue(range)), count))
-				.get(r -> convertToByteRecord(key, r));
+						it -> it.xrange(key, JedisConverters.toBytes(StreamConverters.getLowerValue(range)),
+								JedisConverters.toBytes(StreamConverters.getUpperValue(range)), count),
+						it -> it.xrange(key, JedisConverters.toBytes(StreamConverters.getLowerValue(range)),
+								JedisConverters.toBytes(StreamConverters.getUpperValue(range)), count))
+				.get(r -> StreamConverters.convertToByteRecord(key, r));
 	}
 
 	/*
@@ -311,6 +325,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public List<ByteRecord> xRead(StreamReadOptions readOptions, StreamOffset<byte[]>... streams) {
+
 		Assert.notNull(readOptions, "StreamReadOptions must not be null!");
 		Assert.notNull(streams, "StreamOffsets must not be null!");
 
@@ -318,13 +333,11 @@ class JedisStreamCommands implements RedisStreamCommands {
 			throw new UnsupportedOperationException("'XREAD' cannot be called in pipeline / transaction mode.");
 		}
 
-		final long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock();
-		final int count = readOptions.getCount() != null ? readOptions.getCount().intValue() : Integer.MAX_VALUE;
+		long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock();
+		int count = readOptions.getCount() != null ? readOptions.getCount().intValue() : Integer.MAX_VALUE;
 
-		return connection.invoke().just(it -> {
-			List<byte[]> streamsEntries = it.xread(count, block, toStreamOffsets(streams));
-			return convertToByteRecord(streamsEntries);
-		});
+		return connection.invoke().from(it -> it.xread(count, block, StreamConverters.toStreamOffsets(streams)))
+				.getOrElse(StreamConverters::convertToByteRecords, Collections::emptyList);
 	}
 
 	/*
@@ -334,6 +347,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	@Override
 	public List<ByteRecord> xReadGroup(Consumer consumer, StreamReadOptions readOptions,
 			StreamOffset<byte[]>... streams) {
+
 		Assert.notNull(consumer, "Consumer must not be null!");
 		Assert.notNull(readOptions, "StreamReadOptions must not be null!");
 		Assert.notNull(streams, "StreamOffsets must not be null!");
@@ -342,14 +356,14 @@ class JedisStreamCommands implements RedisStreamCommands {
 			throw new UnsupportedOperationException("'XREADGROUP' cannot be called in pipeline / transaction mode.");
 		}
 
-		final long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock();
-		final int count = readOptions.getCount() == null ? -1 : readOptions.getCount().intValue();
+		long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock();
+		int count = readOptions.getCount() == null ? -1 : readOptions.getCount().intValue();
 
-		return connection.invoke().just(it -> {
-			List<byte[]> streamsEntries = it.xreadGroup(JedisConverters.toBytes(consumer.getGroup()),
-					JedisConverters.toBytes(consumer.getName()), count, block, readOptions.isNoack(), toStreamOffsets(streams));
-			return convertToByteRecord(streamsEntries);
-		});
+		return connection.invoke().from(it -> {
+
+			return it.xreadGroup(JedisConverters.toBytes(consumer.getGroup()), JedisConverters.toBytes(consumer.getName()),
+					count, block, readOptions.isNoack(), StreamConverters.toStreamOffsets(streams));
+		}).getOrElse(StreamConverters::convertToByteRecords, Collections::emptyList);
 	}
 
 	/*
@@ -358,6 +372,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public List<ByteRecord> xRevRange(byte[] key, Range<String> range, RedisZSetCommands.Limit limit) {
+
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 		Assert.notNull(limit, "Limit must not be null!");
@@ -365,8 +380,9 @@ class JedisStreamCommands implements RedisStreamCommands {
 		int count = limit.isUnlimited() ? Integer.MAX_VALUE : limit.getCount();
 		return connection.invoke()
 				.from(BinaryJedis::xrevrange, MultiKeyPipelineBase::xrevrange, key,
-						JedisConverters.toBytes(getUpperValue(range)), JedisConverters.toBytes(getLowerValue(range)), count)
-				.get(it -> convertToByteRecord(key, it));
+						JedisConverters.toBytes(StreamConverters.getUpperValue(range)),
+						JedisConverters.toBytes(StreamConverters.getLowerValue(range)), count)
+				.get(it -> StreamConverters.convertToByteRecord(key, it));
 	}
 
 	/*
@@ -384,6 +400,7 @@ class JedisStreamCommands implements RedisStreamCommands {
 	 */
 	@Override
 	public Long xTrim(byte[] key, long count, boolean approximateTrimming) {
+
 		Assert.notNull(key, "Key must not be null!");
 
 		return connection.invoke().just(BinaryJedis::xtrim, MultiKeyPipelineBase::xtrim, key, count, approximateTrimming);
@@ -397,57 +414,4 @@ class JedisStreamCommands implements RedisStreamCommands {
 		return connection.isQueueing();
 	}
 
-	private byte[][] entryIdsToBytes(RecordId[] recordIds) {
-
-		final byte[][] bids = new byte[recordIds.length][];
-		for (int i = 0; i < recordIds.length; ++i) {
-			RecordId id = recordIds[i];
-			bids[i] = JedisConverters.toBytes(id.getValue());
-		}
-
-		return bids;
-	}
-
-	private byte[][] entryIdsToBytes(List<RecordId> recordIds) {
-
-		final byte[][] bids = new byte[recordIds.size()][];
-		for (int i = 0; i < recordIds.size(); ++i) {
-			RecordId id = recordIds.get(i);
-			bids[i] = JedisConverters.toBytes(id.getValue());
-		}
-
-		return bids;
-	}
-
-	private String getLowerValue(Range<String> range) {
-
-		if (range.getLowerBound().equals(Range.Bound.unbounded())) {
-			return "-";
-		}
-
-		return range.getLowerBound().getValue().orElse("-");
-	}
-
-	private String getUpperValue(Range<String> range) {
-
-		if (range.getUpperBound().equals(Range.Bound.unbounded())) {
-			return "+";
-		}
-
-		return range.getUpperBound().getValue().orElse("+");
-	}
-
-	private List<Object> mapToList(Map<String, Object> map) {
-		List<Object> sources = new ArrayList<>(map.size() * 2);
-		map.forEach((k, v) -> {
-			sources.add(k);
-			sources.add(v);
-		});
-		return sources;
-	}
-
-	private Map<byte[], byte[]> toStreamOffsets(StreamOffset<byte[]>... streams) {
-		return Arrays.stream(streams)
-				.collect(Collectors.toMap(k -> k.getKey(), v -> JedisConverters.toBytes(v.getOffset().getOffset())));
-	}
 }

src/main/java/org/springframework/data/redis/connection/jedis/StreamConverters.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2021 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.
@@ -15,25 +15,29 @@
  */
 package org.springframework.data.redis.connection.jedis;
 
+import redis.clients.jedis.StreamEntry;
+import redis.clients.jedis.StreamEntryID;
+import redis.clients.jedis.StreamPendingEntry;
+import redis.clients.jedis.util.SafeEncoder;
+
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Collections;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.stream.ByteRecord;
 import org.springframework.data.redis.connection.stream.Consumer;
 import org.springframework.data.redis.connection.stream.PendingMessage;
 import org.springframework.data.redis.connection.stream.PendingMessages;
 import org.springframework.data.redis.connection.stream.RecordId;
+import org.springframework.data.redis.connection.stream.StreamOffset;
 import org.springframework.data.redis.connection.stream.StreamRecords;
 
-import redis.clients.jedis.StreamPendingEntry;
-import redis.clients.jedis.util.SafeEncoder;
-
 /**
  * Converters for Redis Stream-specific types.
  * <p/>
@@ -41,26 +45,83 @@ import redis.clients.jedis.util.SafeEncoder;
  * serialization/deserialization happens here).
  *
  * @author dengliming
+ * @author Mark Paluch
  * @since 2.3
  */
-@SuppressWarnings({ "rawtypes" })
 class StreamConverters {
 
-	static final List<ByteRecord> convertToByteRecord(byte[] key, Object source) {
-		if (null == source) {
-			return Collections.emptyList();
+	static byte[][] entryIdsToBytes(List<RecordId> recordIds) {
+
+		byte[][] target = new byte[recordIds.size()][];
+
+		for (int i = 0; i < recordIds.size(); ++i) {
+			RecordId id = recordIds.get(i);
+			target[i] = JedisConverters.toBytes(id.getValue());
 		}
+
+		return target;
+	}
+
+	static String getLowerValue(Range<String> range) {
+
+		if (range.getLowerBound().equals(Range.Bound.unbounded())) {
+			return "-";
+		}
+
+		return range.getLowerBound().getValue().orElse("-");
+	}
+
+	static String getUpperValue(Range<String> range) {
+
+		if (range.getUpperBound().equals(Range.Bound.unbounded())) {
+			return "+";
+		}
+
+		return range.getUpperBound().getValue().orElse("+");
+	}
+
+	static List<Object> mapToList(Map<String, Object> map) {
+
+		List<Object> sources = new ArrayList<>(map.size() * 2);
+		map.forEach((k, v) -> {
+			sources.add(k);
+
+			if (v instanceof StreamEntryID) {
+				sources.add(v.toString());
+			} else if (v instanceof StreamEntry) {
+				List<Object> entries = new ArrayList<>(2);
+				StreamEntry streamEntry = (StreamEntry) v;
+				entries.add(streamEntry.getID().toString());
+				entries.add(streamEntry.getFields());
+				sources.add(entries);
+			} else {
+				sources.add(v);
+			}
+		});
+		return sources;
+	}
+
+	static Map<byte[], byte[]> toStreamOffsets(StreamOffset<byte[]>[] streams) {
+		return Arrays.stream(streams)
+				.collect(Collectors.toMap(StreamOffset::getKey, v -> JedisConverters.toBytes(v.getOffset().getOffset())));
+	}
+
+	static List<ByteRecord> convertToByteRecord(byte[] key, Object source) {
+
 		List<List<Object>> objectList = (List<List<Object>>) source;
 		List<ByteRecord> result = new ArrayList<>(objectList.size() / 2);
+
 		if (objectList.isEmpty()) {
 			return result;
 		}
 
 		for (List<Object> res : objectList) {
+
 			if (res == null) {
 				result.add(null);
 				continue;
 			}
+
 			String entryIdString = SafeEncoder.encode((byte[]) res.get(0));
 			List<byte[]> hash = (List<byte[]>) res.get(1);
 
@@ -75,15 +136,15 @@ class StreamConverters {
 		return result;
 	}
 
-	static final List<ByteRecord> convertToByteRecord(List<byte[]> sources) {
-		if (sources == null) {
-			return Collections.emptyList();
-		}
-		List<ByteRecord> result = new ArrayList<>();
-		for (Object streamObj : sources) {
-			List<Object> stream = (List<Object>) streamObj;
+	static List<ByteRecord> convertToByteRecords(List<?> sources) {
+
+		List<ByteRecord> result = new ArrayList<>(sources.size() / 2);
+
+		for (Object source : sources) {
+			List<Object> stream = (List<Object>) source;
 			result.addAll(convertToByteRecord((byte[]) stream.get(0), stream.get(1)));
 		}
+
 		return result;
 	}
 
@@ -98,10 +159,6 @@ class StreamConverters {
 	static org.springframework.data.redis.connection.stream.PendingMessages toPendingMessages(String groupName,
 			org.springframework.data.domain.Range<?> range, List<StreamPendingEntry> response) {
 
-		if (null == response) {
-			return null;
-		}
-
 		List<PendingMessage> messages = response.stream()
 				.map(streamPendingEntry -> new PendingMessage(RecordId.of(streamPendingEntry.getID().toString()),
 						Consumer.from(groupName, streamPendingEntry.getConsumerName()),