25
25
import java .time .Duration ;
26
26
import java .util .ArrayList ;
27
27
import java .util .List ;
28
- import java .util .function .BiFunction ;
29
28
30
29
import org .springframework .core .convert .converter .Converter ;
31
30
import org .springframework .data .redis .connection .RedisStreamCommands .XClaimOptions ;
32
- import org .springframework .data .redis .connection .convert .ListConverter ;
33
31
import org .springframework .data .redis .connection .stream .ByteRecord ;
34
32
import org .springframework .data .redis .connection .stream .Consumer ;
35
33
import org .springframework .data .redis .connection .stream .PendingMessagesSummary ;
53
51
@ SuppressWarnings ({ "rawtypes" })
54
52
class StreamConverters {
55
53
56
- private static final Converter <List <StreamMessage <byte [], byte []>>, List <RecordId >> MESSAGEs_TO_IDs = new ListConverter <>(
57
- messageToIdConverter ());
58
-
59
- private static final BiFunction <List <PendingMessage >, String , org .springframework .data .redis .connection .stream .PendingMessages > PENDING_MESSAGES_CONVERTER = (
60
- source , groupName ) -> {
61
-
62
-
63
- List <org .springframework .data .redis .connection .stream .PendingMessage > messages = source .stream ()
64
- .map (it -> {
65
-
66
- RecordId id = RecordId .of (it .getId ());
67
- Consumer consumer = Consumer .from (groupName , it .getConsumer ());
68
-
69
- return new org .springframework .data .redis .connection .stream .PendingMessage (id , consumer ,
70
- Duration .ofMillis (it .getMsSinceLastDelivery ()), it .getRedeliveryCount ());
71
-
72
- }).toList ();
73
-
74
- return new org .springframework .data .redis .connection .stream .PendingMessages (groupName , messages );
75
-
76
- };
77
-
78
- private static final BiFunction <PendingMessages , String , PendingMessagesSummary > PENDING_MESSAGES_SUMMARY_CONVERTER = (
79
- source , groupName ) -> {
80
-
81
- org .springframework .data .domain .Range <String > range = source .getMessageIds ().isUnbounded ()
82
- ? org .springframework .data .domain .Range .unbounded ()
83
- : org .springframework .data .domain .Range .open (source .getMessageIds ().getLower ().getValue (),
84
- source .getMessageIds ().getUpper ().getValue ());
85
-
86
- return new PendingMessagesSummary (groupName , source .getCount (), range , source .getConsumerMessageCount ());
87
- };
88
-
89
54
/**
90
55
* Convert {@link StreamReadOptions} to Lettuce's {@link XReadArgs}.
91
56
*
@@ -126,7 +91,18 @@ static Converter<StreamMessage<byte[], byte[]>, RecordId> messageToIdConverter()
126
91
*/
127
92
static org .springframework .data .redis .connection .stream .PendingMessages toPendingMessages (String groupName ,
128
93
org .springframework .data .domain .Range <?> range , List <PendingMessage > source ) {
129
- return PENDING_MESSAGES_CONVERTER .apply (source , groupName ).withinRange (range );
94
+
95
+ List <org .springframework .data .redis .connection .stream .PendingMessage > messages = source .stream ().map (it -> {
96
+
97
+ RecordId id = RecordId .of (it .getId ());
98
+ Consumer consumer = Consumer .from (groupName , it .getConsumer ());
99
+
100
+ return new org .springframework .data .redis .connection .stream .PendingMessage (id , consumer ,
101
+ Duration .ofMillis (it .getMsSinceLastDelivery ()), it .getRedeliveryCount ());
102
+
103
+ }).toList ();
104
+
105
+ return new org .springframework .data .redis .connection .stream .PendingMessages (groupName , messages ).withinRange (range );
130
106
}
131
107
132
108
/**
@@ -138,7 +114,13 @@ static org.springframework.data.redis.connection.stream.PendingMessages toPendin
138
114
* @since 2.3
139
115
*/
140
116
static PendingMessagesSummary toPendingMessagesInfo (String groupName , PendingMessages source ) {
141
- return PENDING_MESSAGES_SUMMARY_CONVERTER .apply (source , groupName );
117
+
118
+ org .springframework .data .domain .Range <String > range = source .getMessageIds ().isUnbounded ()
119
+ ? org .springframework .data .domain .Range .unbounded ()
120
+ : org .springframework .data .domain .Range .open (source .getMessageIds ().getLower ().getValue (),
121
+ source .getMessageIds ().getUpper ().getValue ());
122
+
123
+ return new PendingMessagesSummary (groupName , source .getCount (), range , source .getConsumerMessageCount ());
142
124
}
143
125
144
126
/**
0 commit comments