From 51ea73afc68ab9f0fc9b074b7b2420fe6f4bdd8e Mon Sep 17 00:00:00 2001
From: Christoph Strobl <cstrobl@vmware.com>
Date: Tue, 1 Sep 2020 13:00:22 +0200
Subject: [PATCH] DATAREDIS-1207 - Fix LettuceStreamCommands.xPending Range
 conversion..

Original Pull Request: #556
---
 .../lettuce/LettuceStreamCommands.java        |  2 +-
 .../AbstractConnectionIntegrationTests.java   | 23 +++++++++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java
index de9c27ed4..af2f4cc38 100644
--- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java
+++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java
@@ -442,7 +442,7 @@ class LettuceStreamCommands implements RedisStreamCommands {
 	public PendingMessages xPending(byte[] key, String groupName, XPendingOptions options) {
 
 		byte[] group = LettuceConverters.toBytes(groupName);
-		io.lettuce.core.Range<String> range = RangeConverter.toRangeWithDefault(options.getRange(), "-", "+");
+		io.lettuce.core.Range<String> range = RangeConverter.toRangeWithDefault(options.getRange(), "-", "+", Function.identity());
 		io.lettuce.core.Limit limit = options.isLimited() ? io.lettuce.core.Limit.from(options.getCount())
 				: io.lettuce.core.Limit.unlimited();
 
diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java
index d74b53b2b..cf73f6397 100644
--- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java
+++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java
@@ -3212,6 +3212,29 @@ public abstract class AbstractConnectionIntegrationTests {
 		assertThat(pending.get(0).getIdAsString()).isNotNull();
 	}
 
+	@Test // DATAREDIS-1207
+	@IfProfileValue(name = "redisVersion", value = "5.0")
+	@WithRedisDriver({ RedisDriver.LETTUCE })
+	public void xPendingShouldWorkWithBoundedRange() {
+
+		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.xPending(KEY_1, "my-group", org.springframework.data.domain.Range.open("0-0", "+"), 10L));
+
+		List<Object> results = getResults();
+		assertThat(results).hasSize(4);
+		PendingMessages pending = (PendingMessages) results.get(3);
+
+		assertThat(pending.size()).isOne();
+		assertThat(pending.get(0).getConsumerName()).isEqualTo("my-consumer");
+		assertThat(pending.get(0).getGroupName()).isEqualTo("my-group");
+		assertThat(pending.get(0).getTotalDeliveryCount()).isOne();
+		assertThat(pending.get(0).getIdAsString()).isNotNull();
+	}
+
 	@Test // DATAREDIS-1084
 	@IfProfileValue(name = "redisVersion", value = "5.0")
 	@WithRedisDriver({ RedisDriver.LETTUCE })