Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-602 - Polishing.

Browse Source 
Remove mutable state from `ReactiveRedisTemplate`.
Set up required args constructors for passing `RedisSerializationContext` and `ConnectionFactory` while removing `InitializingBean` and `ClassLoaderAware` interfaces forcing the template to hold mutable state.
Introduced some tiny static helpers for creating `RedisSerializationContext` from Serializers and added defaulting.

Moving forward `RedisTemplate` should also be migrated to using the SerializationContext getting rid of mutable state.

Additionally some minor renames, JavaDoc updates and refactored methods to lambda style.

Original Pull Request: #239
This commit is contained in:
Christoph Strobl
2017-03-23 11:27:18 +01:00
parent 81657c530b
commit 07984271d8
32 changed files with 1347 additions and 1565 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommands.java
View File

@@ -93,6 +93,7 @@ public class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
*/
@Override
public Flux<MultiValueResponse<ByteBuffer, ByteBuffer>> keys(Publisher<ByteBuffer> patterns) {
return connection.execute(cmd -> Flux.from(patterns).flatMap(pattern -> {
Assert.notNull(pattern, "Pattern must not be null!");

147
src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.data.redis.core;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -35,25 +36,31 @@ import org.springframework.data.geo.Point;
import org.springframework.data.redis.connection.ReactiveGeoCommands;
import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs;
import org.springframework.data.redis.serializer.ReactiveSerializationContext;
import org.springframework.util.Assert;
/**
* Default implementation of {@link ReactiveGeoOperations}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V> {
private final ReactiveRedisTemplate<?, ?> template;
private final ReactiveSerializationContext<K, V> serializationContext;
private final RedisSerializationContext<K, V> serializationContext;
/**
* Create new instance of {@link DefaultReactiveGeoOperations}.
*
* @param template must not be {@literal null}.
* @param serializationContext must not be {@literal null}.
*/
public DefaultReactiveGeoOperations(ReactiveRedisTemplate<?, ?> template,
ReactiveSerializationContext<K, V> serializationContext) {
RedisSerializationContext<K, V> serializationContext) {
Assert.notNull(template, "ReactiveRedisTemplate must not be null!");
Assert.notNull(serializationContext, "ReactiveSerializationContext must not be null!");
Assert.notNull(serializationContext, "RedisSerializationContext must not be null!");
this.template = template;
this.serializationContext = serializationContext;
@@ -92,7 +99,7 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
public Mono<Long> geoAdd(K key, Map<V, Point> memberCoordinateMap) {
Assert.notNull(key, "Key must not be null!");
Assert.notNull(memberCoordinateMap, "Map must not be null!");
Assert.notNull(memberCoordinateMap, "MemberCoordinateMap must not be null!");
return createMono(connection -> {
@@ -129,16 +136,13 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
public Flux<Long> geoAdd(K key, Publisher<? extends Collection<GeoLocation<V>>> locations) {
Assert.notNull(key, "Key must not be null!");
Assert.notNull(locations, "GeoLocations must not be null!");
Assert.notNull(locations, "Locations must not be null!");
return createFlux(connection -> {
return Flux.from(locations)
.map(locationList -> locationList.stream()
.map(location -> new GeoLocation<>(rawValue(location.getName()), location.getPoint()))
.collect(Collectors.toList()))
.flatMap(list -> connection.geoAdd(rawKey(key), list));
});
return createFlux(connection -> Flux.from(locations)
.map(locationList -> locationList.stream()
.map(location -> new GeoLocation<>(rawValue(location.getName()), location.getPoint()))
.collect(Collectors.toList()))
.flatMap(list -> connection.geoAdd(rawKey(key), list)));
}
/* (non-Javadoc)
@@ -188,17 +192,13 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
public final Mono<List<String>> geoHash(K key, V... members) {
Assert.notNull(key, "Key must not be null!");
Assert.notNull(members, "Member must not be null!");
Assert.notEmpty(members, "Members must not be empty!");
Assert.notEmpty(members, "Members must not be null or empty!");
Assert.noNullElements(members, "Members must not contain null elements!");
return createMono(connection -> {
return Flux.fromArray(members) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.geoHash(rawKey(key), serialized));
});
return createMono(connection -> Flux.fromArray(members) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.geoHash(rawKey(key), serialized)));
}
/* (non-Javadoc)
@@ -221,17 +221,13 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
public final Mono<List<Point>> geoPos(K key, V... members) {
Assert.notNull(key, "Key must not be null!");
Assert.notNull(members, "Member must not be null!");
Assert.notEmpty(members, "Members must not be empty!");
Assert.notEmpty(members, "Members must not be null or empty!");
Assert.noNullElements(members, "Members must not contain null elements!");
return createMono(connection -> {
return Flux.fromArray(members) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.geoPos(rawKey(key), serialized));
});
return createMono(connection -> Flux.fromArray(members) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.geoPos(rawKey(key), serialized)));
}
/* (non-Javadoc)
@@ -243,13 +239,10 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
Assert.notNull(key, "Key must not be null!");
Assert.notNull(within, "Circle must not be null!");
return createMono(connection -> {
return connection.geoRadius(rawKey(key), within) //
.flatMap(Flux::fromIterable) //
.map(location -> new GeoLocation<>(readValue(location.getName()), location.getPoint())) //
.collectList();
});
return createMono(connection -> connection.geoRadius(rawKey(key), within) //
.flatMap(Flux::fromIterable) //
.map(location -> new GeoLocation<>(readValue(location.getName()), location.getPoint())) //
.collectList());
}
/* (non-Javadoc)
@@ -262,16 +255,13 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
Assert.notNull(within, "Circle must not be null!");
Assert.notNull(args, "GeoRadiusCommandArgs must not be null!");
return createMono(connection -> {
return connection.geoRadius(rawKey(key), within, args) //
.flatMap(Flux::fromIterable) //
.map(geoResult -> new GeoResult<>(
new GeoLocation<>(readValue(geoResult.getContent().getName()), geoResult.getContent().getPoint()),
geoResult.getDistance())) //
.collectList() //
.map(GeoResults::new);
});
return createMono(connection -> connection.geoRadius(rawKey(key), within, args) //
.flatMap(Flux::fromIterable) //
.map(geoResult -> new GeoResult<>(
new GeoLocation<>(readValue(geoResult.getContent().getName()), geoResult.getContent().getPoint()),
geoResult.getDistance())) //
.collectList() //
.map(GeoResults::new));
}
/* (non-Javadoc)
@@ -283,13 +273,10 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
Assert.notNull(key, "Key must not be null!");
Assert.notNull(member, "Member must not be null!");
return createMono(connection -> {
return connection.geoRadiusByMember(rawKey(key), rawValue(member), new Distance(radius)) //
.flatMap(Flux::fromIterable) //
.map(geoLocation -> new GeoLocation<>(readValue(geoLocation.getName()), geoLocation.getPoint())) //
.collectList();
});
return createMono(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), new Distance(radius)) //
.flatMap(Flux::fromIterable) //
.map(geoLocation -> new GeoLocation<>(readValue(geoLocation.getName()), geoLocation.getPoint())) //
.collectList());
}
/* (non-Javadoc)
@@ -302,13 +289,10 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
Assert.notNull(member, "Member must not be null!");
Assert.notNull(distance, "Distance must not be null!");
return createMono(connection -> {
return connection.geoRadiusByMember(rawKey(key), rawValue(member), distance) //
.flatMap(Flux::fromIterable) //
.map(geoLocation -> new GeoLocation<>(readValue(geoLocation.getName()), geoLocation.getPoint())) //
.collectList();
});
return createMono(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), distance) //
.flatMap(Flux::fromIterable) //
.map(geoLocation -> new GeoLocation<>(readValue(geoLocation.getName()), geoLocation.getPoint())) //
.collectList());
}
/* (non-Javadoc)
@@ -323,16 +307,13 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
Assert.notNull(distance, "Distance must not be null!");
Assert.notNull(args, "GeoRadiusCommandArgs must not be null!");
return createMono(connection -> {
return connection.geoRadiusByMember(rawKey(key), rawValue(member), distance, args) //
.flatMap(Flux::fromIterable) //
.map(geoResult -> new GeoResult<>(
new GeoLocation<>(readValue(geoResult.getContent().getName()), geoResult.getContent().getPoint()),
geoResult.getDistance())) //
.collectList() //
.map(GeoResults::new);
});
return createMono(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), distance, args) //
.flatMap(Flux::fromIterable) //
.map(geoResult -> new GeoResult<>(
new GeoLocation<>(readValue(geoResult.getContent().getName()), geoResult.getContent().getPoint()),
geoResult.getDistance())) //
.collectList() //
.map(GeoResults::new));
}
/* (non-Javadoc)
@@ -343,17 +324,13 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
public final Mono<Long> geoRemove(K key, V... members) {
Assert.notNull(key, "Key must not be null!");
Assert.notNull(members, "Member must not be null!");
Assert.notEmpty(members, "Members must not be empty!");
Assert.notEmpty(members, "Members must not be null or empty!");
Assert.noNullElements(members, "Members must not contain null elements!");
return template.createMono(connection -> {
return Flux.fromArray(members) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.zSetCommands().zRem(rawKey(key), serialized));
});
return template.createMono(connection -> Flux.fromArray(members) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.zSetCommands().zRem(rawKey(key), serialized)));
}
/* (non-Javadoc)
@@ -382,14 +359,14 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
}
private ByteBuffer rawKey(K key) {
return serializationContext.key().write(key);
return serializationContext.getKeySerializationPair().write(key);
}
private ByteBuffer rawValue(V value) {
return serializationContext.value().write(value);
return serializationContext.getValueSerializationPair().write(value);
}
private V readValue(ByteBuffer buffer) {
return serializationContext.value().read(buffer);
return serializationContext.getValueSerializationPair().read(buffer);
}
}

95
src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java
View File

@@ -28,25 +28,32 @@ import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.ReactiveHashCommands;
import org.springframework.data.redis.serializer.ReactiveSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.util.Assert;
/**
* Default implementation of {@link ReactiveHashOperations}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
public class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations<H, HK, HV> {
private final ReactiveRedisTemplate<?, ?> template;
private final ReactiveSerializationContext<H, ?> serializationContext;
private final RedisSerializationContext<H, ?> serializationContext;
/**
* Creates new instance of {@link DefaultReactiveHashOperations}.
*
* @param template must not be {@literal null}.
* @param serializationContext must not be {@literal null}.
*/
public DefaultReactiveHashOperations(ReactiveRedisTemplate<?, ?> template,
ReactiveSerializationContext<H, ?> serializationContext) {
RedisSerializationContext<H, ?> serializationContext) {
Assert.notNull(template, "ReactiveRedisTemplate must not be null!");
Assert.notNull(serializationContext, "ReactiveSerializationContext must not be null!");
Assert.notNull(serializationContext, "RedisSerializationContext must not be null!");
this.template = template;
this.serializationContext = serializationContext;
@@ -64,13 +71,10 @@ public class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOpe
Assert.notEmpty(hashKeys, "Hash keys must not be empty!");
Assert.noNullElements(hashKeys, "Hash keys must not contain null elements!");
return createMono(connection -> {
return Flux.fromArray(hashKeys) //
.map(o -> (HK) o).map(this::rawHashKey) //
.collectList() //
.then(hks -> connection.hDel(rawKey(key), hks));
});
return createMono(connection -> Flux.fromArray(hashKeys) //
.map(o -> (HK) o).map(this::rawHashKey) //
.collectList() //
.then(hks -> connection.hDel(rawKey(key), hks)));
}
/* (non-Javadoc)
@@ -109,20 +113,10 @@ public class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOpe
Assert.notNull(hashKeys, "Hash keys must not be null!");
Assert.notEmpty(hashKeys, "Hash keys must not be empty!");
return createMono(connection -> {
return Flux.fromIterable(hashKeys) //
.map(this::rawHashKey) //
.collectList() //
.then(hks -> connection.hMGet(rawKey(key), hks)).map(byteBuffers -> {
List<HV> values = new ArrayList<HV>(byteBuffers.size());
for (ByteBuffer byteBuffer : byteBuffers) {
values.add(readHashValue(byteBuffer));
}
return values;
});
});
return createMono(connection -> Flux.fromIterable(hashKeys) //
.map(this::rawHashKey) //
.collectList() //
.then(hks -> connection.hMGet(rawKey(key), hks)).map(this::deserializeHashValues));
}
/* (non-Javadoc)
@@ -187,13 +181,9 @@ public class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOpe
Assert.notNull(key, "Key must not be null!");
Assert.notNull(map, "Map must not be null!");
return createMono(connection -> {
return Flux.fromIterable(() -> map.entrySet().iterator()) //
.collectMap(entry -> rawHashKey(entry.getKey()), entry -> rawHashValue(entry.getValue())) //
.flatMap(serialized -> connection.hMSet(rawKey(key), serialized));
});
return createMono(connection -> Flux.fromIterable(() -> map.entrySet().iterator()) //
.collectMap(entry -> rawHashKey(entry.getKey()), entry -> rawHashValue(entry.getValue())) //
.flatMap(serialized -> connection.hMSet(rawKey(key), serialized)));
}
/* (non-Javadoc)
@@ -245,16 +235,7 @@ public class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOpe
Assert.notNull(key, "Key must not be null!");
return createMono(connection -> connection.hGetAll(rawKey(key)) //
.map(map -> {
Map<HK, HV> deserialized = new LinkedHashMap<>(map.size());
map.forEach((k, v) -> {
deserialized.put(readHashKey(k), readHashValue(v));
});
return deserialized;
}));
.map(this::deserializeHashEntries));
}
/* (non-Javadoc)
@@ -276,24 +257,44 @@ public class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOpe
}
private ByteBuffer rawKey(H key) {
return serializationContext.key().write(key);
return serializationContext.getKeySerializationPair().write(key);
}
private ByteBuffer rawHashKey(HK key) {
return serializationContext.hashKey().write(key);
return serializationContext.getHashKeySerializationPair().write(key);
}
private ByteBuffer rawHashValue(HV key) {
return serializationContext.hashValue().write(key);
return serializationContext.getHashValueSerializationPair().write(key);
}
@SuppressWarnings("unchecked")
private HK readHashKey(ByteBuffer value) {
return (HK) serializationContext.hashKey().read(value);
return (HK) serializationContext.getHashKeySerializationPair().read(value);
}
@SuppressWarnings("unchecked")
private HV readHashValue(ByteBuffer value) {
return (HV) serializationContext.hashValue().read(value);
return (HV) serializationContext.getHashValueSerializationPair().read(value);
}
private Map<HK, HV> deserializeHashEntries(Map<ByteBuffer, ByteBuffer> source) {
Map<HK, HV> deserialized = new LinkedHashMap<>(source.size());
source.forEach((k, v) -> {
deserialized.put(readHashKey(k), readHashValue(v));
});
return deserialized;
}
private List<HV> deserializeHashValues(List<ByteBuffer> source) {
List<HV> values = new ArrayList<HV>(source.size());
for (ByteBuffer byteBuffer : source) {
values.add(readHashValue(byteBuffer));
}
return values;
}
}

61
src/main/java/org/springframework/data/redis/core/DefaultReactiveHyperLogLogOperations.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.data.redis.core;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -23,25 +24,31 @@ import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.ReactiveHyperLogLogCommands;
import org.springframework.data.redis.serializer.ReactiveSerializationContext;
import org.springframework.util.Assert;
/**
* Default implementation of {@link ReactiveHyperLogLogOperations}.
*
* @author Mark Paluch
* @auhtor Christoph Strobl
* @since 2.0
*/
public class DefaultReactiveHyperLogLogOperations<K, V> implements ReactiveHyperLogLogOperations<K, V> {
private final ReactiveRedisTemplate<?, ?> template;
private final ReactiveSerializationContext<K, V> serializationContext;
private final RedisSerializationContext<K, V> serializationContext;
/**
* Creates new instance of {@link DefaultReactiveHyperLogLogOperations}.
*
* @param template must not be {@literal null}.
* @param serializationContext must not be {@literal null}.
*/
public DefaultReactiveHyperLogLogOperations(ReactiveRedisTemplate<?, ?> template,
ReactiveSerializationContext<K, V> serializationContext) {
RedisSerializationContext<K, V> serializationContext) {
Assert.notNull(template, "ReactiveRedisTemplate must not be null!");
Assert.notNull(serializationContext, "ReactiveSerializationContext must not be null!");
Assert.notNull(serializationContext, "RedisSerializationContext must not be null!");
this.template = template;
this.serializationContext = serializationContext;
@@ -55,17 +62,13 @@ public class DefaultReactiveHyperLogLogOperations<K, V> implements ReactiveHyper
public final Mono<Long> add(K key, V... values) {
Assert.notNull(key, "Key must not be null!");
Assert.notNull(values, "Values must not be null!");
Assert.notEmpty(values, "Values must not be empty!");
Assert.notEmpty(values, "Values must not be null or empty!");
Assert.noNullElements(values, "Values must not contain null elements!");
return createMono(connection -> {
return Flux.fromArray(values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serializedValues -> connection.pfAdd(rawKey(key), serializedValues));
});
return createMono(connection -> Flux.fromArray(values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serializedValues -> connection.pfAdd(rawKey(key), serializedValues)));
}
/* (non-Javadoc)
@@ -75,17 +78,13 @@ public class DefaultReactiveHyperLogLogOperations<K, V> implements ReactiveHyper
@SafeVarargs
public final Mono<Long> size(K... keys) {
Assert.notNull(keys, "Keys must not be null!");
Assert.notEmpty(keys, "Keys must not be empty!");
Assert.notEmpty(keys, "Keys must not be null or empty!");
Assert.noNullElements(keys, "Keys must not contain null elements!");
return createMono(connection -> {
return Flux.fromArray(keys) //
.map(this::rawKey) //
.collectList() //
.flatMap(connection::pfCount);
});
return createMono(connection -> Flux.fromArray(keys) //
.map(this::rawKey) //
.collectList() //
.flatMap(connection::pfCount));
}
/* (non-Javadoc)
@@ -96,17 +95,13 @@ public class DefaultReactiveHyperLogLogOperations<K, V> implements ReactiveHyper
public final Mono<Boolean> union(K destination, K... sourceKeys) {
Assert.notNull(destination, "Destination key must not be null!");
Assert.notNull(sourceKeys, "Source keys must not be null!");
Assert.notEmpty(sourceKeys, "Source keys must not be empty!");
Assert.notEmpty(sourceKeys, "Source keys must not be null or empty!");
Assert.noNullElements(sourceKeys, "Source keys must not contain null elements!");
return createMono(connection -> {
return Flux.fromArray(sourceKeys) //
.map(this::rawKey) //
.collectList() //
.flatMap(serialized -> connection.pfMerge(rawKey(destination), serialized));
});
return createMono(connection -> Flux.fromArray(sourceKeys) //
.map(this::rawKey) //
.collectList() //
.flatMap(serialized -> connection.pfMerge(rawKey(destination), serialized)));
}
/* (non-Javadoc)
@@ -128,10 +123,10 @@ public class DefaultReactiveHyperLogLogOperations<K, V> implements ReactiveHyper
}
private ByteBuffer rawKey(K key) {
return serializationContext.key().write(key);
return serializationContext.getKeySerializationPair().write(key);
}
private ByteBuffer rawValue(V value) {
return serializationContext.value().write(value);
return serializationContext.getValueSerializationPair().write(value);
}
}

79
src/main/java/org/springframework/data/redis/core/DefaultReactiveListOperations.java
View File

@@ -31,25 +31,32 @@ import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.ReactiveListCommands;
import org.springframework.data.redis.connection.RedisListCommands.Position;
import org.springframework.data.redis.serializer.ReactiveSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.util.Assert;
/**
* Default implementation of {@link ReactiveListOperations}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
public class DefaultReactiveListOperations<K, V> implements ReactiveListOperations<K, V> {
private final ReactiveRedisTemplate<?, ?> template;
private final ReactiveSerializationContext<K, V> serializationContext;
private final RedisSerializationContext<K, V> serializationContext;
/**
* Create new instance of {@link DefaultReactiveListOperations}.
*
* @param template must not be {@literal null}.
* @param serializationContext must not be {@literal null}.
*/
public DefaultReactiveListOperations(ReactiveRedisTemplate<?, ?> template,
ReactiveSerializationContext<K, V> serializationContext) {
RedisSerializationContext<K, V> serializationContext) {
Assert.notNull(template, "ReactiveRedisTemplate must not be null!");
Assert.notNull(serializationContext, "ReactiveSerializationContext must not be null!");
Assert.notNull(serializationContext, "RedisSerializationContext must not be null!");
this.template = template;
this.serializationContext = serializationContext;
@@ -63,16 +70,7 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
Assert.notNull(key, "Key must not be null!");
return createMono(connection -> connection.lRange(rawKey(key), start, end).map(raw -> {
List<V> result = new ArrayList<V>(raw.size());
for (ByteBuffer buffer : raw) {
result.add(readValue(buffer));
}
return result;
}));
return createMono(connection -> connection.lRange(rawKey(key), start, end).map(this::deserializeValues));
}
/* (non-Javadoc)
@@ -112,7 +110,7 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
@SafeVarargs
public final Mono<Long> leftPushAll(K key, V... values) {
Assert.notNull(values, "Values must not be null!");
Assert.notEmpty(values, "Values must not be null or empty!");
return leftPushAll(key, Arrays.asList(values));
}
@@ -124,16 +122,12 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
public Mono<Long> leftPushAll(K key, Collection<V> values) {
Assert.notNull(key, "Key must not be null!");
Assert.notNull(values, "Values must not be null!");
Assert.notEmpty(values, "Values must not be empty!");
Assert.notEmpty(values, "Values must not be null or empty!");
return createMono(connection -> {
return Flux.fromIterable(values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.lPush(rawKey(key), serialized));
});
return createMono(connection -> Flux.fromIterable(values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.lPush(rawKey(key), serialized)));
}
/* (non-Javadoc)
@@ -185,16 +179,12 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
public Mono<Long> rightPushAll(K key, Collection<V> values) {
Assert.notNull(key, "Key must not be null!");
Assert.notNull(values, "Values must not be null!");
Assert.notEmpty(values, "Values must not be empty!");
Assert.notEmpty(values, "Values must not be null or empty!");
return createMono(connection -> {
return Flux.fromIterable(values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.rPush(rawKey(key), serialized));
});
return createMono(connection -> Flux.fromIterable(values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.rPush(rawKey(key), serialized)));
}
/* (non-Javadoc)
@@ -273,7 +263,7 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
Assert.notNull(key, "Key must not be null!");
Assert.notNull(timeout, "Duration must not be null!");
Assert.isTrue(isZeroOrGreater1Second(timeout), "Duration must be either zero or greater or equal to 1 second");
Assert.isTrue(isZeroOrGreater1Second(timeout), "Duration must be either zero or greater or equal to 1 second!");
return createMono(connection -> connection.blPop(Collections.singletonList(rawKey(key)), timeout)
.map(popResult -> readValue(popResult.getValue())));
@@ -298,7 +288,7 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
Assert.notNull(key, "Key must not be null!");
Assert.notNull(timeout, "Duration must not be null!");
Assert.isTrue(isZeroOrGreater1Second(timeout), "Duration must be either zero or greater or equal to 1 second");
Assert.isTrue(isZeroOrGreater1Second(timeout), "Duration must be either zero or greater or equal to 1 second!");
return createMono(connection -> connection.brPop(Collections.singletonList(rawKey(key)), timeout)
.map(popResult -> readValue(popResult.getValue())));
@@ -326,7 +316,7 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
Assert.notNull(sourceKey, "Source key must not be null!");
Assert.notNull(destinationKey, "Destination key must not be null!");
Assert.notNull(timeout, "Duration must not be null!");
Assert.isTrue(isZeroOrGreater1Second(timeout), "Duration must be either zero or greater or equal to 1 second");
Assert.isTrue(isZeroOrGreater1Second(timeout), "Duration must be either zero or greater or equal to 1 second!");
return createMono(
connection -> connection.bRPopLPush(rawKey(sourceKey), rawKey(destinationKey), timeout).map(this::readValue));
@@ -355,14 +345,25 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
}
private ByteBuffer rawKey(K key) {
return serializationContext.key().write(key);
return serializationContext.getKeySerializationPair().write(key);
}
private ByteBuffer rawValue(V value) {
return serializationContext.value().write(value);
return serializationContext.getValueSerializationPair().write(value);
}
private V readValue(ByteBuffer buffer) {
return serializationContext.value().read(buffer);
return serializationContext.getValueSerializationPair().read(buffer);
}
private List<V> deserializeValues(List<ByteBuffer> source) {
List<V> result = new ArrayList<V>(source.size());
for (ByteBuffer buffer : source) {
result.add(readValue(buffer));
}
return result;
}
}

127
src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java
View File

@@ -29,25 +29,32 @@ import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.ReactiveSetCommands;
import org.springframework.data.redis.serializer.ReactiveSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.util.Assert;
/**
* Default implementation of {@link ReactiveSetOperations}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations<K, V> {
private final ReactiveRedisTemplate<?, ?> template;
private final ReactiveSerializationContext<K, V> serializationContext;
private final RedisSerializationContext<K, V> serializationContext;
/**
* Creates new {@link DefaultReactiveSetOperations}.
*
* @param template must not be {@literal null}.
* @param serializationContext must not be {@literal null}.
*/
public DefaultReactiveSetOperations(ReactiveRedisTemplate<?, ?> template,
ReactiveSerializationContext<K, V> serializationContext) {
RedisSerializationContext<K, V> serializationContext) {
Assert.notNull(template, "ReactiveRedisTemplate must not be null!");
Assert.notNull(serializationContext, "ReactiveSerializationContext must not be null!");
Assert.notNull(serializationContext, "RedisSerializationContext must not be null!");
this.template = template;
this.serializationContext = serializationContext;
@@ -65,13 +72,10 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
return createMono(connection -> connection.sAdd(rawKey(key), rawValue(values[0])));
}
return createMono(connection -> {
return Flux.fromArray(values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.sAdd(rawKey(key), serialized));
});
return createMono(connection -> Flux.fromArray(values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.sAdd(rawKey(key), serialized)));
}
/* (non-Javadoc)
@@ -87,13 +91,10 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
return createMono(connection -> connection.sRem(rawKey(key), rawValue((V) values[0])));
}
return createMono(connection -> {
return Flux.fromArray((V[]) values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.sRem(rawKey(key), serialized));
});
return createMono(connection -> Flux.fromArray((V[]) values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.sRem(rawKey(key), serialized)));
}
/* (non-Javadoc)
@@ -163,16 +164,11 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
Assert.notNull(key, "Key must not be null!");
Assert.notNull(otherKeys, "Other keys must not be null!");
List<K> keys = getKeys(key, otherKeys);
return createMono(connection -> {
return Flux.fromIterable(keys) //
.map(this::rawKey) //
.collectList() //
.flatMap(connection::sInter) //
.map(this::readValueSet);
});
return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
.map(this::rawKey) //
.collectList() //
.flatMap(connection::sInter) //
.map(this::readValueSet));
}
/* (non-Javadoc)
@@ -198,15 +194,10 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
Assert.notNull(otherKeys, "Other keys must not be null!");
Assert.notNull(destKey, "Destination key must not be null!");
List<K> keys = getKeys(key, otherKeys);
return createMono(connection -> {
return Flux.fromIterable(keys) //
.map(this::rawKey) //
.collectList() //
.flatMap(rawKeys -> connection.sInterStore(rawKey(destKey), rawKeys));
});
return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
.map(this::rawKey) //
.collectList() //
.flatMap(rawKeys -> connection.sInterStore(rawKey(destKey), rawKeys)));
}
/* (non-Javadoc)
@@ -230,16 +221,11 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
Assert.notNull(key, "Key must not be null!");
Assert.notNull(otherKeys, "Other keys must not be null!");
List<K> keys = getKeys(key, otherKeys);
return createMono(connection -> {
return Flux.fromIterable(keys) //
.map(this::rawKey) //
.collectList() //
.flatMap(connection::sUnion) //
.map(this::readValueSet);
});
return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
.map(this::rawKey) //
.collectList() //
.flatMap(connection::sUnion) //
.map(this::readValueSet));
}
/* (non-Javadoc)
@@ -265,15 +251,10 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
Assert.notNull(otherKeys, "Other keys must not be null!");
Assert.notNull(destKey, "Destination key must not be null!");
List<K> keys = getKeys(key, otherKeys);
return createMono(connection -> {
return Flux.fromIterable(keys) //
.map(this::rawKey) //
.collectList() //
.flatMap(rawKeys -> connection.sUnionStore(rawKey(destKey), rawKeys));
});
return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
.map(this::rawKey) //
.collectList() //
.flatMap(rawKeys -> connection.sUnionStore(rawKey(destKey), rawKeys)));
}
/* (non-Javadoc)
@@ -297,16 +278,11 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
Assert.notNull(key, "Key must not be null!");
Assert.notNull(otherKeys, "Other keys must not be null!");
List<K> keys = getKeys(key, otherKeys);
return createMono(connection -> {
return Flux.fromIterable(keys) //
.map(this::rawKey) //
.collectList() //
.flatMap(connection::sDiff) //
.map(this::readValueSet);
});
return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
.map(this::rawKey) //
.collectList() //
.flatMap(connection::sDiff) //
.map(this::readValueSet));
}
/* (non-Javadoc)
@@ -332,15 +308,10 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
Assert.notNull(otherKeys, "Other keys must not be null!");
Assert.notNull(destKey, "Destination key must not be null!");
List<K> keys = getKeys(key, otherKeys);
return createMono(connection -> {
return Flux.fromIterable(keys) //
.map(this::rawKey) //
.collectList() //
.flatMap(rawKeys -> connection.sDiffStore(rawKey(destKey), rawKeys));
});
return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
.map(this::rawKey) //
.collectList() //
.flatMap(rawKeys -> connection.sDiffStore(rawKey(destKey), rawKeys)));
}
/* (non-Javadoc)
@@ -406,7 +377,7 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
}
private ByteBuffer rawKey(K key) {
return serializationContext.key().write(key);
return serializationContext.getKeySerializationPair().write(key);
}
private List<K> getKeys(K key, Collection<K> otherKeys) {
@@ -420,11 +391,11 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
}
private ByteBuffer rawValue(V value) {
return serializationContext.value().write(value);
return serializationContext.getValueSerializationPair().write(value);
}
private V readValue(ByteBuffer buffer) {
return serializationContext.value().read(buffer);
return serializationContext.getValueSerializationPair().read(buffer);
}
private Set<V> readValueSet(Collection<ByteBuffer> raw) {

71
src/main/java/org/springframework/data/redis/core/DefaultReactiveValueOperations.java
View File

@@ -30,26 +30,33 @@ import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.ReactiveStringCommands;
import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.data.redis.serializer.ReactiveSerializationContext;
import org.springframework.data.redis.serializer.ReactiveSerializationContext.SerializationTuple;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
import org.springframework.util.Assert;
/**
* Default implementation of {@link ReactiveValueOperations}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
public class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K, V> {
private final ReactiveRedisTemplate<?, ?> template;
private final ReactiveSerializationContext<K, V> serializationContext;
private final RedisSerializationContext<K, V> serializationContext;
/**
* Creates new {@link DefaultReactiveValueOperations}.
*
* @param template must not be {@literal null}.
* @param serializationContext must not be {@literal null}.
*/
public DefaultReactiveValueOperations(ReactiveRedisTemplate<?, ?> template,
ReactiveSerializationContext<K, V> serializationContext) {
RedisSerializationContext<K, V> serializationContext) {
Assert.notNull(template, "ReactiveRedisTemplate must not be null!");
Assert.notNull(serializationContext, "ReactiveSerializationContext must not be null!");
Assert.notNull(serializationContext, "RedisSerializationContext must not be null!");
this.template = template;
this.serializationContext = serializationContext;
@@ -170,20 +177,7 @@ public class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperat
Assert.notNull(keys, "Keys must not be null!");
return createMono(connection -> Flux.fromIterable(keys).map(key()::write).collectList().flatMap(connection::mGet)
.map(byteBuffers -> {
List<V> result = new ArrayList<>(byteBuffers.size());
for (ByteBuffer buffer : byteBuffers) {
if (buffer == null) {
result.add(null);
} else {
result.add(readValue(buffer));
}
}
return result;
}));
.map(this::deserializeValues));
}
/* (non-Javadoc)
@@ -195,7 +189,8 @@ public class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperat
Assert.notNull(key, "Key must not be null!");
Assert.notNull(value, "Value must not be null!");
return createMono(connection -> connection.append(rawKey(key), serializationContext.string().write(value)));
return createMono(
connection -> connection.append(rawKey(key), serializationContext.getStringSerializationPair().write(value)));
}
/* (non-Javadoc)
@@ -207,7 +202,7 @@ public class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperat
Assert.notNull(key, "Key must not be null!");
return createMono(connection -> connection.getRange(rawKey(key), start, end) //
.map(string()::read));
.map(stringSerializationPair()::read));
}
/* (non-Javadoc)
@@ -273,26 +268,42 @@ public class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperat
}
private ByteBuffer rawKey(K key) {
return serializationContext.key().write(key);
return serializationContext.getKeySerializationPair().write(key);
}
private ByteBuffer rawValue(V value) {
return serializationContext.value().write(value);
return serializationContext.getValueSerializationPair().write(value);
}
private V readValue(ByteBuffer buffer) {
return serializationContext.value().read(buffer);
return serializationContext.getValueSerializationPair().read(buffer);
}
private SerializationTuple<String> string() {
return serializationContext.string();
private SerializationPair<String> stringSerializationPair() {
return serializationContext.getStringSerializationPair();
}
private SerializationTuple<K> key() {
return serializationContext.key();
private SerializationPair<K> key() {
return serializationContext.getKeySerializationPair();
}
private SerializationTuple<V> value() {
return serializationContext.value();
private SerializationPair<V> value() {
return serializationContext.getValueSerializationPair();
}
private List<V> deserializeValues(List<ByteBuffer> source) {
List<V> result = new ArrayList<>(source.size());
for (ByteBuffer buffer : source) {
if (buffer == null) {
result.add(null);
} else {
result.add(readValue(buffer));
}
}
return result;
}
}

69
src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java
View File

@@ -34,7 +34,7 @@ import org.springframework.data.redis.connection.ReactiveZSetCommands;
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
import org.springframework.data.redis.serializer.ReactiveSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.util.ByteUtils;
import org.springframework.util.Assert;
@@ -42,18 +42,25 @@ import org.springframework.util.Assert;
* Default implementation of {@link ReactiveZSetOperations}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V> {
private final ReactiveRedisTemplate<?, ?> template;
private final ReactiveSerializationContext<K, V> serializationContext;
private final RedisSerializationContext<K, V> serializationContext;
/**
* Creates new {@link DefaultReactiveZSetOperations}.
*
* @param template must not be {@literal null}.
* @param serializationContext must not be {@literal null}.
*/
public DefaultReactiveZSetOperations(ReactiveRedisTemplate<?, ?> template,
ReactiveSerializationContext<K, V> serializationContext) {
RedisSerializationContext<K, V> serializationContext) {
Assert.notNull(template, "ReactiveRedisTemplate must not be null!");
Assert.notNull(serializationContext, "ReactiveSerializationContext must not be null!");
Assert.notNull(serializationContext, "RedisSerializationContext must not be null!");
this.template = template;
this.serializationContext = serializationContext;
@@ -79,13 +86,10 @@ public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperatio
Assert.notNull(key, "Key must not be null!");
Assert.notNull(tuples, "Key must not be null!");
return createMono(connection -> {
return Flux.fromIterable(tuples) //
.map(t -> new DefaultTuple(ByteUtils.getBytes(rawValue(t.getValue())), t.getScore())) //
.collectList() //
.flatMap(serialized -> connection.zAdd(rawKey(key), serialized));
});
return createMono(connection -> Flux.fromIterable(tuples) //
.map(t -> new DefaultTuple(ByteUtils.getBytes(rawValue(t.getValue())), t.getScore())) //
.collectList() //
.flatMap(serialized -> connection.zAdd(rawKey(key), serialized)));
}
/* (non-Javadoc)
@@ -102,13 +106,10 @@ public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperatio
return createMono(connection -> connection.zRem(rawKey(key), rawValue((V) values[0])));
}
return createMono(connection -> {
return Flux.fromArray((V[]) values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.zRem(rawKey(key), serialized));
});
return createMono(connection -> Flux.fromArray((V[]) values) //
.map(this::rawValue) //
.collectList() //
.flatMap(serialized -> connection.zRem(rawKey(key), serialized)));
}
/* (non-Javadoc)
@@ -378,15 +379,10 @@ public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperatio
Assert.notNull(otherKeys, "Other keys must not be null!");
Assert.notNull(destKey, "Destination key must not be null!");
List<K> keys = getKeys(key, otherKeys);
return createMono(connection -> {
return Flux.fromIterable(keys) //
.map(this::rawKey) //
.collectList() //
.flatMap(serialized -> connection.zUnionStore(rawKey(destKey), serialized));
});
return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
.map(this::rawKey) //
.collectList() //
.flatMap(serialized -> connection.zUnionStore(rawKey(destKey), serialized)));
}
/* (non-Javadoc)
@@ -412,15 +408,10 @@ public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperatio
Assert.notNull(otherKeys, "Other keys must not be null!");
Assert.notNull(destKey, "Destination key must not be null!");
List<K> keys = getKeys(key, otherKeys);
return createMono(connection -> {
return Flux.fromIterable(keys) //
.map(this::rawKey) //
.collectList() //
.flatMap(serialized -> connection.zInterStore(rawKey(destKey), serialized));
});
return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
.map(this::rawKey) //
.collectList() //
.flatMap(serialized -> connection.zInterStore(rawKey(destKey), serialized)));
}
/* (non-Javadoc)
@@ -492,7 +483,7 @@ public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperatio
}
private ByteBuffer rawKey(K key) {
return serializationContext.key().write(key);
return serializationContext.getKeySerializationPair().write(key);
}
private List<K> getKeys(K key, Collection<K> otherKeys) {
@@ -506,11 +497,11 @@ public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperatio
}
private ByteBuffer rawValue(V value) {
return serializationContext.value().write(value);
return serializationContext.getValueSerializationPair().write(value);
}
private V readValue(ByteBuffer buffer) {
return serializationContext.value().read(buffer);
return serializationContext.getValueSerializationPair().read(buffer);
}
private Set<V> readValueSet(Collection<ByteBuffer> raw) {

35
src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.data.redis.core;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -23,13 +24,13 @@ import java.time.Instant;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.serializer.ReactiveSerializationContext;
/**
* Interface that specified a basic set of Redis operations, implemented by {@link ReactiveRedisTemplate}. Not often
* used but a useful option for extensibility and testability (as it can be easily mocked or stubbed).
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
public interface ReactiveRedisOperations<K, V> {
@@ -45,7 +46,7 @@ public interface ReactiveRedisOperations<K, V> {
*
* @param <T> return type
* @param action callback object that specifies the Redis action
* @return a result object returned by the action or <tt>null</tt>
* @return a result object returned by the action or {@link Flux#empty()}.
*/
<T> Flux<T> execute(ReactiveRedisCallback<T> action);
@@ -187,12 +188,12 @@ public interface ReactiveRedisOperations<K, V> {
/**
* Returns the operations performed on simple values (or Strings in Redis terminology) given a
* {@link ReactiveSerializationContext}.
* {@link RedisSerializationContext}.
*
* @param serializationContext serializers to be used with the returned operations, must not be {@literal null}.
* @return value operations.
*/
<K, V> ReactiveValueOperations<K, V> opsForValue(ReactiveSerializationContext<K, V> serializationContext);
<K, V> ReactiveValueOperations<K, V> opsForValue(RedisSerializationContext<K, V> serializationContext);
/**
* Returns the operations performed on list values.
@@ -202,12 +203,12 @@ public interface ReactiveRedisOperations<K, V> {
ReactiveListOperations<K, V> opsForList();
/**
* Returns the operations performed on list values given a {@link ReactiveSerializationContext}.
* Returns the operations performed on list values given a {@link RedisSerializationContext}.
*
* @param serializationContext serializers to be used with the returned operations, must not be {@literal null}.
* @return list operations.
*/
<K, V> ReactiveListOperations<K, V> opsForList(ReactiveSerializationContext<K, V> serializationContext);
<K, V> ReactiveListOperations<K, V> opsForList(RedisSerializationContext<K, V> serializationContext);
/**
* Returns the operations performed on set values.
@@ -217,12 +218,12 @@ public interface ReactiveRedisOperations<K, V> {
ReactiveSetOperations<K, V> opsForSet();
/**
* Returns the operations performed on set values given a {@link ReactiveSerializationContext}.
* Returns the operations performed on set values given a {@link RedisSerializationContext}.
*
* @param serializationContext serializers to be used with the returned operations, must not be {@literal null}.
* @return set operations.
*/
<K, V> ReactiveSetOperations<K, V> opsForSet(ReactiveSerializationContext<K, V> serializationContext);
<K, V> ReactiveSetOperations<K, V> opsForSet(RedisSerializationContext<K, V> serializationContext);
/**
* Returns the operations performed on zset values (also known as sorted sets).
@@ -233,12 +234,12 @@ public interface ReactiveRedisOperations<K, V> {
/**
* Returns the operations performed on zset values (also known as sorted sets) given a
* {@link ReactiveSerializationContext}.
* {@link RedisSerializationContext}.
*
* @param serializationContext serializers to be used with the returned operations, must not be {@literal null}.
* @return zset operations.
*/
<K, V> ReactiveZSetOperations<K, V> opsForZSet(ReactiveSerializationContext<K, V> serializationContext);
<K, V> ReactiveZSetOperations<K, V> opsForZSet(RedisSerializationContext<K, V> serializationContext);
/**
* Returns the operations performed on multisets using HyperLogLog.
@@ -248,12 +249,12 @@ public interface ReactiveRedisOperations<K, V> {
ReactiveHyperLogLogOperations<K, V> opsForHyperLogLog();
/**
* Returns the operations performed on multisets using HyperLogLog given a {@link ReactiveSerializationContext}.
* Returns the operations performed on multisets using HyperLogLog given a {@link RedisSerializationContext}.
*
* @param serializationContext serializers to be used with the returned operations, must not be {@literal null}.
* @return never {@literal null}.
*/
<K, V> ReactiveHyperLogLogOperations<K, V> opsForHyperLogLog(ReactiveSerializationContext<K, V> serializationContext);
<K, V> ReactiveHyperLogLogOperations<K, V> opsForHyperLogLog(RedisSerializationContext<K, V> serializationContext);
/**
* Returns the operations performed on hash values.
@@ -265,14 +266,14 @@ public interface ReactiveRedisOperations<K, V> {
<HK, HV> ReactiveHashOperations<K, HK, HV> opsForHash();
/**
* Returns the operations performed on hash values given a {@link ReactiveSerializationContext}.
* Returns the operations performed on hash values given a {@link RedisSerializationContext}.
*
* @param serializationContext serializers to be used with the returned operations, must not be {@literal null}.
* @param <HK> hash key (or field) type.
* @param <HV> hash value type.
* @return hash operations.
*/
<K, HK, HV> ReactiveHashOperations<K, HK, HV> opsForHash(ReactiveSerializationContext<K, ?> serializationContext);
<K, HK, HV> ReactiveHashOperations<K, HK, HV> opsForHash(RedisSerializationContext<K, ?> serializationContext);
/**
* Returns geospatial specific operations interface.
@@ -287,10 +288,10 @@ public interface ReactiveRedisOperations<K, V> {
* @param serializationContext serializers to be used with the returned operations, must not be {@literal null}.
* @return geospatial specific operations.
*/
<K, V> ReactiveGeoOperations<K, V> opsForGeo(ReactiveSerializationContext<K, V> serializationContext);
<K, V> ReactiveGeoOperations<K, V> opsForGeo(RedisSerializationContext<K, V> serializationContext);
/**
* @return the {@link ReactiveSerializationContext}.
* @return the {@link RedisSerializationContext}.
*/
ReactiveSerializationContext<K, V> getSerializationContext();
RedisSerializationContext<K, V> getSerializationContext();
}

550
src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java
View File

@@ -25,60 +25,66 @@ import java.time.Instant;
import java.util.List;
import org.reactivestreams.Publisher;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.ReactiveSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
* Central abstraction for reactive Redis data access.
* Central abstraction for reactive Redis data access implementing {@link ReactiveRedisOperations}.
* <p/>
* Performs automatic serialization/deserialization between the given objects and the underlying binary data in the
* Redis store. By default, it uses Java serialization for its objects (through {@link JdkSerializationRedisSerializer}
* ).
* <p/>
* Once configured, this class is thread-safe.
* Redis store.
* <p/>
* Note that while the template is generified, it is up to the serializers/deserializers to properly convert the given
* Objects to and from binary data.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
* @param <K> the Redis key type against which the template works (usually a String)
* @param <V> the Redis value type against which the template works
*/
public class ReactiveRedisTemplate<K, V>
implements BeanClassLoaderAware, InitializingBean, ReactiveRedisOperations<K, V> {
public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V> {
private ReactiveRedisConnectionFactory connectionFactory;
private boolean exposeConnection = true;
private boolean initialized = false;
private boolean enableDefaultSerializer = true;
private RedisSerializer<?> defaultSerializer;
private ClassLoader classLoader;
private ReactiveSerializationContextSupport<K, V> serializationContext = new MutableReactiveSerializationContext<>();
// cache singleton objects (where possible)
private ReactiveValueOperations<K, V> valueOps;
private ReactiveListOperations<K, V> listOps;
private ReactiveSetOperations<K, V> setOps;
private ReactiveZSetOperations<K, V> zSetOps;
private ReactiveHyperLogLogOperations<K, V> hyperLogLogOps;
private ReactiveGeoOperations<K, V> geoOps;
private final ReactiveRedisConnectionFactory connectionFactory;
private final RedisSerializationContext<K, V> serializationContext;
private final boolean exposeConnection;
/**
* Construct a new {@link ReactiveRedisTemplate} instance.
* Creates new {@link ReactiveRedisTemplate} using given {@link ReactiveRedisConnectionFactory} and
* {@link RedisSerializationContext}.
*
* @param connectionFactory must not be {@literal null}.
* @param serializationContext must not be {@literal null}.
*/
public ReactiveRedisTemplate() {}
public ReactiveRedisTemplate(ReactiveRedisConnectionFactory connectionFactory,
RedisSerializationContext<K, V> serializationContext) {
this(connectionFactory, serializationContext, false);
}
/**
* Creates new {@link ReactiveRedisTemplate} using given {@link ReactiveRedisConnectionFactory} and
* {@link RedisSerializationContext}.
*
* @param connectionFactory must not be {@literal null}.
* @param serializationContext must not be {@literal null}.
* @param exposeConnection flag indicating to expose the connection used.
*/
public ReactiveRedisTemplate(ReactiveRedisConnectionFactory connectionFactory,
RedisSerializationContext<K, V> serializationContext, boolean exposeConnection) {
Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
Assert.notNull(serializationContext, "SerializationContext must not be null!");
this.connectionFactory = connectionFactory;
this.serializationContext = serializationContext;
this.exposeConnection = exposeConnection;
}
/**
* Returns the connectionFactory.
@@ -89,214 +95,19 @@ public class ReactiveRedisTemplate<K, V>
return connectionFactory;
}
/**
* Sets the connection factory.
*
* @param connectionFactory The connectionFactory to set.
*/
public void setConnectionFactory(ReactiveRedisConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
/**
* @param enableDefaultSerializer Whether or not the default serializer should be used. If not, any serializers not
* explicilty set will remain null and values will not be serialized or deserialized.
*/
public void setEnableDefaultSerializer(boolean enableDefaultSerializer) {
this.enableDefaultSerializer = enableDefaultSerializer;
}
/**
* Returns the default serializer used by this template.
*
* @return template default serializer
*/
public RedisSerializer<?> getDefaultSerializer() {
return defaultSerializer;
}
/**
* Sets the default serializer to use for this template. All serializers (expect the
* {@link #setStringSerializer(RedisSerializer)}) are initialized to this value unless explicitly set. Defaults to
* {@link JdkSerializationRedisSerializer}.
*
* @param serializer default serializer to use
*/
public void setDefaultSerializer(RedisSerializer<?> serializer) {
this.defaultSerializer = serializer;
}
/**
* Returns the key serializer used by this template.
*
* @return the key serializer used by this template.
*/
public RedisSerializer<K> getKeySerializer() {
return serializationContext.getKeySerializer();
}
/**
* Sets the key serializer to be used by this template. Defaults to {@link #getDefaultSerializer()}.
*
* @param serializer the key serializer to be used by this template.
*/
@SuppressWarnings("unchecked")
public void setKeySerializer(RedisSerializer<?> serializer) {
getMutableSerializationContext().setKeySerializer((RedisSerializer) serializer);
}
/**
* Returns the value serializer used by this template.
*
* @return the value serializer used by this template.
*/
public RedisSerializer<V> getValueSerializer() {
return serializationContext.getValueSerializer();
}
/**
* Sets the value serializer to be used by this template. Defaults to {@link #getDefaultSerializer()}.
*
* @param serializer the value serializer to be used by this template.
*/
@SuppressWarnings("unchecked")
public void setValueSerializer(RedisSerializer<?> serializer) {
getMutableSerializationContext().setValueSerializer((RedisSerializer) serializer);
}
/**
* Returns the hashKeySerializer.
*
* @return Returns the hashKeySerializer
*/
public RedisSerializer<?> getHashKeySerializer() {
return serializationContext.getHashKeySerializer();
}
/**
* Sets the hash key (or field) serializer to be used by this template. Defaults to {@link #getDefaultSerializer()}.
*
* @param hashKeySerializer The hashKeySerializer to set.
*/
public void setHashKeySerializer(RedisSerializer<?> hashKeySerializer) {
getMutableSerializationContext().setHashKeySerializer(hashKeySerializer);
}
/**
* Returns the hashValueSerializer.
*
* @return Returns the hashValueSerializer
*/
public RedisSerializer<?> getHashValueSerializer() {
return serializationContext.getHashValueSerializer();
}
/**
* Sets the hash value serializer to be used by this template. Defaults to {@link #getDefaultSerializer()}.
*
* @param hashValueSerializer The hashValueSerializer to set.
*/
public void setHashValueSerializer(RedisSerializer<?> hashValueSerializer) {
getMutableSerializationContext().setHashValueSerializer(hashValueSerializer);
}
/**
* Returns the stringSerializer.
*
* @return Returns the stringSerializer
*/
public RedisSerializer<String> getStringSerializer() {
return serializationContext.getStringSerializer();
}
/**
* Sets the string value serializer to be used by this template (when the arguments or return types are always
* strings). Defaults to {@link StringRedisSerializer}.
*
* @param stringSerializer The stringValueSerializer to set.
* @see ValueOperations#get(Object, long, long)
*/
public void setStringSerializer(RedisSerializer<String> stringSerializer) {
getMutableSerializationContext().setStringSerializer(stringSerializer);
}
/**
* Set the {@link ClassLoader} to be used for the default {@link JdkSerializationRedisSerializer} in case no other
* {@link RedisSerializer} is explicitly set as the default one.
*
* @param classLoader can be {@literal null}.
* @see org.springframework.beans.factory.BeanClassLoaderAware#setBeanClassLoader
*/
@Override
public void setBeanClassLoader(ClassLoader classLoader) {
this.classLoader = classLoader;
}
/**
* Initializes the properties and creates and sets an immutable {@link ReactiveSerializationContext}. Serializers
* cannot be changed after properties are initialized.
*/
@Override
public void afterPropertiesSet() {
boolean defaultUsed = false;
if (this.defaultSerializer == null) {
this.defaultSerializer = new JdkSerializationRedisSerializer(
this.classLoader != null ? this.classLoader : this.getClass().getClassLoader());
}
if (this.enableDefaultSerializer) {
if (this.serializationContext.getKeySerializer() == null) {
setKeySerializer(this.defaultSerializer);
defaultUsed = true;
}
if (this.serializationContext.getValueSerializer() == null) {
setValueSerializer(this.defaultSerializer);
defaultUsed = true;
}
if (this.serializationContext.getHashKeySerializer() == null) {
setHashKeySerializer(this.defaultSerializer);
defaultUsed = true;
}
if (this.serializationContext.getHashValueSerializer() == null) {
setHashValueSerializer(this.defaultSerializer);
defaultUsed = true;
}
}
if (this.enableDefaultSerializer && defaultUsed) {
Assert.notNull(this.defaultSerializer, "Default serializer is null and not all serializers initialized");
}
this.serializationContext = new ImmutableReactiveSerializationContext<K, V>(serializationContext);
this.initialized = true;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForValue()
*/
@Override
public ReactiveValueOperations<K, V> opsForValue() {
if (valueOps == null) {
valueOps = opsForValue(serializationContext);
}
return valueOps;
return opsForValue(serializationContext);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForValue(org.springframework.data.redis.serializer.ReactiveSerializationContext)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForValue(org.springframework.data.redis.serializer.RedisSerializationContext)
*/
@Override
public <K1, V1> ReactiveValueOperations<K1, V1> opsForValue(
ReactiveSerializationContext<K1, V1> serializationContext) {
public <K1, V1> ReactiveValueOperations<K1, V1> opsForValue(RedisSerializationContext<K1, V1> serializationContext) {
return new DefaultReactiveValueOperations<>(this, serializationContext);
}
@@ -305,19 +116,14 @@ public class ReactiveRedisTemplate<K, V>
*/
@Override
public ReactiveListOperations<K, V> opsForList() {
if (listOps == null) {
listOps = opsForList(serializationContext);
}
return listOps;
return opsForList(serializationContext);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForList(org.springframework.data.redis.serializer.ReactiveSerializationContext)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForList(org.springframework.data.redis.serializer.RedisSerializationContext)
*/
@Override
public <K1, V1> ReactiveListOperations<K1, V1> opsForList(ReactiveSerializationContext<K1, V1> serializationContext) {
public <K1, V1> ReactiveListOperations<K1, V1> opsForList(RedisSerializationContext<K1, V1> serializationContext) {
return new DefaultReactiveListOperations<>(this, serializationContext);
}
@@ -325,19 +131,14 @@ public class ReactiveRedisTemplate<K, V>
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForSet()
*/
public ReactiveSetOperations<K, V> opsForSet() {
if (setOps == null) {
setOps = opsForSet(serializationContext);
}
return setOps;
return opsForSet(serializationContext);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForSet(org.springframework.data.redis.serializer.ReactiveSerializationContext)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForSet(org.springframework.data.redis.serializer.RedisSerializationContext)
*/
@Override
public <K1, V1> ReactiveSetOperations<K1, V1> opsForSet(ReactiveSerializationContext<K1, V1> serializationContext) {
public <K1, V1> ReactiveSetOperations<K1, V1> opsForSet(RedisSerializationContext<K1, V1> serializationContext) {
return new DefaultReactiveSetOperations<>(this, serializationContext);
}
@@ -345,19 +146,14 @@ public class ReactiveRedisTemplate<K, V>
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForZSet()
*/
public ReactiveZSetOperations<K, V> opsForZSet() {
if (zSetOps == null) {
zSetOps = opsForZSet(serializationContext);
}
return zSetOps;
return opsForZSet(serializationContext);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForZSet(org.springframework.data.redis.serializer.ReactiveSerializationContext)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForZSet(org.springframework.data.redis.serializer.RedisSerializationContext)
*/
@Override
public <K1, V1> ReactiveZSetOperations<K1, V1> opsForZSet(ReactiveSerializationContext<K1, V1> serializationContext) {
public <K1, V1> ReactiveZSetOperations<K1, V1> opsForZSet(RedisSerializationContext<K1, V1> serializationContext) {
return new DefaultReactiveZSetOperations<>(this, serializationContext);
}
@@ -366,20 +162,15 @@ public class ReactiveRedisTemplate<K, V>
*/
@Override
public ReactiveHyperLogLogOperations<K, V> opsForHyperLogLog() {
if (hyperLogLogOps == null) {
hyperLogLogOps = opsForHyperLogLog(serializationContext);
}
return hyperLogLogOps;
return opsForHyperLogLog(serializationContext);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForHyperLogLog(org.springframework.data.redis.serializer.ReactiveSerializationContext)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForHyperLogLog(org.springframework.data.redis.serializer.RedisSerializationContext)
*/
@Override
public <K1, V1> ReactiveHyperLogLogOperations<K1, V1> opsForHyperLogLog(
ReactiveSerializationContext<K1, V1> serializationContext) {
RedisSerializationContext<K1, V1> serializationContext) {
return new DefaultReactiveHyperLogLogOperations<>(this, serializationContext);
}
@@ -392,11 +183,11 @@ public class ReactiveRedisTemplate<K, V>
}
/* (non-Javadoc)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForHash(org.springframework.data.redis.serializer.ReactiveSerializationContext)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForHash(org.springframework.data.redis.serializer.RedisSerializationContext)
*/
@Override
public <K1, HK, HV> ReactiveHashOperations<K1, HK, HV> opsForHash(
ReactiveSerializationContext<K1, ?> serializationContext) {
RedisSerializationContext<K1, ?> serializationContext) {
return new DefaultReactiveHashOperations<>(this, serializationContext);
}
@@ -405,19 +196,14 @@ public class ReactiveRedisTemplate<K, V>
*/
@Override
public ReactiveGeoOperations<K, V> opsForGeo() {
if (geoOps == null) {
geoOps = opsForGeo(serializationContext);
}
return geoOps;
return opsForGeo(serializationContext);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForGeo(org.springframework.data.redis.serializer.ReactiveSerializationContext)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForGeo(org.springframework.data.redis.serializer.RedisSerializationContext)
*/
@Override
public <K1, V1> ReactiveGeoOperations<K1, V1> opsForGeo(ReactiveSerializationContext<K1, V1> serializationContext) {
public <K1, V1> ReactiveGeoOperations<K1, V1> opsForGeo(RedisSerializationContext<K1, V1> serializationContext) {
return new DefaultReactiveGeoOperations<>(this, serializationContext);
}
@@ -440,7 +226,6 @@ public class ReactiveRedisTemplate<K, V>
*/
public <T> Flux<T> execute(ReactiveRedisCallback<T> action, boolean exposeConnection) {
Assert.isTrue(initialized, "template not initialized; call afterPropertiesSet() before using it");
Assert.notNull(action, "Callback object must not be null");
ReactiveRedisConnectionFactory factory = getConnectionFactory();
@@ -498,7 +283,6 @@ public class ReactiveRedisTemplate<K, V>
*/
private <T> Publisher<T> doInConnection(ReactiveRedisCallback<T> action, boolean exposeConnection) {
Assert.isTrue(initialized, "template not initialized; call afterPropertiesSet() before using it");
Assert.notNull(action, "Callback object must not be null");
ReactiveRedisConnectionFactory factory = getConnectionFactory();
@@ -727,231 +511,15 @@ public class ReactiveRedisTemplate<K, V>
* @see org.springframework.data.redis.core.ReactiveRedisOperations#serialization()
*/
@Override
public ReactiveSerializationContext<K, V> getSerializationContext() {
public RedisSerializationContext<K, V> getSerializationContext() {
return serializationContext;
}
@SuppressWarnings("unchecked")
private MutableReactiveSerializationContext<K, V> getMutableSerializationContext() {
Assert.state(serializationContext instanceof MutableReactiveSerializationContext,
() -> String.format("Client configuration must be instance of MutableReactiveSerializationContext but is %s",
ClassUtils.getShortName(serializationContext.getClass())));
return (MutableReactiveSerializationContext) serializationContext;
}
private ByteBuffer rawKey(K key) {
return getSerializationContext().key().getWriter().write(key);
return getSerializationContext().getKeySerializationPair().getWriter().write(key);
}
private K readKey(ByteBuffer buffer) {
return getSerializationContext().key().getReader().read(buffer);
}
/**
* @author Mark Paluch
*/
static abstract class ReactiveSerializationContextSupport<K, V> implements ReactiveSerializationContext<K, V> {
@Override
public abstract SerializationTuple<K> key();
@Override
public abstract SerializationTuple<V> value();
@Override
public abstract SerializationTuple<String> string();
@Override
public abstract <HK> SerializationTuple<HK> hashKey();
@Override
public abstract <HV> SerializationTuple<HV> hashValue();
public abstract RedisSerializer<K> getKeySerializer();
public abstract RedisSerializer<V> getValueSerializer();
public abstract RedisSerializer<?> getHashKeySerializer();
public abstract RedisSerializer<?> getHashValueSerializer();
public abstract RedisSerializer<String> getStringSerializer();
}
/**
* @author Mark Paluch
*/
static class ImmutableReactiveSerializationContext<K, V> extends ReactiveSerializationContextSupport<K, V> {
private RedisSerializer<K> keySerializer;
private final SerializationTuple<K> keyTuple;
private RedisSerializer<V> valueSerializer;
private final SerializationTuple<V> valueTuple;
private RedisSerializer<?> hashKeySerializer;
private final SerializationTuple<?> hashKeyTuple;
private RedisSerializer<?> hashValueSerializer;
private final SerializationTuple<?> hashValueTuple;
private RedisSerializer<String> stringSerializer;
private final SerializationTuple<String> stringTuple;
public ImmutableReactiveSerializationContext(ReactiveSerializationContextSupport<K, V> context) {
keySerializer = context.getKeySerializer();
keyTuple = context.key();
valueSerializer = context.getValueSerializer();
valueTuple = context.value();
hashKeySerializer = context.getHashKeySerializer();
hashKeyTuple = context.hashKey();
hashValueSerializer = context.getHashValueSerializer();
hashValueTuple = context.hashValue();
stringSerializer = context.getStringSerializer();
stringTuple = context.string();
}
@Override
public SerializationTuple<K> key() {
return keyTuple;
}
@Override
public SerializationTuple<V> value() {
return valueTuple;
}
@Override
public SerializationTuple<String> string() {
return stringTuple;
}
@Override
@SuppressWarnings("unchecked")
public <HK> SerializationTuple<HK> hashKey() {
return (SerializationTuple) hashKeyTuple;
}
@Override
@SuppressWarnings("unchecked")
public <HV> SerializationTuple<HV> hashValue() {
return (SerializationTuple) hashValueTuple;
}
public RedisSerializer<K> getKeySerializer() {
return keySerializer;
}
public RedisSerializer<V> getValueSerializer() {
return valueSerializer;
}
public RedisSerializer<?> getHashKeySerializer() {
return hashKeySerializer;
}
public RedisSerializer<?> getHashValueSerializer() {
return hashValueSerializer;
}
public RedisSerializer<String> getStringSerializer() {
return stringSerializer;
}
}
/**
* @author Mark Paluch
*/
static class MutableReactiveSerializationContext<K, V> extends ReactiveSerializationContextSupport<K, V> {
private RedisSerializer<K> keySerializer;
private SerializationTuple<K> keyTuple = SerializationTuple.raw();
private RedisSerializer<V> valueSerializer;
private SerializationTuple<V> valueTuple = SerializationTuple.raw();
private RedisSerializer<?> hashKeySerializer;
private SerializationTuple<?> hashKeyTuple = SerializationTuple.raw();
private RedisSerializer<?> hashValueSerializer;
private SerializationTuple<?> hashValueTuple = SerializationTuple.raw();
private RedisSerializer<String> stringSerializer = new StringRedisSerializer();
private SerializationTuple<String> stringTuple = SerializationTuple.fromSerializer(stringSerializer);
@Override
public SerializationTuple<K> key() {
return keyTuple;
}
@Override
public SerializationTuple<V> value() {
return valueTuple;
}
@Override
public SerializationTuple<String> string() {
return stringTuple;
}
@Override
@SuppressWarnings("unchecked")
public <HK> SerializationTuple<HK> hashKey() {
return (SerializationTuple) hashKeyTuple;
}
@Override
@SuppressWarnings("unchecked")
public <HV> SerializationTuple<HV> hashValue() {
return (SerializationTuple) hashValueTuple;
}
public RedisSerializer<K> getKeySerializer() {
return keySerializer;
}
public void setKeySerializer(RedisSerializer<K> keySerializer) {
this.keySerializer = keySerializer;
this.keyTuple = SerializationTuple.fromSerializer(keySerializer);
}
public RedisSerializer<V> getValueSerializer() {
return valueSerializer;
}
public void setValueSerializer(RedisSerializer<V> valueSerializer) {
this.valueSerializer = valueSerializer;
this.valueTuple = SerializationTuple.fromSerializer(valueSerializer);
}
public RedisSerializer<?> getHashKeySerializer() {
return hashKeySerializer;
}
public void setHashKeySerializer(RedisSerializer<?> hashKeySerializer) {
this.hashKeySerializer = hashKeySerializer;
this.hashKeyTuple = SerializationTuple.fromSerializer(hashKeySerializer);
}
public RedisSerializer<?> getHashValueSerializer() {
return hashValueSerializer;
}
public void setHashValueSerializer(RedisSerializer<?> hashValueSerializer) {
this.hashValueSerializer = hashValueSerializer;
this.hashValueTuple = SerializationTuple.fromSerializer(hashValueSerializer);
}
public RedisSerializer<String> getStringSerializer() {
return stringSerializer;
}
public void setStringSerializer(RedisSerializer<String> stringSerializer) {
this.stringSerializer = stringSerializer;
this.stringTuple = SerializationTuple.fromSerializer(stringSerializer);
}
return getSerializationContext().getKeySerializationPair().getReader().read(buffer);
}
}

11
src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java
View File

@@ -54,17 +54,6 @@ public interface ReactiveZSetOperations<K, V> {
*/
Mono<Long> addAll(K key, Collection<? extends TypedTuple<V>> tuples);
/**
* Add {@code tuples} to a sorted set at {@code key}, or update its {@code score} if it already exists.
*
* @param key must not be {@literal null}.
* @param tuples must not be {@literal null}.
* @return
* @see <a href="http://redis.io/commands/zadd">Redis Documentation: ZADD</a>
*/
// TODO
// Mono<Long> add(K key, Set<TypedTuple<V>> tuples);
/**
* Remove {@code values} from sorted set. Return number of removed elements.
*

204
src/main/java/org/springframework/data/redis/serializer/DefaultReactiveSerializationContextBuilder.java
View File

@@ -1,204 +0,0 @@
/*
* Copyright 2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.serializer;
import org.springframework.data.redis.serializer.ReactiveSerializationContext.ReactiveSerializationContextBuilder;
import org.springframework.data.redis.serializer.ReactiveSerializationContext.SerializationTuple;
import org.springframework.util.Assert;
/**
* Default implementation of {@link ReactiveSerializationContextBuilder}.
*
* @author Mark Paluch
* @since 2.0
*/
public class DefaultReactiveSerializationContextBuilder<K, V> implements ReactiveSerializationContextBuilder<K, V> {
private SerializationTuple<K> keyTuple;
private SerializationTuple<V> valueTuple;
private SerializationTuple<?> hashKeyTuple;
private SerializationTuple<?> hashValueTuple;
private SerializationTuple<String> stringTuple = SerializationTuple.fromSerializer(new StringRedisSerializer());
@Override
public ReactiveSerializationContextBuilder<K, V> key(SerializationTuple<K> tuple) {
Assert.notNull(tuple, "SerializationTuple must not be null!");
this.keyTuple = tuple;
return this;
}
@Override
public ReactiveSerializationContextBuilder<K, V> key(RedisElementReader<K> reader, RedisElementWriter<K> writer) {
return key(SerializationTuple.just(reader, writer));
}
@Override
public ReactiveSerializationContextBuilder<K, V> key(RedisSerializer<K> serializer) {
return key(SerializationTuple.fromSerializer(serializer));
}
@Override
public ReactiveSerializationContextBuilder<K, V> value(SerializationTuple<V> tuple) {
Assert.notNull(tuple, "SerializationTuple must not be null!");
this.valueTuple = tuple;
return this;
}
@Override
public ReactiveSerializationContextBuilder<K, V> value(RedisElementReader<V> reader, RedisElementWriter<V> writer) {
return value(SerializationTuple.just(reader, writer));
}
@Override
public ReactiveSerializationContextBuilder<K, V> value(RedisSerializer<V> serializer) {
return value(SerializationTuple.fromSerializer(serializer));
}
@Override
public ReactiveSerializationContextBuilder<K, V> hashKey(SerializationTuple<?> tuple) {
Assert.notNull(tuple, "SerializationTuple must not be null!");
this.hashKeyTuple = tuple;
return this;
}
@Override
public ReactiveSerializationContextBuilder<K, V> hashKey(RedisElementReader<?> reader, RedisElementWriter<?> writer) {
return hashKey(SerializationTuple.just(reader, writer));
}
@Override
public ReactiveSerializationContextBuilder<K, V> hashKey(RedisSerializer<?> serializer) {
return hashKey(SerializationTuple.fromSerializer(serializer));
}
@Override
public ReactiveSerializationContextBuilder<K, V> hashValue(SerializationTuple<?> tuple) {
Assert.notNull(tuple, "SerializationTuple must not be null!");
this.hashValueTuple = tuple;
return this;
}
@Override
public ReactiveSerializationContextBuilder<K, V> hashValue(RedisElementReader<?> reader,
RedisElementWriter<?> writer) {
return hashValue(SerializationTuple.just(reader, writer));
}
@Override
public ReactiveSerializationContextBuilder<K, V> hashValue(RedisSerializer<?> serializer) {
return hashValue(SerializationTuple.fromSerializer(serializer));
}
@Override
public ReactiveSerializationContextBuilder<K, V> string(SerializationTuple<String> tuple) {
Assert.notNull(tuple, "SerializationTuple must not be null!");
this.hashValueTuple = tuple;
return this;
}
@Override
public ReactiveSerializationContextBuilder<K, V> string(RedisElementReader<String> reader,
RedisElementWriter<String> writer) {
return string(SerializationTuple.just(reader, writer));
}
@Override
public ReactiveSerializationContextBuilder<K, V> string(RedisSerializer<String> serializer) {
return string(SerializationTuple.fromSerializer(serializer));
}
@Override
public ReactiveSerializationContext<K, V> build() {
Assert.notNull(keyTuple, "Key SerializationTuple must not be null!");
Assert.notNull(valueTuple, "Value SerializationTuple must not be null!");
Assert.notNull(hashKeyTuple, "HashKey SerializationTuple must not be null!");
Assert.notNull(hashValueTuple, "ValueKey SerializationTuple must not be null!");
return new DefaultReactiveSerializationContext<K, V>(keyTuple, valueTuple, hashKeyTuple, hashValueTuple,
stringTuple);
}
static class DefaultReactiveSerializationContext<K, V> implements ReactiveSerializationContext<K, V> {
private final SerializationTuple<K> keyTuple;
private final SerializationTuple<V> valueTuple;
private final SerializationTuple<?> hashKeyTuple;
private final SerializationTuple<?> hashValueTuple;
private final SerializationTuple<String> stringTuple;
public DefaultReactiveSerializationContext(SerializationTuple<K> keyTuple, SerializationTuple<V> valueTuple,
SerializationTuple<?> hashKeyTuple, SerializationTuple<?> hashValueTuple,
SerializationTuple<String> stringTuple) {
this.keyTuple = keyTuple;
this.valueTuple = valueTuple;
this.hashKeyTuple = hashKeyTuple;
this.hashValueTuple = hashValueTuple;
this.stringTuple = stringTuple;
}
@Override
public SerializationTuple<K> key() {
return keyTuple;
}
@Override
public SerializationTuple<V> value() {
return valueTuple;
}
@Override
@SuppressWarnings("unchecked")
public <HK> SerializationTuple<HK> hashKey() {
return (SerializationTuple) hashKeyTuple;
}
@Override
@SuppressWarnings("unchecked")
public <HV> SerializationTuple<HV> hashValue() {
return (SerializationTuple) hashValueTuple;
}
@Override
public SerializationTuple<String> string() {
return stringTuple;
}
}
}

3
src/main/java/org/springframework/data/redis/serializer/DefaultRedisElementReader.java
View File

@@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
* Default implementation of {@link RedisElementReader}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
@RequiredArgsConstructor
@@ -41,7 +42,7 @@ class DefaultRedisElementReader<T> implements RedisElementReader<T> {
return (T) buffer;
}
byte[] bytes = new byte[buffer.remaining()];
byte[] bytes = new byte[buffer.slice().remaining()];
buffer.get(bytes);
return serializer.deserialize(bytes);

177
src/main/java/org/springframework/data/redis/serializer/DefaultRedisSerializationContext.java Normal file
View File

@@ -0,0 +1,177 @@
/*
* Copyright 2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.serializer;
import org.springframework.util.Assert;
/**
* Default implementation of {@link RedisSerializationContext}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
class DefaultRedisSerializationContext<K, V> implements RedisSerializationContext<K, V> {
private final SerializationPair<K> keyTuple;
private final SerializationPair<V> valueTuple;
private final SerializationPair<?> hashKeyTuple;
private final SerializationPair<?> hashValueTuple;
private final SerializationPair<String> stringTuple;
private DefaultRedisSerializationContext(SerializationPair<K> keyTuple, SerializationPair<V> valueTuple,
SerializationPair<?> hashKeyTuple, SerializationPair<?> hashValueTuple, SerializationPair<String> stringTuple) {
this.keyTuple = keyTuple;
this.valueTuple = valueTuple;
this.hashKeyTuple = hashKeyTuple;
this.hashValueTuple = hashValueTuple;
this.stringTuple = stringTuple;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContext#getKeySerializationPair()
*/
@Override
public SerializationPair<K> getKeySerializationPair() {
return keyTuple;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContext#getValueSerializationPair()
*/
@Override
public SerializationPair<V> getValueSerializationPair() {
return valueTuple;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContext#getHashKeySerializationPair()
*/
@Override
@SuppressWarnings("unchecked")
public <HK> SerializationPair<HK> getHashKeySerializationPair() {
return (SerializationPair) hashKeyTuple;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContext#getHashValueSerializationPair()
*/
@Override
@SuppressWarnings("unchecked")
public <HV> SerializationPair<HV> getHashValueSerializationPair() {
return (SerializationPair) hashValueTuple;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContext#getStringSerializationPair()
*/
@Override
public SerializationPair<String> getStringSerializationPair() {
return stringTuple;
}
/**
* Default implementation of {@link RedisSerializationContextBuilder}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
static class DefaultRedisSerializationContextBuilder<K, V> implements RedisSerializationContextBuilder<K, V> {
private SerializationPair<K> keyTuple;
private SerializationPair<V> valueTuple;
private SerializationPair<?> hashKeyTuple;
private SerializationPair<?> hashValueTuple;
private SerializationPair<String> stringTuple = SerializationPair.fromSerializer(new StringRedisSerializer());
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContextBuilder#key(SerializationPair)
*/
@Override
public RedisSerializationContextBuilder<K, V> key(SerializationPair<K> tuple) {
Assert.notNull(tuple, "SerializationPair must not be null!");
this.keyTuple = tuple;
return this;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContextBuilder#value(SerializationPair)
*/
@Override
public RedisSerializationContextBuilder<K, V> value(SerializationPair<V> tuple) {
Assert.notNull(tuple, "SerializationPair must not be null!");
this.valueTuple = tuple;
return this;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContextBuilder#hashKey(SerializationPair)
*/
@Override
public RedisSerializationContextBuilder<K, V> hashKey(SerializationPair<?> tuple) {
Assert.notNull(tuple, "SerializationPair must not be null!");
this.hashKeyTuple = tuple;
return this;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContextBuilder#hashValue(SerializationPair)
*/
@Override
public RedisSerializationContextBuilder<K, V> hashValue(SerializationPair<?> tuple) {
Assert.notNull(tuple, "SerializationPair must not be null!");
this.hashValueTuple = tuple;
return this;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContextBuilder#string(SerializationPair)
*/
@Override
public RedisSerializationContextBuilder<K, V> string(SerializationPair<String> tuple) {
Assert.notNull(tuple, "SerializationPair must not be null!");
this.hashValueTuple = tuple;
return this;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContextBuilder#build()
*/
@Override
public RedisSerializationContext<K, V> build() {
Assert.notNull(keyTuple, "Key SerializationPair must not be null!");
Assert.notNull(valueTuple, "Value SerializationPair must not be null!");
Assert.notNull(hashKeyTuple, "HashKey SerializationPair must not be null!");
Assert.notNull(hashValueTuple, "ValueKey SerializationPair must not be null!");
return new DefaultRedisSerializationContext<K, V>(keyTuple, valueTuple, hashKeyTuple, hashValueTuple,
stringTuple);
}
}
}

22
src/main/java/org/springframework/data/redis/serializer/DefaultSerializationTuple.java → src/main/java/org/springframework/data/redis/serializer/DefaultSerializationPair.java
View File

@@ -15,34 +15,26 @@
*/
package org.springframework.data.redis.serializer;
import org.springframework.data.redis.serializer.ReactiveSerializationContext.SerializationTuple;
import lombok.Getter;
import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
/**
* Default implementation of {@link SerializationTuple}.
* Default implementation of {@link SerializationPair}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
class DefaultSerializationTuple<T> implements SerializationTuple<T> {
@Getter
class DefaultSerializationPair<T> implements SerializationPair<T> {
private final RedisElementReader<T> reader;
private final RedisElementWriter<T> writer;
@SuppressWarnings("unchecked")
protected DefaultSerializationTuple(RedisElementReader<? extends T> reader, RedisElementWriter<? extends T> writer) {
protected DefaultSerializationPair(RedisElementReader<? extends T> reader, RedisElementWriter<? extends T> writer) {
this.reader = (RedisElementReader) reader;
this.writer = (RedisElementWriter) writer;
}
@Override
public RedisElementReader<T> getReader() {
return reader;
}
@Override
public RedisElementWriter<T> getWriter() {
return writer;
}
}

285
src/main/java/org/springframework/data/redis/serializer/ReactiveSerializationContext.java
View File

@@ -1,285 +0,0 @@
/*
* Copyright 2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.serializer;
import java.nio.ByteBuffer;
import org.springframework.util.Assert;
/**
* Serialization context for reactive use.
* <p>
* This context provides {@link SerializationTuple}s for key, value, hash-key (field), hash-value and {@link String}
* serialization and deserialization.
*
* @author Mark Paluch
* @since 2.0
* @see RedisElementWriter
* @see RedisElementReader
*/
public interface ReactiveSerializationContext<K, V> {
/**
* Creates a new {@link ReactiveSerializationContextBuilder}.
*
* @param <K> expected key type.
* @param <V> expected value type.
* @return a new {@link ReactiveSerializationContextBuilder}.
*/
static <K, V> ReactiveSerializationContextBuilder<K, V> builder() {
return new DefaultReactiveSerializationContextBuilder<>();
}
/**
* @return {@link SerializationTuple} for key-typed serialization and deserialization.
*/
SerializationTuple<K> key();
/**
* @return {@link SerializationTuple} for value-typed serialization and deserialization.
*/
SerializationTuple<V> value();
/**
* @return {@link SerializationTuple} for hash-key-typed serialization and deserialization.
*/
<HK> SerializationTuple<HK> hashKey();
/**
* @return {@link SerializationTuple} for hash-value-typed serialization and deserialization.
*/
<HV> SerializationTuple<HV> hashValue();
/**
* @return {@link SerializationTuple} for {@link String}-typed serialization and deserialization.
*/
SerializationTuple<String> string();
/**
* Typed serialization tuple.
*/
interface SerializationTuple<T> {
/**
* Creates a {@link SerializationTuple} adapter given {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return a {@link SerializationTuple} adapter for {@link RedisSerializer}.
*/
static <T> SerializationTuple<T> fromSerializer(RedisSerializer<T> serializer) {
Assert.notNull(serializer, "RedisSerializer must not be null!");
return new RedisSerializerTupleAdapter<T>(serializer);
}
/**
* Creates a {@link SerializationTuple} adapter given {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return a {@link SerializationTuple} encapsulating {@link RedisElementReader} and {@link RedisElementWriter}.
*/
static <T> SerializationTuple<T> just(RedisElementReader<? extends T> reader,
RedisElementWriter<? extends T> writer) {
Assert.notNull(reader, "RedisElementReader must not be null!");
Assert.notNull(writer, "RedisElementWriter must not be null!");
return new DefaultSerializationTuple<>(reader, writer);
}
/**
* Creates a pass-thru {@link SerializationTuple} to pass-thru {@link ByteBuffer} objects.
*
* @return a pass-thru {@link SerializationTuple}.
*/
static <T> SerializationTuple<T> raw() {
return RedisSerializerTupleAdapter.raw();
}
/**
* @return the {@link RedisElementReader}.
*/
RedisElementReader<T> getReader();
/**
* Deserialize a {@link ByteBuffer} into the according type.
*
* @param buffer must not be {@literal null}.
* @return the deserialized value.
*/
default T read(ByteBuffer buffer) {
return getReader().read(buffer);
}
/**
* @return the {@link RedisElementWriter}.
*/
RedisElementWriter<T> getWriter();
/**
* Serialize a {@code element} to its {@link ByteBuffer} representation.
*
* @param element
* @return the {@link ByteBuffer} representing {@code element} in its binary form.
*/
default ByteBuffer write(T element) {
return getWriter().write(element);
}
}
/**
* Builder for {@link ReactiveSerializationContext}.
*/
interface ReactiveSerializationContextBuilder<K, V> {
/**
* Set the key {@link SerializationTuple}.
*
* @param tuple must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> key(SerializationTuple<K> tuple);
/**
* Set the key {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> key(RedisElementReader<K> reader, RedisElementWriter<K> writer);
/**
* Set the key {@link SerializationTuple} given a {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> key(RedisSerializer<K> serializer);
/**
* Set the value {@link SerializationTuple}.
*
* @param tuple must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> value(SerializationTuple<V> tuple);
/**
* Set the value {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> value(RedisElementReader<V> reader, RedisElementWriter<V> writer);
/**
* Set the value {@link SerializationTuple} given a {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> value(RedisSerializer<V> serializer);
/**
* Set the hash key {@link SerializationTuple}.
*
* @param tuple must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> hashKey(SerializationTuple<?> tuple);
/**
* Set the hash key {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> hashKey(RedisElementReader<? extends Object> reader,
RedisElementWriter<? extends Object> writer);
/**
* Set the hash key {@link SerializationTuple} given a {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> hashKey(RedisSerializer<? extends Object> serializer);
/**
* Set the hash value {@link SerializationTuple}.
*
* @param tuple must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> hashValue(SerializationTuple<?> tuple);
/**
* Set the hash value {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> hashValue(RedisElementReader<? extends Object> reader,
RedisElementWriter<? extends Object> writer);
/**
* Set the hash value {@link SerializationTuple} given a {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> hashValue(RedisSerializer<? extends Object> serializer);
/**
* Set the string {@link SerializationTuple}.
*
* @param tuple must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> string(SerializationTuple<String> tuple);
/**
* Set the string {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> string(RedisElementReader<String> reader,
RedisElementWriter<String> writer);
/**
* Set the string {@link SerializationTuple} given a {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return {@literal this} builder.
*/
ReactiveSerializationContextBuilder<K, V> string(RedisSerializer<String> serializer);
/**
* Builds a {@link ReactiveSerializationContext}.
*
* @return the {@link ReactiveSerializationContext}.
*/
ReactiveSerializationContext<K, V> build();
}
}

406
src/main/java/org/springframework/data/redis/serializer/RedisSerializationContext.java Normal file
View File

@@ -0,0 +1,406 @@
/*
* Copyright 2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.serializer;
import java.nio.ByteBuffer;
import org.springframework.util.Assert;
/**
* Serialization context for reactive use.
* <p />
* This context provides {@link SerializationPair}s for key, value, hash-key (field), hash-value and {@link String}
* serialization and deserialization.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
* @see RedisElementWriter
* @see RedisElementReader
*/
public interface RedisSerializationContext<K, V> {
/**
* Creates a new {@link RedisSerializationContextBuilder}.
*
* @param <K> expected key type.
* @param <V> expected value type.
* @return a new {@link RedisSerializationContextBuilder}.
*/
@SuppressWarnings("unchecked")
static <K, V> RedisSerializationContextBuilder<K, V> newSerializationContext() {
return new DefaultRedisSerializationContext.DefaultRedisSerializationContextBuilder();
}
/**
* Creates a new {@link RedisSerializationContextBuilder} using a given default {@link RedisSerializer}.
*
* @param defaultSerializer must not be {@literal null}.
* @param <K> expected key type.
* @param <V> expected value type.
* @return a new {@link RedisSerializationContextBuilder}.
*/
static <K, V> RedisSerializationContextBuilder<K, V> newSerializationContext(RedisSerializer<?> defaultSerializer) {
Assert.notNull(defaultSerializer, "DefaultSerializer must not be null!");
return newSerializationContext(SerializationPair.fromSerializer(defaultSerializer));
}
/**
* Creates a new {@link RedisSerializationContextBuilder} using a given default {@link SerializationPair}.
*
* @param serializationPair must not be {@literal null}.
* @param <K> expected key type.
* @param <V> expected value type.
* @return a new {@link RedisSerializationContextBuilder}.
*/
@SuppressWarnings("unchecked")
static <K, V> RedisSerializationContextBuilder<K, V> newSerializationContext(SerializationPair<?> serializationPair) {
Assert.notNull(serializationPair, "SerializationPair must not be null!");
return new DefaultRedisSerializationContext.DefaultRedisSerializationContextBuilder() //
.key(serializationPair).value(serializationPair) //
.hashKey(serializationPair).hashValue(serializationPair);
}
/**
* Creates a new {@link RedisSerializationContext} using a {@link SerializationPair#raw()} serialization pair.
*
* @return
*/
static RedisSerializationContext<byte[], byte[]> raw() {
return just(SerializationPair.raw());
}
/**
* Creates a new {@link RedisSerializationContext} using a {@link StringRedisSerializer}.
*
* @return
*/
static RedisSerializationContext<String, String> string() {
return fromSerializer(new StringRedisSerializer());
}
/**
* Creates a new {@link RedisSerializationContext} using the given {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @param <T>
* @return
*/
static <T> RedisSerializationContext<T, T> fromSerializer(RedisSerializer<T> serializer) {
return just(SerializationPair.fromSerializer(serializer));
}
/**
* Creates a new {@link RedisSerializationContext} using the given {@link SerializationPair}.
*
* @param serializationPair
* @param <T>
* @return
*/
static <T> RedisSerializationContext<T, T> just(SerializationPair<T> serializationPair) {
return RedisSerializationContext.<T, T>newSerializationContext(serializationPair).build();
}
/**
* @return {@link SerializationPair} for key-typed serialization and deserialization.
*/
SerializationPair<K> getKeySerializationPair();
/**
* @return {@link SerializationPair} for value-typed serialization and deserialization.
*/
SerializationPair<V> getValueSerializationPair();
/**
* @return {@link SerializationPair} for hash-key-typed serialization and deserialization.
*/
<HK> SerializationPair<HK> getHashKeySerializationPair();
/**
* @return {@link SerializationPair} for hash-value-typed serialization and deserialization.
*/
<HV> SerializationPair<HV> getHashValueSerializationPair();
/**
* @return {@link SerializationPair} for {@link String}-typed serialization and deserialization.
*/
SerializationPair<String> getStringSerializationPair();
/**
* Typed serialization tuple.
*
* @author Mark Paluch
* @author Christoph Strobl
*/
interface SerializationPair<T> {
/**
* Creates a {@link SerializationPair} adapter given {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return a {@link SerializationPair} adapter for {@link RedisSerializer}.
*/
static <T> SerializationPair<T> fromSerializer(RedisSerializer<T> serializer) {
Assert.notNull(serializer, "RedisSerializer must not be null!");
return new RedisSerializerToSerializationPairAdapter<T>(serializer);
}
/**
* Creates a {@link SerializationPair} adapter given {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return a {@link SerializationPair} encapsulating {@link RedisElementReader} and {@link RedisElementWriter}.
*/
static <T> SerializationPair<T> just(RedisElementReader<? extends T> reader,
RedisElementWriter<? extends T> writer) {
Assert.notNull(reader, "RedisElementReader must not be null!");
Assert.notNull(writer, "RedisElementWriter must not be null!");
return new DefaultSerializationPair<>(reader, writer);
}
/**
* Creates a pass through {@link SerializationPair} to pass-thru {@link ByteBuffer} objects.
*
* @return a pass through {@link SerializationPair}.
*/
static <T> SerializationPair<T> raw() {
return RedisSerializerToSerializationPairAdapter.raw();
}
/**
* @return the {@link RedisElementReader}.
*/
RedisElementReader<T> getReader();
/**
* Deserialize a {@link ByteBuffer} into the according type.
*
* @param buffer must not be {@literal null}.
* @return the deserialized value.
*/
default T read(ByteBuffer buffer) {
return getReader().read(buffer);
}
/**
* @return the {@link RedisElementWriter}.
*/
RedisElementWriter<T> getWriter();
/**
* Serialize a {@code element} to its {@link ByteBuffer} representation.
*
* @param element
* @return the {@link ByteBuffer} representing {@code element} in its binary form.
*/
default ByteBuffer write(T element) {
return getWriter().write(element);
}
}
/**
* Builder for {@link RedisSerializationContext}.
*
* @author Mark Paluch
* @author Christoph Strobl
*/
interface RedisSerializationContextBuilder<K, V> {
/**
* Set the key {@link SerializationPair}.
*
* @param pair must not be {@literal null}.
* @return {@literal this} builder.
*/
RedisSerializationContextBuilder<K, V> key(SerializationPair<K> pair);
/**
* Set the key {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return {@literal this} builder.
*/
default RedisSerializationContextBuilder<K, V> key(RedisElementReader<K> reader, RedisElementWriter<K> writer) {
key(SerializationPair.just(reader, writer));
return this;
}
/**
* Set the key {@link SerializationPair} given a {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return {@literal this} builder.
*/
default RedisSerializationContextBuilder<K, V> key(RedisSerializer<K> serializer) {
key(SerializationPair.fromSerializer(serializer));
return this;
}
/**
* Set the value {@link SerializationPair}.
*
* @param pair must not be {@literal null}.
* @return {@literal this} builder.
*/
RedisSerializationContextBuilder<K, V> value(SerializationPair<V> pair);
/**
* Set the value {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return {@literal this} builder.
*/
default RedisSerializationContextBuilder<K, V> value(RedisElementReader<V> reader, RedisElementWriter<V> writer) {
value(SerializationPair.just(reader, writer));
return this;
}
/**
* Set the value {@link SerializationPair} given a {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return {@literal this} builder.
*/
default RedisSerializationContextBuilder<K, V> value(RedisSerializer<V> serializer) {
value(SerializationPair.fromSerializer(serializer));
return this;
}
/**
* Set the hash key {@link SerializationPair}.
*
* @param pair must not be {@literal null}.
* @return {@literal this} builder.
*/
RedisSerializationContextBuilder<K, V> hashKey(SerializationPair<?> pair);
/**
* Set the hash key {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return {@literal this} builder.
*/
default RedisSerializationContextBuilder<K, V> hashKey(RedisElementReader<? extends Object> reader,
RedisElementWriter<? extends Object> writer) {
hashKey(SerializationPair.just(reader, writer));
return this;
}
/**
* Set the hash key {@link SerializationPair} given a {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return {@literal this} builder.
*/
default RedisSerializationContextBuilder<K, V> hashKey(RedisSerializer<? extends Object> serializer) {
hashKey(SerializationPair.fromSerializer(serializer));
return this;
}
/**
* Set the hash value {@link SerializationPair}.
*
* @param pair must not be {@literal null}.
* @return {@literal this} builder.
*/
RedisSerializationContextBuilder<K, V> hashValue(SerializationPair<?> pair);
/**
* Set the hash value {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return {@literal this} builder.
*/
default RedisSerializationContextBuilder<K, V> hashValue(RedisElementReader<? extends Object> reader,
RedisElementWriter<? extends Object> writer) {
hashValue(SerializationPair.just(reader, writer));
return this;
}
/**
* Set the hash value {@link SerializationPair} given a {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return {@literal this} builder.
*/
default RedisSerializationContextBuilder<K, V> hashValue(RedisSerializer<? extends Object> serializer) {
hashValue(SerializationPair.fromSerializer(serializer));
return this;
}
/**
* Set the string {@link SerializationPair}.
*
* @param pair must not be {@literal null}.
* @return {@literal this} builder.
*/
RedisSerializationContextBuilder<K, V> string(SerializationPair<String> pair);
/**
* Set the string {@link RedisElementReader} and {@link RedisElementWriter}.
*
* @param reader must not be {@literal null}.
* @param writer must not be {@literal null}.
* @return {@literal this} builder.
*/
default RedisSerializationContextBuilder<K, V> string(RedisElementReader<String> reader,
RedisElementWriter<String> writer) {
string(SerializationPair.just(reader, writer));
return this;
}
/**
* Set the string {@link SerializationPair} given a {@link RedisSerializer}.
*
* @param serializer must not be {@literal null}.
* @return {@literal this} builder.
*/
default RedisSerializationContextBuilder<K, V> string(RedisSerializer<String> serializer) {
string(SerializationPair.fromSerializer(serializer));
return this;
}
/**
* Builds a {@link RedisSerializationContext}.
*
* @return the {@link RedisSerializationContext}.
*/
RedisSerializationContext<K, V> build();
}
}

74
src/main/java/org/springframework/data/redis/serializer/RedisSerializerToSerializationPairAdapter.java Normal file
View File

@@ -0,0 +1,74 @@
/*
* Copyright 2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.serializer;
import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
import org.springframework.util.Assert;
/**
* Adapter to delegate serialization/deserialization to {@link RedisSerializer}.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
class RedisSerializerToSerializationPairAdapter<T> implements SerializationPair<T> {
private final static RedisSerializerToSerializationPairAdapter<?> RAW = new RedisSerializerToSerializationPairAdapter<>(
null);
private final DefaultSerializationPair pair;
protected RedisSerializerToSerializationPairAdapter(RedisSerializer<T> serializer) {
pair = new DefaultSerializationPair(new DefaultRedisElementReader<>(serializer),
new DefaultRedisElementWriter<>(serializer));
}
@SuppressWarnings("unchecked")
public static <T> SerializationPair<T> raw() {
return (SerializationPair) RAW;
}
/**
* Create a {@link SerializationPair} from given {@link RedisSerializer}.
*
* @param redisSerializer must not be {@literal null}.
* @param <T>
* @return
*/
public static <T> SerializationPair<T> from(RedisSerializer<T> redisSerializer) {
Assert.notNull(redisSerializer, "RedisSerializer must not be null!");
return new RedisSerializerToSerializationPairAdapter<>(redisSerializer);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair#reader()
*/
@Override
public RedisElementReader<T> getReader() {
return pair.getReader();
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair#writer()
*/
@Override
public RedisElementWriter<T> getWriter() {
return pair.getWriter();
}
}

63
src/main/java/org/springframework/data/redis/serializer/RedisSerializerTupleAdapter.java
View File

@@ -1,63 +0,0 @@
/*
* Copyright 2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.serializer;
import org.springframework.data.redis.serializer.ReactiveSerializationContext.SerializationTuple;
/**
* Adapter to delegate serialization/deserialization to {@link RedisSerializer}.
*
* @author Mark Paluch
* @since 2.0
*/
class RedisSerializerTupleAdapter<T> implements SerializationTuple<T> {
private final static RedisSerializerTupleAdapter<?> RAW = new RedisSerializerTupleAdapter<>(null);
private final RedisElementReader<T> reader;
private final RedisElementWriter<T> writer;
protected RedisSerializerTupleAdapter(RedisSerializer<T> serializer) {
reader = new DefaultRedisElementReader<>(serializer);
writer = new DefaultRedisElementWriter<>(serializer);
}
@SuppressWarnings("unchecked")
public static <T> SerializationTuple<T> raw() {
return (SerializationTuple) RAW;
}
public static <T> SerializationTuple<T> from(RedisSerializer<T> redisSerializer) {
return new RedisSerializerTupleAdapter<>(redisSerializer);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.ReactiveSerializationContext.SerializationTuple#reader()
*/
@Override
public RedisElementReader<T> getReader() {
return reader;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.serializer.ReactiveSerializationContext.SerializationTuple#writer()
*/
@Override
public RedisElementWriter<T> getWriter() {
return writer;
}
}