Skip to content

Commit f5ad90a

Browse files
jinkshowermp911de
authored andcommitted
Fix XAddOptions maxlen handling and XPendingOptions validation.
Closes #2982 Original pull request: #2985
1 parent 600f9dd commit f5ad90a

File tree

5 files changed

+104
-12
lines changed

5 files changed

+104
-12
lines changed

src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java

+14-8
Original file line numberDiff line numberDiff line change
@@ -335,7 +335,7 @@ public Long getMaxlen() {
335335
* @since 2.3
336336
*/
337337
public boolean hasMaxlen() {
338-
return maxlen != null && maxlen > 0;
338+
return maxlen != null;
339339
}
340340

341341
/**
@@ -654,7 +654,7 @@ default Mono<PendingMessagesSummary> xPending(ByteBuffer key, String groupName)
654654
Assert.notNull(key, "Key must not be null");
655655
Assert.notNull(groupName, "GroupName must not be null");
656656

657-
return xPendingSummary(Mono.just(new PendingRecordsCommand(key, groupName, null, Range.unbounded(), null))).next()
657+
return xPendingSummary(Mono.just(PendingRecordsCommand.pending(key, groupName))).next()
658658
.map(CommandResponse::getOutput);
659659
}
660660

@@ -695,7 +695,7 @@ default Mono<PendingMessages> xPending(ByteBuffer key, Consumer consumer) {
695695
*/
696696
@Nullable
697697
default Mono<PendingMessages> xPending(ByteBuffer key, String groupName, String consumerName) {
698-
return xPending(Mono.just(new PendingRecordsCommand(key, groupName, consumerName, Range.unbounded(), null))).next()
698+
return xPending(Mono.just(PendingRecordsCommand.pending(key, groupName).consumer(consumerName))).next()
699699
.map(CommandResponse::getOutput);
700700
}
701701

@@ -712,7 +712,7 @@ default Mono<PendingMessages> xPending(ByteBuffer key, String groupName, String
712712
* @since 2.3
713713
*/
714714
default Mono<PendingMessages> xPending(ByteBuffer key, String groupName, Range<?> range, Long count) {
715-
return xPending(Mono.just(new PendingRecordsCommand(key, groupName, null, range, count))).next()
715+
return xPending(Mono.just(PendingRecordsCommand.pending(key, groupName).range(range, count))).next()
716716
.map(CommandResponse::getOutput);
717717
}
718718

@@ -748,8 +748,8 @@ default Mono<PendingMessages> xPending(ByteBuffer key, Consumer consumer, Range<
748748
*/
749749
default Mono<PendingMessages> xPending(ByteBuffer key, String groupName, String consumerName, Range<?> range,
750750
Long count) {
751-
return xPending(Mono.just(new PendingRecordsCommand(key, groupName, consumerName, range, count))).next()
752-
.map(CommandResponse::getOutput);
751+
return xPending(Mono.just(PendingRecordsCommand.pending(key, groupName).consumer(consumerName).range(range, count)))
752+
.next().map(CommandResponse::getOutput);
753753
}
754754

755755
/**
@@ -801,9 +801,15 @@ static PendingRecordsCommand pending(ByteBuffer key, String groupName) {
801801
/**
802802
* Create new {@link PendingRecordsCommand} with given {@link Range} and limit.
803803
*
804+
* @param range must not be {@literal null}.
805+
* @param count the max number of messages to return. Must not be negative.
804806
* @return new instance of {@link XPendingOptions}.
805807
*/
806-
public PendingRecordsCommand range(Range<String> range, Long count) {
808+
public PendingRecordsCommand range(Range<?> range, Long count) {
809+
810+
Assert.notNull(range, "Range must not be null");
811+
Assert.isTrue(count > -1, "Count must not be negative");
812+
807813
return new PendingRecordsCommand(getKey(), groupName, consumerName, range, count);
808814
}
809815

@@ -855,7 +861,7 @@ public boolean hasConsumer() {
855861
* @return {@literal true} count is set.
856862
*/
857863
public boolean isLimited() {
858-
return count != null && count > -1;
864+
return count != null;
859865
}
860866
}
861867

src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java

+12-3
Original file line numberDiff line numberDiff line change
@@ -214,7 +214,7 @@ public Long getMaxlen() {
214214
* @return {@literal true} if {@literal MAXLEN} is set.
215215
*/
216216
public boolean hasMaxlen() {
217-
return maxlen != null && maxlen > 0;
217+
return maxlen != null;
218218
}
219219

220220
/**
@@ -788,19 +788,28 @@ public static XPendingOptions unbounded() {
788788
/**
789789
* Create new {@link XPendingOptions} with an unbounded {@link Range} ({@literal - +}).
790790
*
791-
* @param count the max number of messages to return. Must not be {@literal null}.
791+
* @param count the max number of messages to return. Must not be negative.
792792
* @return new instance of {@link XPendingOptions}.
793793
*/
794794
public static XPendingOptions unbounded(Long count) {
795+
796+
Assert.isTrue(count > -1, "Count must not be negative");
797+
795798
return new XPendingOptions(null, Range.unbounded(), count);
796799
}
797800

798801
/**
799802
* Create new {@link XPendingOptions} with given {@link Range} and limit.
800803
*
804+
* @param range must not be {@literal null}.
805+
* @param count the max number of messages to return. Must not be negative.
801806
* @return new instance of {@link XPendingOptions}.
802807
*/
803808
public static XPendingOptions range(Range<?> range, Long count) {
809+
810+
Assert.notNull(range, "Range must not be null");
811+
Assert.isTrue(count > -1, "Count must not be negative");
812+
804813
return new XPendingOptions(null, range, count);
805814
}
806815

@@ -848,7 +857,7 @@ public boolean hasConsumer() {
848857
* @return {@literal true} count is set.
849858
*/
850859
public boolean isLimited() {
851-
return count != null && count > -1;
860+
return count != null;
852861
}
853862
}
854863

src/main/java/org/springframework/data/redis/connection/stream/StreamReadOptions.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,7 @@ public StreamReadOptions block(Duration timeout) {
9696
*/
9797
public StreamReadOptions count(long count) {
9898

99-
Assert.isTrue(count > 0, "Count must be greater or equal to zero");
99+
Assert.isTrue(count > 0, "Count must be greater than zero");
100100

101101
return new StreamReadOptions(block, count, noack);
102102
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
package org.springframework.data.redis.connection;
2+
3+
import static org.assertj.core.api.Assertions.*;
4+
5+
import java.nio.ByteBuffer;
6+
7+
import org.junit.jupiter.api.Test;
8+
9+
import org.springframework.data.domain.Range;
10+
import org.springframework.data.redis.connection.ReactiveStreamCommands.PendingRecordsCommand;
11+
12+
/**
13+
* Unit tests for {@link ReactiveStreamCommands}.
14+
*
15+
* @author jinkshower
16+
*/
17+
class ReactiveStreamCommandsUnitTests {
18+
19+
@Test // GH-2982
20+
void pendingRecordsCommandRangeShouldThrowExceptionWhenRangeIsNull() {
21+
22+
ByteBuffer key = ByteBuffer.wrap("my-stream".getBytes());
23+
String groupName = "my-group";
24+
25+
PendingRecordsCommand command = PendingRecordsCommand.pending(key, groupName);
26+
27+
assertThatThrownBy(() -> command.range(null, 10L)).isInstanceOf(IllegalArgumentException.class);
28+
}
29+
30+
@Test // GH-2982
31+
void pendingRecordsCommandRangeShouldThrowExceptionWhenCountIsNegative() {
32+
33+
ByteBuffer key = ByteBuffer.wrap("my-stream".getBytes());
34+
String groupName = "my-group";
35+
36+
PendingRecordsCommand command = PendingRecordsCommand.pending(key, groupName);
37+
Range<?> range = Range.closed("0", "10");
38+
39+
assertThatThrownBy(() -> command.range(range, -1L)).isInstanceOf(IllegalArgumentException.class);
40+
}
41+
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
package org.springframework.data.redis.connection;
2+
3+
import static org.assertj.core.api.Assertions.*;
4+
5+
import org.junit.jupiter.api.Test;
6+
7+
import org.springframework.data.domain.Range;
8+
import org.springframework.data.redis.connection.RedisStreamCommands.XPendingOptions;
9+
10+
/**
11+
* Unit tests for {@link RedisStreamCommands}.
12+
*
13+
* @author jinkshower
14+
*/
15+
class RedisStreamCommandsUnitTests {
16+
17+
@Test // GH-2982
18+
void xPendingOptionsUnboundedShouldThrowExceptionWhenCountIsNegative() {
19+
20+
assertThatThrownBy(() -> XPendingOptions.unbounded(-1L)).isInstanceOf(IllegalArgumentException.class);
21+
}
22+
23+
@Test // GH-2982
24+
void xPendingOptionsRangeShouldThrowExceptionWhenRangeIsNull() {
25+
26+
assertThatThrownBy(() -> XPendingOptions.range(null, 10L)).isInstanceOf(IllegalArgumentException.class);
27+
}
28+
29+
@Test // GH-2982
30+
void xPendingOptionsRangeShouldThrowExceptionWhenCountIsNegative() {
31+
32+
Range<?> range = Range.closed("0", "10");
33+
34+
assertThatThrownBy(() -> XPendingOptions.range(range, -1L)).isInstanceOf(IllegalArgumentException.class);
35+
}
36+
}

0 commit comments

Comments
 (0)