Skip to content

Commit a3b710b

Browse files
sazzad16sazzad16
authored
Consider null values in empty StreamPendingSummary (#3793)
* Consider null values in empty StreamPendingSummary * java stream
1 parent a02b2eb commit a3b710b

File tree

2 files changed

+49
-32
lines changed

2 files changed

+49
-32
lines changed

‎src/main/java/redis/clients/jedis/BuilderFactory.java

+8-9
Original file line numberDiff line numberDiff line change
@@ -1812,15 +1812,14 @@ public StreamPendingSummary build(Object data) {
18121812
}
18131813

18141814
List<Object> objectList = (List<Object>) data;
1815-
long total = BuilderFactory.LONG.build(objectList.get(0));
1816-
String minId = SafeEncoder.encode((byte[]) objectList.get(1));
1817-
String maxId = SafeEncoder.encode((byte[]) objectList.get(2));
1818-
List<List<Object>> consumerObjList = (List<List<Object>>) objectList.get(3);
1819-
Map<String, Long> map = new HashMap<>(consumerObjList.size());
1820-
for (List<Object> consumerObj : consumerObjList) {
1821-
map.put(SafeEncoder.encode((byte[]) consumerObj.get(0)), Long.parseLong(SafeEncoder.encode((byte[]) consumerObj.get(1))));
1822-
}
1823-
return new StreamPendingSummary(total, new StreamEntryID(minId), new StreamEntryID(maxId), map);
1815+
long total = LONG.build(objectList.get(0));
1816+
StreamEntryID minId = STREAM_ENTRY_ID.build(objectList.get(1));
1817+
StreamEntryID maxId = STREAM_ENTRY_ID.build(objectList.get(2));
1818+
Map<String, Long> map = objectList.get(3) == null ? null
1819+
: ((List<List<Object>>) objectList.get(3)).stream().collect(
1820+
Collectors.toMap(pair -> STRING.build(pair.get(0)),
1821+
pair -> Long.parseLong(STRING.build(pair.get(1)))));
1822+
return new StreamPendingSummary(total, minId, maxId, map);
18241823
}
18251824

18261825
@Override

‎src/test/java/redis/clients/jedis/commands/jedis/StreamsCommandsTest.java

+41-23
Original file line numberDiff line numberDiff line change
@@ -599,13 +599,22 @@ public void xack() {
599599

600600
@Test
601601
public void xpendingWithParams() {
602+
final String stream = "xpendeing-stream";
603+
604+
assertEquals("OK", jedis.xgroupCreate(stream, "xpendeing-group", null, true));
605+
606+
// Get the summary from empty stream
607+
StreamPendingSummary emptySummary = jedis.xpending(stream, "xpendeing-group");
608+
assertEquals(0, emptySummary.getTotal());
609+
assertNull(emptySummary.getMinId());
610+
assertNull(emptySummary.getMaxId());
611+
assertNull(emptySummary.getConsumerMessageCount());
612+
602613
Map<String, String> map = new HashMap<>();
603614
map.put("f1", "v1");
604-
StreamEntryID id1 = jedis.xadd("xpendeing-stream", (StreamEntryID) null, map);
615+
StreamEntryID id1 = jedis.xadd(stream, (StreamEntryID) null, map);
605616

606-
assertEquals("OK", jedis.xgroupCreate("xpendeing-stream", "xpendeing-group", null, false));
607-
608-
Map<String, StreamEntryID> streamQeury1 = singletonMap("xpendeing-stream", StreamEntryID.XREADGROUP_UNDELIVERED_ENTRY);
617+
Map<String, StreamEntryID> streamQeury1 = singletonMap(stream, StreamEntryID.XREADGROUP_UNDELIVERED_ENTRY);
609618

610619
// Read the event from Stream put it on pending
611620
List<Entry<String, List<StreamEntry>>> range = jedis.xreadGroup("xpendeing-group",
@@ -614,8 +623,14 @@ public void xpendingWithParams() {
614623
assertEquals(1, range.get(0).getValue().size());
615624
assertEquals(map, range.get(0).getValue().get(0).getFields());
616625

626+
// Get the summary about the pending messages
627+
StreamPendingSummary pendingSummary = jedis.xpending(stream, "xpendeing-group");
628+
assertEquals(1, pendingSummary.getTotal());
629+
assertEquals(id1, pendingSummary.getMinId());
630+
assertEquals(1l, pendingSummary.getConsumerMessageCount().get("xpendeing-consumer").longValue());
631+
617632
// Get the pending event
618-
List<StreamPendingEntry> pendingRange = jedis.xpending("xpendeing-stream", "xpendeing-group",
633+
List<StreamPendingEntry> pendingRange = jedis.xpending(stream, "xpendeing-group",
619634
new XPendingParams().count(3).consumer("xpendeing-consumer"));
620635
assertEquals(1, pendingRange.size());
621636
assertEquals(id1, pendingRange.get(0).getID());
@@ -624,40 +639,41 @@ public void xpendingWithParams() {
624639
assertTrue(pendingRange.get(0).toString().contains("xpendeing-consumer"));
625640

626641
// Without consumer
627-
pendingRange = jedis.xpending("xpendeing-stream", "xpendeing-group", new XPendingParams().count(3));
642+
pendingRange = jedis.xpending(stream, "xpendeing-group", new XPendingParams().count(3));
628643
assertEquals(1, pendingRange.size());
629644
assertEquals(id1, pendingRange.get(0).getID());
630645
assertEquals(1, pendingRange.get(0).getDeliveredTimes());
631646
assertEquals("xpendeing-consumer", pendingRange.get(0).getConsumerName());
632647

633648
// with idle
634-
pendingRange = jedis.xpending("xpendeing-stream", "xpendeing-group",
649+
pendingRange = jedis.xpending(stream, "xpendeing-group",
635650
new XPendingParams().idle(Duration.ofMinutes(1).toMillis()).count(3));
636651
assertEquals(0, pendingRange.size());
637652
}
638653

639654
@Test
640655
public void xpendingRange() {
656+
final String stream = "xpendeing-stream";
641657
Map<String, String> map = new HashMap<>();
642658
map.put("foo", "bar");
643-
StreamEntryID m1 = jedis.xadd("xpendeing-stream", (StreamEntryID) null, map);
644-
StreamEntryID m2 = jedis.xadd("xpendeing-stream", (StreamEntryID) null, map);
645-
jedis.xgroupCreate("xpendeing-stream", "xpendeing-group", null, false);
659+
StreamEntryID m1 = jedis.xadd(stream, (StreamEntryID) null, map);
660+
StreamEntryID m2 = jedis.xadd(stream, (StreamEntryID) null, map);
661+
jedis.xgroupCreate(stream, "xpendeing-group", null, false);
646662

647663
// read 1 message from the group with each consumer
648-
Map<String, StreamEntryID> streamQeury = singletonMap("xpendeing-stream", StreamEntryID.XREADGROUP_UNDELIVERED_ENTRY);
664+
Map<String, StreamEntryID> streamQeury = singletonMap(stream, StreamEntryID.XREADGROUP_UNDELIVERED_ENTRY);
649665
jedis.xreadGroup("xpendeing-group", "consumer1", XReadGroupParams.xReadGroupParams().count(1), streamQeury);
650666
jedis.xreadGroup("xpendeing-group", "consumer2", XReadGroupParams.xReadGroupParams().count(1), streamQeury);
651667

652-
List<StreamPendingEntry> response = jedis.xpending("xpendeing-stream", "xpendeing-group",
668+
List<StreamPendingEntry> response = jedis.xpending(stream, "xpendeing-group",
653669
XPendingParams.xPendingParams("(0", "+", 5));
654670
assertEquals(2, response.size());
655671
assertEquals(m1, response.get(0).getID());
656672
assertEquals("consumer1", response.get(0).getConsumerName());
657673
assertEquals(m2, response.get(1).getID());
658674
assertEquals("consumer2", response.get(1).getConsumerName());
659675

660-
response = jedis.xpending("xpendeing-stream", "xpendeing-group",
676+
response = jedis.xpending(stream, "xpendeing-group",
661677
XPendingParams.xPendingParams(StreamEntryID.MINIMUM_ID, StreamEntryID.MAXIMUM_ID, 5));
662678
assertEquals(2, response.size());
663679
assertEquals(m1, response.get(0).getID());
@@ -668,18 +684,19 @@ public void xpendingRange() {
668684

669685
@Test
670686
public void xclaimWithParams() {
687+
final String stream = "xpendeing-stream";
671688
Map<String, String> map = new HashMap<>();
672689
map.put("f1", "v1");
673-
jedis.xadd("xpendeing-stream", (StreamEntryID) null, map);
690+
jedis.xadd(stream, (StreamEntryID) null, map);
674691

675-
assertEquals("OK", jedis.xgroupCreate("xpendeing-stream", "xpendeing-group", null, false));
692+
assertEquals("OK", jedis.xgroupCreate(stream, "xpendeing-group", null, false));
676693

677694
// Read the event from Stream put it on pending
678695
jedis.xreadGroup("xpendeing-group", "xpendeing-consumer", XReadGroupParams.xReadGroupParams().count(1).block(1),
679-
singletonMap("xpendeing-stream", StreamEntryID.XREADGROUP_UNDELIVERED_ENTRY));
696+
singletonMap(stream, StreamEntryID.XREADGROUP_UNDELIVERED_ENTRY));
680697

681698
// Get the pending event
682-
List<StreamPendingEntry> pendingRange = jedis.xpending("xpendeing-stream", "xpendeing-group",
699+
List<StreamPendingEntry> pendingRange = jedis.xpending(stream, "xpendeing-group",
683700
XPendingParams.xPendingParams().count(3).consumer("xpendeing-consumer"));
684701

685702
// Sleep for 100ms so we can claim events pending for more than 50ms
@@ -689,7 +706,7 @@ public void xclaimWithParams() {
689706
e.printStackTrace();
690707
}
691708

692-
List<StreamEntry> streamEntrys = jedis.xclaim("xpendeing-stream", "xpendeing-group",
709+
List<StreamEntry> streamEntrys = jedis.xclaim(stream, "xpendeing-group",
693710
"xpendeing-consumer2", 50, XClaimParams.xClaimParams().idle(0).retryCount(0),
694711
pendingRange.get(0).getID());
695712
assertEquals(1, streamEntrys.size());
@@ -699,18 +716,19 @@ public void xclaimWithParams() {
699716

700717
@Test
701718
public void xclaimJustId() {
719+
final String stream = "xpendeing-stream";
702720
Map<String, String> map = new HashMap<>();
703721
map.put("f1", "v1");
704-
jedis.xadd("xpendeing-stream", (StreamEntryID) null, map);
722+
jedis.xadd(stream, (StreamEntryID) null, map);
705723

706-
assertEquals("OK", jedis.xgroupCreate("xpendeing-stream", "xpendeing-group", null, false));
724+
assertEquals("OK", jedis.xgroupCreate(stream, "xpendeing-group", null, false));
707725

708726
// Read the event from Stream put it on pending
709727
jedis.xreadGroup("xpendeing-group", "xpendeing-consumer", XReadGroupParams.xReadGroupParams().count(1).block(1),
710-
singletonMap("xpendeing-stream", StreamEntryID.XREADGROUP_UNDELIVERED_ENTRY));
728+
singletonMap(stream, StreamEntryID.XREADGROUP_UNDELIVERED_ENTRY));
711729

712730
// Get the pending event
713-
List<StreamPendingEntry> pendingRange = jedis.xpending("xpendeing-stream", "xpendeing-group",
731+
List<StreamPendingEntry> pendingRange = jedis.xpending(stream, "xpendeing-group",
714732
XPendingParams.xPendingParams().count(3).consumer("xpendeing-consumer"));
715733
// Sleep for 100ms so we can claim events pending for more than 50ms
716734
try {
@@ -719,7 +737,7 @@ public void xclaimJustId() {
719737
e.printStackTrace();
720738
}
721739

722-
List<StreamEntryID> streamEntryIDS = jedis.xclaimJustId("xpendeing-stream", "xpendeing-group",
740+
List<StreamEntryID> streamEntryIDS = jedis.xclaimJustId(stream, "xpendeing-group",
723741
"xpendeing-consumer2", 50, XClaimParams.xClaimParams().idle(0).retryCount(0),
724742
pendingRange.get(0).getID());
725743
assertEquals(1, streamEntryIDS.size());

0 commit comments

Comments
 (0)