Consider exclusive Range for Jedis Stream range/pending commands.

We now consider `Bound.exclusive(…)` for XRANGE, XREVRANGE and XPENDING commands through Jedis.

Closes #2044

Author: mp911de

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

@@ -63,21 +63,20 @@ static byte[][] entryIdsToBytes(List<RecordId> recordIds) {
 	}
 
 	static String getLowerValue(Range<String> range) {
-
-		if (range.getLowerBound().equals(Range.Bound.unbounded())) {
-			return "-";
-		}
-
-		return range.getLowerBound().getValue().orElse("-");
+		return getValue(range.getLowerBound(), "-");
 	}
 
 	static String getUpperValue(Range<String> range) {
+		return getValue(range.getUpperBound(), "+");
+	}
+
+	private static String getValue(Range.Bound<String> bound, String fallbackValue) {
 
-		if (range.getUpperBound().equals(Range.Bound.unbounded())) {
-			return "+";
+		if (bound.equals(Range.Bound.unbounded())) {
+			return fallbackValue;
 		}
 
-		return range.getUpperBound().getValue().orElse("+");
+		return bound.getValue().map(it -> bound.isInclusive() ? it : "(" + it).orElse(fallbackValue);
 	}
 
 	static List<Object> mapToList(Map<String, Object> map) {

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

@@ -3490,7 +3490,7 @@ public void xPendingShouldLoadPendingMessages() {
 		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("-", "+"), 10L));
+		actual.add(connection.xPending(KEY_1, "my-group", org.springframework.data.domain.Range.unbounded(), 10L));
 
 		List<Object> results = getResults();
 		assertThat(results).hasSize(4);
@@ -3535,7 +3535,7 @@ public void xPendingShouldLoadPendingMessagesForConsumer() {
 				StreamOffset.create(KEY_1, ReadOffset.lastConsumed())));
 
 		actual.add(connection.xPending(KEY_1, "my-group", "my-consumer",
-				org.springframework.data.domain.Range.open("-", "+"), 10L));
+				org.springframework.data.domain.Range.unbounded(), 10L));
 
 		List<Object> results = getResults();
 		assertThat(results).hasSize(4);
@@ -3558,7 +3558,7 @@ public void xPendingShouldLoadPendingMessagesForNonExistingConsumer() {
 				StreamOffset.create(KEY_1, ReadOffset.lastConsumed())));
 
 		actual.add(connection.xPending(KEY_1, "my-group", "my-consumer-2",
-				org.springframework.data.domain.Range.open("-", "+"), 10L));
+				org.springframework.data.domain.Range.unbounded(), 10L));
 
 		List<Object> results = getResults();
 		assertThat(results).hasSize(4);
@@ -3574,7 +3574,7 @@ void xPendingShouldLoadEmptyPendingMessages() {
 		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.xPending(KEY_1, "my-group", org.springframework.data.domain.Range.open("-", "+"), 10L));
+		actual.add(connection.xPending(KEY_1, "my-group", org.springframework.data.domain.Range.unbounded(), 10L));
 
 		List<Object> results = getResults();
 		assertThat(results).hasSize(3);

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

@@ -34,18 +34,10 @@
 import org.springframework.data.redis.connection.jedis.extension.JedisConnectionFactoryExtension;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.connection.lettuce.extension.LettuceConnectionFactoryExtension;
-import org.springframework.data.redis.connection.stream.Consumer;
-import org.springframework.data.redis.connection.stream.MapRecord;
-import org.springframework.data.redis.connection.stream.ObjectRecord;
-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.StreamOffset;
-import org.springframework.data.redis.connection.stream.StreamReadOptions;
-import org.springframework.data.redis.connection.stream.StreamRecords;
+import org.springframework.data.redis.connection.stream.*;
 import org.springframework.data.redis.test.condition.EnabledOnCommand;
 import org.springframework.data.redis.test.condition.EnabledOnRedisDriver;
+import org.springframework.data.redis.test.condition.EnabledOnRedisVersion;
 import org.springframework.data.redis.test.condition.RedisDetector;
 import org.springframework.data.redis.test.extension.RedisCluster;
 import org.springframework.data.redis.test.extension.RedisStanalone;
@@ -198,6 +190,28 @@ void rangeShouldReportMessages() {
 		assertThat(message.getId()).isEqualTo(messageId1);
 	}
 
+	@ParameterizedRedisTest // GH-2044
+	@EnabledOnRedisVersion("6.2")
+	void exclusiveRangeShouldReportMessages() {
+
+		K key = keyFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = hashValueFactory.instance();
+
+		RecordId messageId1 = streamOps.add(key, Collections.singletonMap(hashKey, value));
+		RecordId messageId2 = streamOps.add(key, Collections.singletonMap(hashKey, value));
+
+		List<MapRecord<K, HK, HV>> messages = streamOps.range(key,
+				Range.from(Bound.exclusive(messageId1.getValue())).to(Bound.inclusive(messageId2.getValue())));
+
+		assertThat(messages).hasSize(1).extracting(Record::getId).contains(messageId2);
+
+		messages = streamOps.range(key,
+				Range.from(Bound.inclusive(messageId1.getValue())).to(Bound.exclusive(messageId2.getValue())));
+
+		assertThat(messages).hasSize(1).extracting(Record::getId).contains(messageId1);
+	}
+
 	@ParameterizedRedisTest // DATAREDIS-864
 	void reverseRangeShouldReportMessages() {
 
@@ -213,6 +227,29 @@ void reverseRangeShouldReportMessages() {
 		assertThat(messages).hasSize(2).extracting("id").containsSequence(messageId2, messageId1);
 	}
 
+	@ParameterizedRedisTest // GH-2044
+	@EnabledOnRedisVersion("6.2")
+	void exclusiveReverseRangeShouldReportMessages() {
+
+		K key = keyFactory.instance();
+		HK hashKey = hashKeyFactory.instance();
+		HV value = hashValueFactory.instance();
+
+		RecordId messageId1 = streamOps.add(key, Collections.singletonMap(hashKey, value));
+		RecordId messageId2 = streamOps.add(key, Collections.singletonMap(hashKey, value));
+		RecordId messageId3 = streamOps.add(key, Collections.singletonMap(hashKey, value));
+
+		List<MapRecord<K, HK, HV>> messages = streamOps.reverseRange(key,
+				Range.from(Bound.exclusive(messageId1.getValue())).to(Bound.inclusive(messageId3.getValue())));
+
+		assertThat(messages).hasSize(2).extracting(Record::getId).containsSequence(messageId3, messageId2);
+
+		messages = streamOps.reverseRange(key,
+				Range.from(Bound.inclusive(messageId1.getValue())).to(Bound.exclusive(messageId3.getValue())));
+
+		assertThat(messages).hasSize(2).extracting(Record::getId).containsSequence(messageId2, messageId1);
+	}
+
 	@ParameterizedRedisTest // DATAREDIS-864
 	void reverseRangeShouldConvertSimpleMessages() {