Skip to content

Consider exclusive Range for Jedis Stream range/pending commands #2108

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

Consider exclusive Range for Jedis Stream range/pending commands #2108

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ab35ba7
Prepare issue branch.
mp911de Jun 30, 2021
aab5128
Consider exclusive Range for Jedis Stream range/pending commands.
mp911de Jun 30, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">

<modelVersion>4.0.0</modelVersion>

<groupId>org.springframework.data</groupId>
<artifactId>spring-data-redis</artifactId>
<version>2.6.0-SNAPSHOT</version>
<version>2.6.0-2044-SNAPSHOT</version>

<name>Spring Data Redis</name>

Expand Down
17 changes: 8 additions & 9 deletions src/main/java/org/springframework/data/redis/connection/jedis/StreamConverters.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
8 changes: 4 additions & 4 deletions ...st/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand All @@ -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);
Expand Down
57 changes: 47 additions & 10 deletions ...est/java/org/springframework/data/redis/core/DefaultStreamOperationsIntegrationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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() {

Expand All @@ -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() {

Expand Down