DATAREDIS-1150 - Add ReactiveRedisClusterCommands.

We now provide Redis Cluster commands (CLUSTER INFO, CLUSTER NODES, …) exposed through a reactive API.

Original Pull Request: #533

src/main/java/org/springframework/data/redis/connection/ReactiveClusterCommands.java

@@ -0,0 +1,189 @@
+/*
+ * Copyright 2015-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.connection;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+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
+ * {@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
+ */
+public interface ReactiveClusterCommands {
+
+	/**
+	 * Retrieve cluster node information such as {@literal id}, {@literal host}, {@literal port} and {@literal slots}.
+	 *
+	 * @return never {@literal null}.
+	 * @see <a href="https://redis.io/commands/cluster-nodes">Redis Documentation: CLUSTER NODES</a>
+	 */
+	Flux<RedisClusterNode> clusterGetNodes();
+
+	/**
+	 * Retrieve information about connected slaves for given master node.
+	 *
+	 * @param master must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @see <a href="https://redis.io/commands/cluster-slaves">Redis Documentation: CLUSTER SLAVES</a>
+	 */
+	Flux<RedisClusterNode> clusterGetSlaves(RedisClusterNode master);
+
+	/**
+	 * Retrieve information about masters and their connected slaves.
+	 *
+	 * @return never {@literal null}.
+	 * @see <a href="https://redis.io/commands/cluster-slaves">Redis Documentation: CLUSTER SLAVES</a>
+	 */
+	Mono<Map<RedisClusterNode, Collection<RedisClusterNode>>> clusterGetMasterSlaveMap();
+
+	/**
+	 * Find the slot for a given {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/cluster-keyslot">Redis Documentation: CLUSTER KEYSLOT</a>
+	 */
+	Mono<Integer> clusterGetSlotForKey(ByteBuffer key);
+
+	/**
+	 * Find the {@link RedisClusterNode} serving given {@literal slot}.
+	 *
+	 * @param slot
+	 * @return
+	 */
+	Mono<RedisClusterNode> clusterGetNodeForSlot(int slot);
+
+	/**
+	 * Find the {@link RedisClusterNode} serving given {@literal key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return
+	 */
+	Mono<RedisClusterNode> clusterGetNodeForKey(ByteBuffer key);
+
+	/**
+	 * Get cluster information.
+	 *
+	 * @return
+	 * @see <a href="https://redis.io/commands/cluster-info">Redis Documentation: CLUSTER INFO</a>
+	 */
+	Mono<ClusterInfo> clusterGetClusterInfo();
+
+	/**
+	 * Assign slots to given {@link RedisClusterNode}.
+	 *
+	 * @param node must not be {@literal null}.
+	 * @param slots
+	 * @see <a href="https://redis.io/commands/cluster-addslots">Redis Documentation: CLUSTER ADDSLOTS</a>
+	 */
+	Mono<Void> clusterAddSlots(RedisClusterNode node, int... slots);
+
+	/**
+	 * Assign {@link SlotRange#getSlotsArray()} to given {@link RedisClusterNode}.
+	 *
+	 * @param node must not be {@literal null}.
+	 * @param range must not be {@literal null}.
+	 * @see <a href="https://redis.io/commands/cluster-addslots">Redis Documentation: CLUSTER ADDSLOTS</a>
+	 */
+	Mono<Void> clusterAddSlots(RedisClusterNode node, SlotRange range);
+
+	/**
+	 * Count the number of keys assigned to one {@literal slot}.
+	 *
+	 * @param slot
+	 * @return
+	 * @see <a href="https://redis.io/commands/cluster-countkeysinslot">Redis Documentation: CLUSTER COUNTKEYSINSLOT</a>
+	 */
+	Mono<Long> clusterCountKeysInSlot(int slot);
+
+	/**
+	 * Remove slots from {@link RedisClusterNode}.
+	 *
+	 * @param node must not be {@literal null}.
+	 * @param slots
+	 * @see <a href="https://redis.io/commands/cluster-delslots">Redis Documentation: CLUSTER DELSLOTS</a>
+	 */
+	Mono<Void> clusterDeleteSlots(RedisClusterNode node, int... slots);
+
+	/**
+	 * Removes {@link SlotRange#getSlotsArray()} from given {@link RedisClusterNode}.
+	 *
+	 * @param node must not be {@literal null}.
+	 * @param range must not be {@literal null}.
+	 * @see <a href="https://redis.io/commands/cluster-delslots">Redis Documentation: CLUSTER DELSLOTS</a>
+	 */
+	Mono<Void> clusterDeleteSlotsInRange(RedisClusterNode node, SlotRange range);
+
+	/**
+	 * Remove given {@literal node} from cluster.
+	 *
+	 * @param node must not be {@literal null}.
+	 * @see <a href="https://redis.io/commands/cluster-forget">Redis Documentation: CLUSTER FORGET</a>
+	 */
+	Mono<Void> clusterForget(RedisClusterNode node);
+
+	/**
+	 * Add given {@literal node} to cluster.
+	 *
+	 * @param node must contain {@link RedisClusterNode#getHost() host} and {@link RedisClusterNode#getPort()} and must
+	 *          not be {@literal null}.
+	 * @see <a href="https://redis.io/commands/cluster-meet">Redis Documentation: CLUSTER MEET</a>
+	 */
+	Mono<Void> clusterMeet(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @param slot
+	 * @param mode must not be{@literal null}.
+	 * @see <a href="https://redis.io/commands/cluster-setslot">Redis Documentation: CLUSTER SETSLOT</a>
+	 */
+	Mono<Void> clusterSetSlot(RedisClusterNode node, int slot, AddSlots mode);
+
+	/**
+	 * Get {@literal keys} served by slot.
+	 *
+	 * @param slot
+	 * @param count must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/cluster-getkeysinslot">Redis Documentation: CLUSTER GETKEYSINSLOT</a>
+	 */
+	Flux<ByteBuffer> clusterGetKeysInSlot(int slot, int count);
+
+	/**
+	 * Assign a {@literal slave} to given {@literal master}.
+	 *
+	 * @param master must not be {@literal null}.
+	 * @param replica must not be {@literal null}.
+	 * @see <a href="https://redis.io/commands/cluster-replicate">Redis Documentation: CLUSTER REPLICATE</a>
+	 */
+	Mono<Void> clusterReplicate(RedisClusterNode master, RedisClusterNode replica);
+
+	enum AddSlots {
+		MIGRATING, IMPORTING, STABLE, NODE
+	}
+
+}

src/main/java/org/springframework/data/redis/connection/ReactiveRedisClusterConnection.java

@@ -22,7 +22,7 @@ import reactor.core.publisher.Mono;
  * @author Mark Paluch
  * @since 2.0
  */
-public interface ReactiveRedisClusterConnection extends ReactiveRedisConnection {
+public interface ReactiveRedisClusterConnection extends ReactiveRedisConnection, ReactiveClusterCommands {
 
 	@Override
 	ReactiveClusterKeyCommands keyCommands();

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

@@ -20,17 +20,29 @@ import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
 import io.lettuce.core.api.reactive.RedisReactiveCommands;
 import io.lettuce.core.cluster.RedisClusterClient;
+import io.lettuce.core.cluster.SlotHash;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.stream.Collectors;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.redis.connection.ClusterInfo;
 import org.springframework.data.redis.connection.ClusterTopologyProvider;
 import org.springframework.data.redis.connection.ReactiveRedisClusterConnection;
 import org.springframework.data.redis.connection.RedisClusterNode;
 import org.springframework.data.redis.connection.RedisNode;
+import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -188,6 +200,15 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 		return new LettuceReactiveClusterStreamCommands(this);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterConnection#ping()
+	 */
+	@Override
+	public Mono<String> ping() {
+		return clusterGetNodes().flatMap(node -> execute(node, BaseRedisReactiveCommands::ping)).last();
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterConnection#ping(org.springframework.data.redis.connection.RedisClusterNode)
@@ -197,7 +218,242 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 		return execute(node, BaseRedisReactiveCommands::ping).next();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterGetNodes()
+	 */
+	@Override
+	public Flux<RedisClusterNode> clusterGetNodes() {
+		return Flux.fromIterable(doGetActiveNodes());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterGetSlaves(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public Flux<RedisClusterNode> clusterGetSlaves(RedisClusterNode master) {
+
+		Assert.notNull(master, "Master must not be null!");
+
+		RedisClusterNode nodeToUse = lookup(master);
+
+		return execute(nodeToUse, cmd -> cmd.clusterSlaves(nodeToUse.getId()) //
+				.flatMapIterable(LettuceConverters::toSetOfRedisClusterNodes));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterGetMasterSlaveMap()
+	 */
+	@Override
+	public Mono<Map<RedisClusterNode, Collection<RedisClusterNode>>> clusterGetMasterSlaveMap() {
+
+		return Flux.fromIterable(topologyProvider.getTopology().getActiveMasterNodes()) //
+				.flatMap(node -> {
+					return Mono.just(node).zipWith(execute(node, cmd -> cmd.clusterSlaves(node.getId())) //
+							.collectList() //
+							.map(Converters::toSetOfRedisClusterNodes));
+				}).collect(Collectors.toMap(Tuple2::getT1, Tuple2::getT2));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterGetSlotForKey(java.nio.ByteBuffer)
+	 */
+	@Override
+	public Mono<Integer> clusterGetSlotForKey(ByteBuffer key) {
+		return Mono.fromSupplier(() -> SlotHash.getSlot(key));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterGetNodeForSlot(int)
+	 */
+	@Override
+	public Mono<RedisClusterNode> clusterGetNodeForSlot(int slot) {
+
+		Set<RedisClusterNode> nodes = topologyProvider.getTopology().getSlotServingNodes(slot);
+		return nodes.isEmpty() ? Mono.empty() : Flux.fromIterable(nodes).next();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterGetNodeForKey(java.nio.ByteBuffer)
+	 */
+	@Override
+	public Mono<RedisClusterNode> clusterGetNodeForKey(ByteBuffer key) {
+
+		Assert.notNull(key, "Key must not be null.");
+
+		return clusterGetSlotForKey(key).flatMap(this::clusterGetNodeForSlot);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterGetClusterInfo()
+	 */
+	@Override
+	public Mono<ClusterInfo> clusterGetClusterInfo() {
+		return executeCommandOnArbitraryNode(RedisClusterReactiveCommands::clusterInfo) //
+				.map(LettuceConverters::toProperties) //
+				.map(ClusterInfo::new) //
+				.single();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterAddSlots(org.springframework.data.redis.connection.RedisClusterNode, int[])
+	 */
+	@Override
+	public Mono<Void> clusterAddSlots(RedisClusterNode node, int... slots) {
+		return execute(node, cmd -> cmd.clusterAddSlots(slots)).then();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterAddSlots(org.springframework.data.redis.connection.RedisClusterNode, org.springframework.data.redis.connection.RedisClusterNode.SlotRange)
+	 */
+	@Override
+	public Mono<Void> clusterAddSlots(RedisClusterNode node, RedisClusterNode.SlotRange range) {
+
+		Assert.notNull(range, "Range must not be null.");
+
+		return execute(node, cmd -> cmd.clusterAddSlots(range.getSlotsArray())).then();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterCountKeysInSlot(int)
+	 */
+	@Override
+	public Mono<Long> clusterCountKeysInSlot(int slot) {
+		return execute(cmd -> cmd.clusterCountKeysInSlot(slot)).next();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterDeleteSlots(org.springframework.data.redis.connection.RedisClusterNode, int[])
+	 */
+	@Override
+	public Mono<Void> clusterDeleteSlots(RedisClusterNode node, int... slots) {
+		return execute(node, cmd -> cmd.clusterDelSlots(slots)).then();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterDeleteSlotsInRange(org.springframework.data.redis.connection.RedisClusterNode, org.springframework.data.redis.connection.RedisClusterNode.SlotRange)
+	 */
+	@Override
+	public Mono<Void> clusterDeleteSlotsInRange(RedisClusterNode node, RedisClusterNode.SlotRange range) {
+
+		Assert.notNull(range, "Range must not be null.");
+
+		return execute(node, cmd -> cmd.clusterDelSlots(range.getSlotsArray())).then();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterForget(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@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();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterMeet(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public Mono<Void> clusterMeet(RedisClusterNode node) {
+
+		Assert.notNull(node, "Cluster node must not be null for CLUSTER MEET command!");
+		Assert.hasText(node.getHost(), "Node to meet cluster must have a host!");
+		Assert.isTrue(node.getPort() != null && node.getPort() > 0, "Node to meet cluster must have a port greater 0!");
+
+		return clusterGetNodes()
+				.flatMap(actualNode -> execute(node, cmd -> cmd.clusterMeet(node.getHost(), node.getPort()))).then();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterSetSlot(org.springframework.data.redis.connection.RedisClusterNode, int, org.springframework.data.redis.connection.ReactiveRedisClusterCommands.AddSlots)
+	 */
+	@Override
+	public Mono<Void> clusterSetSlot(RedisClusterNode node, int slot, AddSlots mode) {
+
+		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 -> {
+
+			switch (mode) {
+				case MIGRATING:
+					return cmd.clusterSetSlotMigrating(slot, nodeId);
+				case IMPORTING:
+					return cmd.clusterSetSlotImporting(slot, nodeId);
+				case NODE:
+					return cmd.clusterSetSlotNode(slot, nodeId);
+				case STABLE:
+					return cmd.clusterSetSlotStable(slot);
+				default:
+					throw new InvalidDataAccessApiUsageException("Invalid import mode for cluster slot: " + slot);
+			}
+
+		}).then();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterGetKeysInSlot(int, int)
+	 */
+	@Override
+	public Flux<ByteBuffer> clusterGetKeysInSlot(int slot, int count) {
+		return execute(cmd -> cmd.clusterGetKeysInSlot(slot, count));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisClusterCommands#clusterReplicate(org.springframework.data.redis.connection.RedisClusterNode, org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public Mono<Void> clusterReplicate(RedisClusterNode master, RedisClusterNode replica) {
+
+		RedisClusterNode masterToUse = lookup(master);
+
+		return execute(replica, cmd -> cmd.clusterReplicate(masterToUse.getId())).then();
+	}
+
 	/**
+	 * Run {@link LettuceReactiveCallback} on a random node.
+	 *
+	 * @param callback must not be {@literal null}.
+	 * @throws IllegalArgumentException when {@code node} or {@code callback} is {@literal null}.
+	 * @return {@link Flux} emitting execution results.
+	 */
+	public <T> Flux<T> executeCommandOnArbitraryNode(LettuceReactiveCallback<T> callback) {
+
+		Assert.notNull(callback, "ReactiveCallback must not be null!");
+
+		List<RedisClusterNode> nodes = new ArrayList<>(doGetActiveNodes());
+		int random = new Random().nextInt(nodes.size());
+
+		return execute(nodes.get(random), callback);
+	}
+
+	/**
+	 * Run {@link LettuceReactiveCallback} on given {@link RedisClusterNode}.
+	 *
 	 * @param node must not be {@literal null}.
 	 * @param callback must not be {@literal null}.
 	 * @throws IllegalArgumentException when {@code node} or {@code callback} is {@literal null}.
@@ -205,12 +461,8 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 	 */
 	public <T> Flux<T> execute(RedisNode node, LettuceReactiveCallback<T> callback) {
 
-		try {
-			Assert.notNull(node, "RedisClusterNode must not be null!");
-			Assert.notNull(callback, "ReactiveCallback must not be null!");
-		} catch (IllegalArgumentException e) {
-			return Flux.error(e);
-		}
+		Assert.notNull(node, "RedisClusterNode must not be null!");
+		Assert.notNull(callback, "ReactiveCallback must not be null!");
 
 		return getCommands(node).flatMapMany(callback::doWithCommands).onErrorMap(translateException());
 	}
@@ -244,4 +496,19 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 		return getConnection().flatMap(it -> Mono.fromCompletionStage(it.getConnectionAsync(node.getHost(), node.getPort()))
 				.map(StatefulRedisConnection::reactive));
 	}
+
+	/**
+	 * Lookup a {@link RedisClusterNode} by using either ids {@link RedisClusterNode#getId() node id} or host and port to
+	 * obtain the full node details from the underlying {@link ClusterTopologyProvider}.
+	 *
+	 * @param nodeToLookup the node to lookup.
+	 * @return the {@link RedisClusterNode} from the topology lookup.
+	 */
+	private RedisClusterNode lookup(RedisClusterNode nodeToLookup) {
+		return topologyProvider.getTopology().lookup(nodeToLookup);
+	}
+
+	private Set<RedisClusterNode> doGetActiveNodes() {
+		return topologyProvider.getTopology().getActiveNodes();
+	}
 }