From bf4f41bf52d5ab7d56d3cf9085cded025f0d0490 Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Wed, 17 Jun 2020 09:26:39 +0200 Subject: [PATCH] DATAREDIS-1150 - Polishing Defer cluster topology resolution, update Javadoc and add tests. Original Pull Request: #533 --- .../connection/ReactiveClusterCommands.java | 37 ++++++++++------ ...LettuceReactiveRedisClusterConnection.java | 42 ++++++++++--------- .../LettuceReactiveClusterCommandsTests.java | 13 +++++- 3 files changed, 58 insertions(+), 34 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterCommands.java index 10a5ee4b4..7a227e06a 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterCommands.java @@ -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 Redis Documentation: CLUSTER NODES */ Flux 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 Redis Documentation: CLUSTER SLAVES */ Flux 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 Redis Documentation: CLUSTER KEYSLOT */ Mono 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 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 clusterGetNodeForKey(ByteBuffer key); /** * Get cluster information. * - * @return + * @return never {@literal null}. * @see Redis Documentation: CLUSTER INFO */ Mono 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 Redis Documentation: CLUSTER ADDSLOTS */ Mono 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 Redis Documentation: CLUSTER ADDSLOTS */ Mono 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 Redis Documentation: CLUSTER COUNTKEYSINSLOT */ Mono 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 Redis Documentation: CLUSTER DELSLOTS */ Mono 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 Redis Documentation: CLUSTER DELSLOTS */ Mono 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 Redis Documentation: CLUSTER FORGET */ Mono 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 Redis Documentation: CLUSTER MEET */ Mono 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 Redis Documentation: CLUSTER SETSLOT */ Mono 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 Redis Documentation: CLUSTER GETKEYSINSLOT */ Flux 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 Redis Documentation: CLUSTER REPLICATE */ Mono clusterReplicate(RedisClusterNode master, RedisClusterNode replica); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java index 0aebacccc..0dac853e5 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java @@ -224,7 +224,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti */ @Override public Flux 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>> 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 clusterGetClusterInfo() { + return executeCommandOnArbitraryNode(RedisClusterReactiveCommands::clusterInfo) // .map(LettuceConverters::toProperties) // .map(ClusterInfo::new) // @@ -359,12 +359,14 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti @Override public Mono clusterForget(RedisClusterNode node) { - List 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 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 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 nodes = new ArrayList<>(doGetActiveNodes()); - int random = new Random().nextInt(nodes.size()); + return Mono.fromSupplier(() -> { - return execute(nodes.get(random), callback); + List nodes = new ArrayList<>(doGetActiveNodes()); + int random = new Random().nextInt(nodes.size()); + + return nodes.get(random); + }).flatMapMany(it -> execute(it, callback)); } /** diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterCommandsTests.java index aec89586e..83da4c2ee 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterCommandsTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterCommandsTests.java @@ -22,7 +22,6 @@ import reactor.test.StepVerifier; import java.nio.ByteBuffer; import org.junit.Test; - import org.springframework.data.redis.connection.ReactiveClusterCommands; import org.springframework.data.redis.connection.RedisClusterNode; @@ -33,9 +32,21 @@ import org.springframework.data.redis.connection.RedisClusterNode; * the master node. * * @author Mark Paluch + * @author Christoph Strobl */ public class LettuceReactiveClusterCommandsTests extends LettuceReactiveClusterCommandsTestsBase { + @Test // DATAREDIS-1150 + public void pingShouldReturnPong() { + connection.ping().as(StepVerifier::create).expectNext("PONG").verifyComplete(); + } + + @Test // DATAREDIS-1150 + public void pingShouldReturnPongForServers() { + connection.clusterGetNodes().flatMap(connection::ping).as(StepVerifier::create) + .expectNext("PONG", "PONG", "PONG", "PONG").verifyComplete(); + } + @Test // DATAREDIS-1150 public void clusterGetNodesShouldReturnNodes() {