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

DATAREDIS-1150 - Polishing

Browse Source 
Defer cluster topology resolution, update Javadoc and add tests.

Original Pull Request: #533
This commit is contained in:
Christoph Strobl
2020-06-17 09:26:39 +02:00
parent 9501036754
commit bf4f41bf52
3 changed files with 58 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
src/main/java/org/springframework/data/redis/connection/ReactiveClusterCommands.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2020 the original author or authors.
* Copyright 2020 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.
@@ -25,20 +25,22 @@ import java.util.Map;
import org.springframework.data.redis.connection.RedisClusterNode.SlotRange;
/**
* Interface for the {@literal cluster} commands supported by Redis executed using reactive infrastructure.. A
* Interface for the {@literal cluster} commands supported by Redis executed using reactive infrastructure. A
* {@link RedisClusterNode} can be obtained from {@link #clusterGetNodes()} or it can be constructed using either
* {@link RedisClusterNode#getHost() host} and {@link RedisClusterNode#getPort()} or the {@link RedisClusterNode#getId()
* node Id}.
*
* @author Mark Paluch
* @since 2.3.1
* @author Christoph Strobl
* @since 2.3.2
*/
public interface ReactiveClusterCommands {
/**
* Retrieve cluster node information such as {@literal id}, {@literal host}, {@literal port} and {@literal slots}.
*
* @return never {@literal null}.
* @return a {@link Flux} emitting {@link RedisClusterNode cluster nodes}, an {@link Flux#empty() empty one} if none
* found.
* @see <a href="https://redis.io/commands/cluster-nodes">Redis Documentation: CLUSTER NODES</a>
*/
Flux<RedisClusterNode> clusterGetNodes();
@@ -47,7 +49,8 @@ public interface ReactiveClusterCommands {
* Retrieve information about connected slaves for given master node.
*
* @param master must not be {@literal null}.
* @return never {@literal null}.
* @return a {@link Flux} emitting {@link RedisClusterNode cluster nodes}, an {@link Flux#empty() empty one} if none
* found.
* @see <a href="https://redis.io/commands/cluster-slaves">Redis Documentation: CLUSTER SLAVES</a>
*/
Flux<RedisClusterNode> clusterGetSlaves(RedisClusterNode master);
@@ -64,7 +67,7 @@ public interface ReactiveClusterCommands {
* Find the slot for a given {@code key}.
*
* @param key must not be {@literal null}.
* @return
* @return a {@link Mono} emitting the calculated slog.
* @see <a href="https://redis.io/commands/cluster-keyslot">Redis Documentation: CLUSTER KEYSLOT</a>
*/
Mono<Integer> clusterGetSlotForKey(ByteBuffer key);
@@ -73,7 +76,7 @@ public interface ReactiveClusterCommands {
* Find the {@link RedisClusterNode} serving given {@literal slot}.
*
* @param slot
* @return
* @return a {@link Mono} emitting the {@link RedisClusterNode} handling the given slot.
*/
Mono<RedisClusterNode> clusterGetNodeForSlot(int slot);
@@ -81,14 +84,14 @@ public interface ReactiveClusterCommands {
* Find the {@link RedisClusterNode} serving given {@literal key}.
*
* @param key must not be {@literal null}.
* @return
* @return a {@link Mono} emitting the {@link RedisClusterNode} handling the slot for the given key.
*/
Mono<RedisClusterNode> clusterGetNodeForKey(ByteBuffer key);
/**
* Get cluster information.
*
* @return
* @return never {@literal null}.
* @see <a href="https://redis.io/commands/cluster-info">Redis Documentation: CLUSTER INFO</a>
*/
Mono<ClusterInfo> clusterGetClusterInfo();
@@ -97,7 +100,8 @@ public interface ReactiveClusterCommands {
* Assign slots to given {@link RedisClusterNode}.
*
* @param node must not be {@literal null}.
* @param slots
* @param slots must not be empty.
* @return a {@link Mono} signaling completion.
* @see <a href="https://redis.io/commands/cluster-addslots">Redis Documentation: CLUSTER ADDSLOTS</a>
*/
Mono<Void> clusterAddSlots(RedisClusterNode node, int... slots);
@@ -107,6 +111,7 @@ public interface ReactiveClusterCommands {
*
* @param node must not be {@literal null}.
* @param range must not be {@literal null}.
* @return a {@link Mono} signaling completion.
* @see <a href="https://redis.io/commands/cluster-addslots">Redis Documentation: CLUSTER ADDSLOTS</a>
*/
Mono<Void> clusterAddSlots(RedisClusterNode node, SlotRange range);
@@ -115,7 +120,7 @@ public interface ReactiveClusterCommands {
* Count the number of keys assigned to one {@literal slot}.
*
* @param slot
* @return
* @return a {@link Mono} emitting the number of keys stored at the given slot.
* @see <a href="https://redis.io/commands/cluster-countkeysinslot">Redis Documentation: CLUSTER COUNTKEYSINSLOT</a>
*/
Mono<Long> clusterCountKeysInSlot(int slot);
@@ -124,7 +129,7 @@ public interface ReactiveClusterCommands {
* Remove slots from {@link RedisClusterNode}.
*
* @param node must not be {@literal null}.
* @param slots
* @return a {@link Mono} signaling completion.
* @see <a href="https://redis.io/commands/cluster-delslots">Redis Documentation: CLUSTER DELSLOTS</a>
*/
Mono<Void> clusterDeleteSlots(RedisClusterNode node, int... slots);
@@ -134,6 +139,7 @@ public interface ReactiveClusterCommands {
*
* @param node must not be {@literal null}.
* @param range must not be {@literal null}.
* @return a {@link Mono} signaling completion.
* @see <a href="https://redis.io/commands/cluster-delslots">Redis Documentation: CLUSTER DELSLOTS</a>
*/
Mono<Void> clusterDeleteSlotsInRange(RedisClusterNode node, SlotRange range);
@@ -142,6 +148,7 @@ public interface ReactiveClusterCommands {
* Remove given {@literal node} from cluster.
*
* @param node must not be {@literal null}.
* @return a {@link Mono} signaling completion.
* @see <a href="https://redis.io/commands/cluster-forget">Redis Documentation: CLUSTER FORGET</a>
*/
Mono<Void> clusterForget(RedisClusterNode node);
@@ -151,6 +158,7 @@ public interface ReactiveClusterCommands {
*
* @param node must contain {@link RedisClusterNode#getHost() host} and {@link RedisClusterNode#getPort()} and must
* not be {@literal null}.
* @return a {@link Mono} signaling completion.
* @see <a href="https://redis.io/commands/cluster-meet">Redis Documentation: CLUSTER MEET</a>
*/
Mono<Void> clusterMeet(RedisClusterNode node);
@@ -159,6 +167,7 @@ public interface ReactiveClusterCommands {
* @param node must not be {@literal null}.
* @param slot
* @param mode must not be{@literal null}.
* @return a {@link Mono} signaling completion.
* @see <a href="https://redis.io/commands/cluster-setslot">Redis Documentation: CLUSTER SETSLOT</a>
*/
Mono<Void> clusterSetSlot(RedisClusterNode node, int slot, AddSlots mode);
@@ -168,7 +177,8 @@ public interface ReactiveClusterCommands {
*
* @param slot
* @param count must not be {@literal null}.
* @return
* @return a {@link Flux} emitting the number of requested keys in the given slot, or signalling completion if none
* found.
* @see <a href="https://redis.io/commands/cluster-getkeysinslot">Redis Documentation: CLUSTER GETKEYSINSLOT</a>
*/
Flux<ByteBuffer> clusterGetKeysInSlot(int slot, int count);
@@ -178,6 +188,7 @@ public interface ReactiveClusterCommands {
*
* @param master must not be {@literal null}.
* @param replica must not be {@literal null}.
* @return a {@link Mono} signaling completion.
* @see <a href="https://redis.io/commands/cluster-replicate">Redis Documentation: CLUSTER REPLICATE</a>
*/
Mono<Void> clusterReplicate(RedisClusterNode master, RedisClusterNode replica);

42
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java
View File

@@ -224,7 +224,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
*/
@Override
public Flux<RedisClusterNode> clusterGetNodes() {
return Flux.fromIterable(doGetActiveNodes());
return Flux.fromStream(() -> doGetActiveNodes().stream());
}
/*
@@ -236,10 +236,9 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
Assert.notNull(master, "Master must not be null!");
RedisClusterNode nodeToUse = lookup(master);
return execute(nodeToUse, cmd -> cmd.clusterSlaves(nodeToUse.getId()) //
.flatMapIterable(LettuceConverters::toSetOfRedisClusterNodes));
return Mono.fromSupplier(() -> lookup(master))
.flatMapMany(nodeToUse -> execute(nodeToUse, cmd -> cmd.clusterSlaves(nodeToUse.getId()) //
.flatMapIterable(LettuceConverters::toSetOfRedisClusterNodes)));
}
/*
@@ -249,7 +248,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
@Override
public Mono<Map<RedisClusterNode, Collection<RedisClusterNode>>> clusterGetMasterSlaveMap() {
return Flux.fromIterable(topologyProvider.getTopology().getActiveMasterNodes()) //
return Flux.fromStream(() -> topologyProvider.getTopology().getActiveMasterNodes().stream()) //
.flatMap(node -> {
return Mono.just(node).zipWith(execute(node, cmd -> cmd.clusterSlaves(node.getId())) //
.collectList() //
@@ -295,6 +294,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
*/
@Override
public Mono<ClusterInfo> clusterGetClusterInfo() {
return executeCommandOnArbitraryNode(RedisClusterReactiveCommands::clusterInfo) //
.map(LettuceConverters::toProperties) //
.map(ClusterInfo::new) //
@@ -359,12 +359,14 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
@Override
public Mono<Void> clusterForget(RedisClusterNode node) {
List<RedisClusterNode> nodes = new ArrayList<>(doGetActiveNodes());
RedisClusterNode nodeToRemove = lookup(node);
nodes.remove(nodeToRemove);
return Flux.fromIterable(nodes).flatMap(actualNode -> execute(node, cmd -> cmd.clusterForget(nodeToRemove.getId())))
.then();
return Flux.fromStream(() -> {
List<RedisClusterNode> nodes = new ArrayList<>(doGetActiveNodes());
nodes.remove(nodeToRemove);
return nodes.stream();
}).flatMap(actualNode -> execute(node, cmd -> cmd.clusterForget(nodeToRemove.getId()))).then();
}
/*
@@ -392,11 +394,11 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
Assert.notNull(node, "Node must not be null.");
Assert.notNull(mode, "AddSlots mode must not be null.");
RedisClusterNode nodeToUse = lookup(node);
String nodeId = nodeToUse.getId();
return execute(node, cmd -> {
RedisClusterNode nodeToUse = lookup(node);
String nodeId = nodeToUse.getId();
switch (mode) {
case MIGRATING:
return cmd.clusterSetSlotMigrating(slot, nodeId);
@@ -428,10 +430,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
*/
@Override
public Mono<Void> clusterReplicate(RedisClusterNode master, RedisClusterNode replica) {
RedisClusterNode masterToUse = lookup(master);
return execute(replica, cmd -> cmd.clusterReplicate(masterToUse.getId())).then();
return execute(replica, cmd -> cmd.clusterReplicate(lookup(master).getId())).then();
}
/**
@@ -445,10 +444,13 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
Assert.notNull(callback, "ReactiveCallback must not be null!");
List<RedisClusterNode> nodes = new ArrayList<>(doGetActiveNodes());
int random = new Random().nextInt(nodes.size());
return Mono.fromSupplier(() -> {
return execute(nodes.get(random), callback);
List<RedisClusterNode> nodes = new ArrayList<>(doGetActiveNodes());
int random = new Random().nextInt(nodes.size());
return nodes.get(random);
}).flatMapMany(it -> execute(it, callback));
}
/**