Skip to content

Commit b5f124c

Browse files
mp911dejxblum
mp911de
authored and
jxblum
committed
Terminate stream with error on null values returned by RedisElementReader for top-level elements.
We now emit InvalidDataAccessApiUsageException when a RedisElementReader returns null in the context of a top-level stream to indicate invalid API usage although RedisElementReader.read can generally return null values if these are being collected in a container or value wrapper or parent complex object. Apply consistent wording to operations documentation.
1 parent 66b00e2 commit b5f124c

22 files changed

+276
-99
lines changed

‎src/main/java/org/springframework/data/redis/connection/convert/Converters.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -469,7 +469,7 @@ public static Object parse(Object source, String sourcePath, Map<String, Class<?
469469
* @return
470470
* @since 2.6
471471
*/
472-
public static <K, V> Map.Entry<K, V> entryOf(K key, V value) {
472+
public static <K, V> Map.Entry<K, V> entryOf(@Nullable K key, @Nullable V value) {
473473
return new AbstractMap.SimpleImmutableEntry<>(key, value);
474474
}
475475

‎src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import org.springframework.data.redis.domain.geo.GeoReference.GeoMemberReference;
4040
import org.springframework.data.redis.domain.geo.GeoShape;
4141
import org.springframework.data.redis.serializer.RedisSerializationContext;
42+
import org.springframework.lang.Nullable;
4243
import org.springframework.util.Assert;
4344

4445
/**
@@ -320,6 +321,7 @@ private ByteBuffer rawValue(V value) {
320321
return serializationContext.getValueSerializationPair().write(value);
321322
}
322323

324+
@Nullable
323325
private V readValue(ByteBuffer buffer) {
324326
return serializationContext.getValueSerializationPair().read(buffer);
325327
}

‎src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java‎

Lines changed: 36 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,11 @@
2626
import java.util.function.Function;
2727

2828
import org.reactivestreams.Publisher;
29+
import org.springframework.dao.InvalidDataAccessApiUsageException;
2930
import org.springframework.data.redis.connection.ReactiveHashCommands;
3031
import org.springframework.data.redis.connection.convert.Converters;
3132
import org.springframework.data.redis.serializer.RedisSerializationContext;
33+
import org.springframework.lang.Nullable;
3234
import org.springframework.util.Assert;
3335

3436
/**
@@ -126,7 +128,8 @@ public Mono<HK> randomKey(H key) {
126128

127129
Assert.notNull(key, "Key must not be null");
128130

129-
return createMono(hashCommands -> hashCommands.hRandField(rawKey(key))).map(this::readHashKey);
131+
return template.doCreateMono(connection -> connection //
132+
.hashCommands().hRandField(rawKey(key))).map(this::readRequiredHashKey);
130133
}
131134

132135
@Override
@@ -142,7 +145,8 @@ public Flux<HK> randomKeys(H key, long count) {
142145

143146
Assert.notNull(key, "Key must not be null");
144147

145-
return createFlux(hashCommands -> hashCommands.hRandField(rawKey(key), count)).map(this::readHashKey);
148+
return template.doCreateFlux(connection -> connection //
149+
.hashCommands().hRandField(rawKey(key), count)).map(this::readRequiredHashKey);
146150
}
147151

148152
@Override
@@ -159,8 +163,8 @@ public Flux<HK> keys(H key) {
159163

160164
Assert.notNull(key, "Key must not be null");
161165

162-
return createFlux(hashCommands -> hashCommands.hKeys(rawKey(key)) //
163-
.map(this::readHashKey));
166+
return createFlux(connection -> connection.hKeys(rawKey(key)) //
167+
.map(this::readRequiredHashKey));
164168
}
165169

166170
@Override
@@ -207,8 +211,8 @@ public Flux<HV> values(H key) {
207211

208212
Assert.notNull(key, "Key must not be null");
209213

210-
return createFlux(hashCommands -> hashCommands.hVals(rawKey(key)) //
211-
.map(this::readHashValue));
214+
return createFlux(connection -> connection.hVals(rawKey(key)) //
215+
.map(this::readRequiredHashValue));
212216
}
213217

214218
@Override
@@ -265,13 +269,37 @@ private ByteBuffer rawHashValue(HV key) {
265269
}
266270

267271
@SuppressWarnings("unchecked")
272+
@Nullable
268273
private HK readHashKey(ByteBuffer value) {
269274
return (HK) serializationContext.getHashKeySerializationPair().read(value);
270275
}
271276

277+
private HK readRequiredHashKey(ByteBuffer buffer) {
278+
279+
HK hashKey = readHashKey(buffer);
280+
281+
if (hashKey == null) {
282+
throw new InvalidDataAccessApiUsageException("Deserialized hash key is null");
283+
}
284+
285+
return hashKey;
286+
}
287+
272288
@SuppressWarnings("unchecked")
273-
private HV readHashValue(ByteBuffer value) {
274-
return (HV) (value == null ? value : serializationContext.getHashValueSerializationPair().read(value));
289+
@Nullable
290+
private HV readHashValue(@Nullable ByteBuffer value) {
291+
return (HV) (value == null ? null : serializationContext.getHashValueSerializationPair().read(value));
292+
}
293+
294+
private HV readRequiredHashValue(ByteBuffer buffer) {
295+
296+
HV hashValue = readHashValue(buffer);
297+
298+
if (hashValue == null) {
299+
throw new InvalidDataAccessApiUsageException("Deserialized hash value is null");
300+
}
301+
302+
return hashValue;
275303
}
276304

277305
private Map.Entry<HK, HV> deserializeHashEntry(Map.Entry<ByteBuffer, ByteBuffer> source) {

‎src/main/java/org/springframework/data/redis/core/DefaultReactiveListOperations.java‎

Lines changed: 32 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -27,11 +27,13 @@
2727
import java.util.function.Function;
2828

2929
import org.reactivestreams.Publisher;
30+
import org.springframework.dao.InvalidDataAccessApiUsageException;
3031
import org.springframework.data.redis.connection.ReactiveListCommands;
3132
import org.springframework.data.redis.connection.ReactiveListCommands.Direction;
3233
import org.springframework.data.redis.connection.ReactiveListCommands.LPosCommand;
3334
import org.springframework.data.redis.connection.RedisListCommands.Position;
3435
import org.springframework.data.redis.serializer.RedisSerializationContext;
36+
import org.springframework.lang.Nullable;
3537
import org.springframework.util.Assert;
3638

3739
/**
@@ -59,7 +61,7 @@ public Flux<V> range(K key, long start, long end) {
5961

6062
Assert.notNull(key, "Key must not be null");
6163

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));
6365
}
6466

6567
@Override
@@ -172,8 +174,8 @@ public Mono<V> move(K sourceKey, Direction from, K destinationKey, Direction to)
172174
Assert.notNull(from, "From direction must not be null");
173175
Assert.notNull(to, "To direction must not be null");
174176

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));
177179
}
178180

179181
@Override
@@ -185,8 +187,8 @@ public Mono<V> move(K sourceKey, Direction from, K destinationKey, Direction to,
185187
Assert.notNull(to, "To direction must not be null");
186188
Assert.notNull(timeout, "Timeout must not be null");
187189

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));
190192
}
191193

192194
@Override
@@ -211,7 +213,7 @@ public Mono<V> index(K key, long index) {
211213

212214
Assert.notNull(key, "Key must not be null");
213215

214-
return createMono(listCommands -> listCommands.lIndex(rawKey(key), index).map(this::readValue));
216+
return createMono(connection -> connection.lIndex(rawKey(key), index).map(this::readRequiredValue));
215217
}
216218

217219
@Override
@@ -236,7 +238,7 @@ public Mono<V> leftPop(K key) {
236238

237239
Assert.notNull(key, "Key must not be null");
238240

239-
return createMono(listCommands -> listCommands.lPop(rawKey(key)).map(this::readValue));
241+
return createMono(connection -> connection.lPop(rawKey(key)).map(this::readRequiredValue));
240242

241243
}
242244

@@ -245,7 +247,7 @@ public Flux<V> leftPop(K key, long count) {
245247

246248
Assert.notNull(key, "Key must not be null");
247249

248-
return createFlux(listCommands -> listCommands.lPop(rawKey(key), count).map(this::readValue));
250+
return createFlux(listCommands -> listCommands.lPop(rawKey(key), count).map(this::readRequiredValue));
249251
}
250252

251253
@Override
@@ -255,25 +257,24 @@ public Mono<V> leftPop(K key, Duration timeout) {
255257
Assert.notNull(timeout, "Duration must not be null");
256258
Assert.isTrue(isZeroOrGreaterOneSecond(timeout), "Duration must be either zero or greater or equal to 1 second");
257259

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())));
261262
}
262263

263264
@Override
264265
public Mono<V> rightPop(K key) {
265266

266267
Assert.notNull(key, "Key must not be null");
267268

268-
return createMono(listCommands -> listCommands.rPop(rawKey(key)).map(this::readValue));
269+
return createMono(listCommands -> listCommands.rPop(rawKey(key)).map(this::readRequiredValue));
269270
}
270271

271272
@Override
272273
public Flux<V> rightPop(K key, long count) {
273274

274275
Assert.notNull(key, "Key must not be null");
275276

276-
return createFlux(listCommands -> listCommands.rPop(rawKey(key), count).map(this::readValue));
277+
return createFlux(listCommands -> listCommands.rPop(rawKey(key), count).map(this::readRequiredValue));
277278
}
278279

279280
@Override
@@ -283,9 +284,8 @@ public Mono<V> rightPop(K key, Duration timeout) {
283284
Assert.notNull(timeout, "Duration must not be null");
284285
Assert.isTrue(isZeroOrGreaterOneSecond(timeout), "Duration must be either zero or greater or equal to 1 second");
285286

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())));
289289
}
290290

291291
@Override
@@ -294,8 +294,8 @@ public Mono<V> rightPopAndLeftPush(K sourceKey, K destinationKey) {
294294
Assert.notNull(sourceKey, "Source key must not be null");
295295
Assert.notNull(destinationKey, "Destination key must not be null");
296296

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));
299299
}
300300

301301
@Override
@@ -306,8 +306,8 @@ public Mono<V> rightPopAndLeftPush(K sourceKey, K destinationKey, Duration timeo
306306
Assert.notNull(timeout, "Duration must not be null");
307307
Assert.isTrue(isZeroOrGreaterOneSecond(timeout), "Duration must be either zero or greater or equal to 1 second");
308308

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));
311311
}
312312

313313
@Override
@@ -344,7 +344,19 @@ private ByteBuffer rawValue(V value) {
344344
return serializationContext.getValueSerializationPair().write(value);
345345
}
346346

347+
@Nullable
347348
private V readValue(ByteBuffer buffer) {
348349
return serializationContext.getValueSerializationPair().read(buffer);
349350
}
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+
}
350362
}

‎src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java‎

Lines changed: 24 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,10 @@
2828
import java.util.function.Function;
2929

3030
import org.reactivestreams.Publisher;
31+
import org.springframework.dao.InvalidDataAccessApiUsageException;
3132
import org.springframework.data.redis.connection.ReactiveSetCommands;
3233
import org.springframework.data.redis.serializer.RedisSerializationContext;
34+
import org.springframework.lang.Nullable;
3335
import org.springframework.util.Assert;
3436

3537
/**
@@ -89,15 +91,15 @@ public Mono<V> pop(K key) {
8991

9092
Assert.notNull(key, "Key must not be null");
9193

92-
return createMono(setCommands -> setCommands.sPop(rawKey(key)).map(this::readValue));
94+
return createMono(setCommands -> setCommands.sPop(rawKey(key)).map(this::readRequiredValue));
9395
}
9496

9597
@Override
9698
public Flux<V> pop(K key, long count) {
9799

98100
Assert.notNull(key, "Key must not be null");
99101

100-
return createFlux(setCommands -> setCommands.sPop(rawKey(key), count).map(this::readValue));
102+
return createFlux(setCommands -> setCommands.sPop(rawKey(key), count).map(this::readRequiredValue));
101103
}
102104

103105
@Override
@@ -175,7 +177,7 @@ public Flux<V> intersect(Collection<K> keys) {
175177
.map(this::rawKey) //
176178
.collectList() //
177179
.flatMapMany(setCommands::sInter) //
178-
.map(this::readValue));
180+
.map(this::readRequiredValue));
179181
}
180182

181183
@Override
@@ -237,7 +239,7 @@ public Flux<V> union(Collection<K> keys) {
237239
.map(this::rawKey) //
238240
.collectList() //
239241
.flatMapMany(setCommands::sUnion) //
240-
.map(this::readValue));
242+
.map(this::readRequiredValue));
241243
}
242244

243245
@Override
@@ -299,7 +301,7 @@ public Flux<V> difference(Collection<K> keys) {
299301
.map(this::rawKey) //
300302
.collectList() //
301303
.flatMapMany(setCommands::sDiff) //
302-
.map(this::readValue));
304+
.map(this::readRequiredValue));
303305
}
304306

305307
@Override
@@ -339,7 +341,7 @@ public Flux<V> members(K key) {
339341

340342
Assert.notNull(key, "Key must not be null");
341343

342-
return createFlux(setCommands -> setCommands.sMembers(rawKey(key)).map(this::readValue));
344+
return createFlux(setCommands -> setCommands.sMembers(rawKey(key)).map(this::readRequiredValue));
343345
}
344346

345347
@Override
@@ -348,31 +350,31 @@ public Flux<V> scan(K key, ScanOptions options) {
348350
Assert.notNull(key, "Key must not be null");
349351
Assert.notNull(options, "ScanOptions must not be null");
350352

351-
return createFlux(setCommands -> setCommands.sScan(rawKey(key), options).map(this::readValue));
353+
return createFlux(setCommands -> setCommands.sScan(rawKey(key), options).map(this::readRequiredValue));
352354
}
353355

354356
@Override
355357
public Mono<V> randomMember(K key) {
356358

357359
Assert.notNull(key, "Key must not be null");
358360

359-
return createMono(setCommands -> setCommands.sRandMember(rawKey(key)).map(this::readValue));
361+
return createMono(setCommands -> setCommands.sRandMember(rawKey(key)).map(this::readRequiredValue));
360362
}
361363

362364
@Override
363365
public Flux<V> distinctRandomMembers(K key, long count) {
364366

365367
Assert.isTrue(count > 0, "Negative count not supported; Use randomMembers to allow duplicate elements");
366368

367-
return createFlux(setCommands -> setCommands.sRandMember(rawKey(key), count).map(this::readValue));
369+
return createFlux(setCommands -> setCommands.sRandMember(rawKey(key), count).map(this::readRequiredValue));
368370
}
369371

370372
@Override
371373
public Flux<V> randomMembers(K key, long count) {
372374

373375
Assert.isTrue(count > 0, "Use a positive number for count; This method is already allowing duplicate elements");
374376

375-
return createFlux(setCommands -> setCommands.sRandMember(rawKey(key), -count).map(this::readValue));
377+
return createFlux(setCommands -> setCommands.sRandMember(rawKey(key), -count).map(this::readRequiredValue));
376378
}
377379

378380
@Override
@@ -415,7 +417,19 @@ private ByteBuffer rawValue(V value) {
415417
return serializationContext.getValueSerializationPair().write(value);
416418
}
417419

420+
@Nullable
418421
private V readValue(ByteBuffer buffer) {
419422
return serializationContext.getValueSerializationPair().read(buffer);
420423
}
424+
425+
private V readRequiredValue(ByteBuffer buffer) {
426+
427+
V v = readValue(buffer);
428+
429+
if (v == null) {
430+
throw new InvalidDataAccessApiUsageException("Deserialized set value is null");
431+
}
432+
433+
return v;
434+
}
421435
}

0 commit comments

Comments
 (0)