27
27
import java .util .function .Function ;
28
28
29
29
import org .reactivestreams .Publisher ;
30
+ import org .springframework .dao .InvalidDataAccessApiUsageException ;
30
31
import org .springframework .data .redis .connection .ReactiveListCommands ;
31
32
import org .springframework .data .redis .connection .ReactiveListCommands .Direction ;
32
33
import org .springframework .data .redis .connection .ReactiveListCommands .LPosCommand ;
33
34
import org .springframework .data .redis .connection .RedisListCommands .Position ;
34
35
import org .springframework .data .redis .serializer .RedisSerializationContext ;
36
+ import org .springframework .lang .Nullable ;
35
37
import org .springframework .util .Assert ;
36
38
37
39
/**
@@ -59,7 +61,7 @@ public Flux<V> range(K key, long start, long end) {
59
61
60
62
Assert .notNull (key , "Key must not be null" );
61
63
62
- return createFlux (listCommands -> listCommands .lRange (rawKey (key ), start , end ).map (this ::readValue ));
64
+ return createFlux (connection -> connection .lRange (rawKey (key ), start , end ).map (this ::readRequiredValue ));
63
65
}
64
66
65
67
@ Override
@@ -172,8 +174,8 @@ public Mono<V> move(K sourceKey, Direction from, K destinationKey, Direction to)
172
174
Assert .notNull (from , "From direction must not be null" );
173
175
Assert .notNull (to , "To direction must not be null" );
174
176
175
- return createMono (listCommands ->
176
- listCommands . lMove ( rawKey ( sourceKey ), rawKey ( destinationKey ), from , to ). map (this ::readValue ));
177
+ return createMono (connection -> connection . lMove ( rawKey ( sourceKey ), rawKey ( destinationKey ), from , to )
178
+ . map (this ::readRequiredValue ));
177
179
}
178
180
179
181
@ Override
@@ -185,8 +187,8 @@ public Mono<V> move(K sourceKey, Direction from, K destinationKey, Direction to,
185
187
Assert .notNull (to , "To direction must not be null" );
186
188
Assert .notNull (timeout , "Timeout must not be null" );
187
189
188
- return createMono (listCommands ->
189
- listCommands . bLMove ( rawKey ( sourceKey ), rawKey ( destinationKey ), from , to , timeout ). map (this ::readValue ));
190
+ return createMono (connection -> connection . bLMove ( rawKey ( sourceKey ), rawKey ( destinationKey ), from , to , timeout )
191
+ . map (this ::readRequiredValue ));
190
192
}
191
193
192
194
@ Override
@@ -211,7 +213,7 @@ public Mono<V> index(K key, long index) {
211
213
212
214
Assert .notNull (key , "Key must not be null" );
213
215
214
- return createMono (listCommands -> listCommands .lIndex (rawKey (key ), index ).map (this ::readValue ));
216
+ return createMono (connection -> connection .lIndex (rawKey (key ), index ).map (this ::readRequiredValue ));
215
217
}
216
218
217
219
@ Override
@@ -236,7 +238,7 @@ public Mono<V> leftPop(K key) {
236
238
237
239
Assert .notNull (key , "Key must not be null" );
238
240
239
- return createMono (listCommands -> listCommands .lPop (rawKey (key )).map (this ::readValue ));
241
+ return createMono (connection -> connection .lPop (rawKey (key )).map (this ::readRequiredValue ));
240
242
241
243
}
242
244
@@ -245,7 +247,7 @@ public Flux<V> leftPop(K key, long count) {
245
247
246
248
Assert .notNull (key , "Key must not be null" );
247
249
248
- return createFlux (listCommands -> listCommands .lPop (rawKey (key ), count ).map (this ::readValue ));
250
+ return createFlux (listCommands -> listCommands .lPop (rawKey (key ), count ).map (this ::readRequiredValue ));
249
251
}
250
252
251
253
@ Override
@@ -255,25 +257,24 @@ public Mono<V> leftPop(K key, Duration timeout) {
255
257
Assert .notNull (timeout , "Duration must not be null" );
256
258
Assert .isTrue (isZeroOrGreaterOneSecond (timeout ), "Duration must be either zero or greater or equal to 1 second" );
257
259
258
- return createMono (listCommands ->
259
- listCommands .blPop (Collections .singletonList (rawKey (key )), timeout )
260
- .map (popResult -> readValue (popResult .getValue ())));
260
+ return createMono (connection -> connection .blPop (Collections .singletonList (rawKey (key )), timeout )
261
+ .mapNotNull (popResult -> readValue (popResult .getValue ())));
261
262
}
262
263
263
264
@ Override
264
265
public Mono <V > rightPop (K key ) {
265
266
266
267
Assert .notNull (key , "Key must not be null" );
267
268
268
- return createMono (listCommands -> listCommands .rPop (rawKey (key )).map (this ::readValue ));
269
+ return createMono (listCommands -> listCommands .rPop (rawKey (key )).map (this ::readRequiredValue ));
269
270
}
270
271
271
272
@ Override
272
273
public Flux <V > rightPop (K key , long count ) {
273
274
274
275
Assert .notNull (key , "Key must not be null" );
275
276
276
- return createFlux (listCommands -> listCommands .rPop (rawKey (key ), count ).map (this ::readValue ));
277
+ return createFlux (listCommands -> listCommands .rPop (rawKey (key ), count ).map (this ::readRequiredValue ));
277
278
}
278
279
279
280
@ Override
@@ -283,9 +284,8 @@ public Mono<V> rightPop(K key, Duration timeout) {
283
284
Assert .notNull (timeout , "Duration must not be null" );
284
285
Assert .isTrue (isZeroOrGreaterOneSecond (timeout ), "Duration must be either zero or greater or equal to 1 second" );
285
286
286
- return createMono (listCommands ->
287
- listCommands .brPop (Collections .singletonList (rawKey (key )), timeout )
288
- .map (popResult -> readValue (popResult .getValue ())));
287
+ return createMono (connection -> connection .brPop (Collections .singletonList (rawKey (key )), timeout )
288
+ .mapNotNull (popResult -> readValue (popResult .getValue ())));
289
289
}
290
290
291
291
@ Override
@@ -294,8 +294,8 @@ public Mono<V> rightPopAndLeftPush(K sourceKey, K destinationKey) {
294
294
Assert .notNull (sourceKey , "Source key must not be null" );
295
295
Assert .notNull (destinationKey , "Destination key must not be null" );
296
296
297
- return createMono (listCommands ->
298
- listCommands . rPopLPush ( rawKey ( sourceKey ), rawKey ( destinationKey )). map (this ::readValue ));
297
+ return createMono (connection -> connection . rPopLPush ( rawKey ( sourceKey ), rawKey ( destinationKey ))
298
+ . map (this ::readRequiredValue ));
299
299
}
300
300
301
301
@ Override
@@ -306,8 +306,8 @@ public Mono<V> rightPopAndLeftPush(K sourceKey, K destinationKey, Duration timeo
306
306
Assert .notNull (timeout , "Duration must not be null" );
307
307
Assert .isTrue (isZeroOrGreaterOneSecond (timeout ), "Duration must be either zero or greater or equal to 1 second" );
308
308
309
- return createMono (listCommands ->
310
- listCommands . bRPopLPush ( rawKey ( sourceKey ), rawKey ( destinationKey ), timeout ). map (this ::readValue ));
309
+ return createMono (connection -> connection . bRPopLPush ( rawKey ( sourceKey ), rawKey ( destinationKey ), timeout )
310
+ . map (this ::readRequiredValue ));
311
311
}
312
312
313
313
@ Override
@@ -344,7 +344,19 @@ private ByteBuffer rawValue(V value) {
344
344
return serializationContext .getValueSerializationPair ().write (value );
345
345
}
346
346
347
+ @ Nullable
347
348
private V readValue (ByteBuffer buffer ) {
348
349
return serializationContext .getValueSerializationPair ().read (buffer );
349
350
}
351
+
352
+ private V readRequiredValue (ByteBuffer buffer ) {
353
+
354
+ V v = readValue (buffer );
355
+
356
+ if (v == null ) {
357
+ throw new InvalidDataAccessApiUsageException ("Deserialized list value is null" );
358
+ }
359
+
360
+ return v ;
361
+ }
350
362
}
0 commit comments