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

DATAREDIS-467 - Fix cluster multi key commands.

Browse Source 
We moved away from returning raw map types and now return dedicated objects for command executions in cluster environment. This allows to maintain node information and collecting results from each and every callback. MGET now returns values according to the key position.

Additionally we now prefix info commands in the cluster with the nodes host:port.

Original pull request: #174.
This commit is contained in:
Christoph Strobl
2016-02-24 14:55:11 +01:00
committed by Mark Paluch
parent 6f26a5a073
commit 02742e3b49
7 changed files with 606 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files

341
src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java
View File

@@ -16,7 +16,12 @@
package org.springframework.data.redis.connection;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
@@ -35,13 +40,17 @@ import org.springframework.data.redis.ClusterRedirectException;
import org.springframework.data.redis.ClusterStateFailureException;
import org.springframework.data.redis.ExceptionTranslationStrategy;
import org.springframework.data.redis.TooManyClusterRedirectionsException;
import org.springframework.data.redis.connection.util.ByteArraySet;
import org.springframework.data.redis.connection.util.ByteArrayWrapper;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
/**
* {@link ClusterCommandExecutor} takes care of running commands across the known cluster nodes. By providing an
* {@link AsyncTaskExecutor} the execution behavior can be influenced.
*
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 1.7
@@ -56,7 +65,7 @@ public class ClusterCommandExecutor implements DisposableBean {
/**
* Create a new instance of {@link ClusterCommandExecutor}.
*
*
* @param topologyProvider must not be {@literal null}.
* @param resourceProvider must not be {@literal null}.
* @param exceptionTranslation must not be {@literal null}.
@@ -96,11 +105,11 @@ public class ClusterCommandExecutor implements DisposableBean {
/**
* Run {@link ClusterCommandCallback} on a random node.
*
*
* @param cmd must not be {@literal null}.
* @return
*/
public <T> T executeCommandOnArbitraryNode(ClusterCommandCallback<?, T> cmd) {
public <T> NodeResult<T> executeCommandOnArbitraryNode(ClusterCommandCallback<?, T> cmd) {
Assert.notNull(cmd, "ClusterCommandCallback must not be null!");
List<RedisClusterNode> nodes = new ArrayList<RedisClusterNode>(getClusterTopology().getActiveNodes());
@@ -109,27 +118,26 @@ public class ClusterCommandExecutor implements DisposableBean {
/**
* Run {@link ClusterCommandCallback} on given {@link RedisClusterNode}.
*
*
* @param cmd must not be {@literal null}.
* @param node must not be {@literal null}.
* @throws IllegalArgumentException in case no resource can be acquired for given node.
* @return
*/
public <S, T> T executeCommandOnSingleNode(ClusterCommandCallback<S, T> cmd, RedisClusterNode node) {
public <S, T> NodeResult<T> executeCommandOnSingleNode(ClusterCommandCallback<S, T> cmd, RedisClusterNode node) {
return executeCommandOnSingleNode(cmd, node, 0);
}
private <S, T> T executeCommandOnSingleNode(ClusterCommandCallback<S, T> cmd, RedisClusterNode node, int redirectCount) {
private <S, T> NodeResult<T> executeCommandOnSingleNode(ClusterCommandCallback<S, T> cmd, RedisClusterNode node,
int redirectCount) {
Assert.notNull(cmd, "ClusterCommandCallback must not be null!");
Assert.notNull(node, "RedisClusterNode must not be null!");
if (redirectCount > maxRedirects) {
throw new TooManyClusterRedirectionsException(
String
.format(
"Cannot follow Cluster Redirects over more than %s legs. Please consider increasing the number of redirects to follow. Current value is: %s.",
redirectCount, maxRedirects));
throw new TooManyClusterRedirectionsException(String.format(
"Cannot follow Cluster Redirects over more than %s legs. Please consider increasing the number of redirects to follow. Current value is: %s.",
redirectCount, maxRedirects));
}
RedisClusterNode nodeToUse = lookupNode(node);
@@ -138,7 +146,7 @@ public class ClusterCommandExecutor implements DisposableBean {
Assert.notNull(client, "Could not acquire resource for node. Is your cluster info up to date?");
try {
return cmd.doInCluster(client);
return new NodeResult<T>(node, cmd.doInCluster(client));
} catch (RuntimeException ex) {
RuntimeException translatedException = convertToDataAccessExeption(ex);
@@ -156,7 +164,7 @@ public class ClusterCommandExecutor implements DisposableBean {
/**
* Lookup node from the topology.
*
*
* @param node
* @return
* @throws IllegalArgumentException in case the node could not be resolved to a topology-known node
@@ -171,12 +179,12 @@ public class ClusterCommandExecutor implements DisposableBean {
/**
* Run {@link ClusterCommandCallback} on all reachable master nodes.
*
*
* @param cmd
* @return
* @throws ClusterCommandExecutionFailureException
*/
public <S, T> Map<RedisClusterNode, T> executeCommandOnAllNodes(final ClusterCommandCallback<S, T> cmd) {
public <S, T> MulitNodeResult<T> executeCommandOnAllNodes(final ClusterCommandCallback<S, T> cmd) {
return executeCommandAsyncOnNodes(cmd, getClusterTopology().getActiveMasterNodes());
}
@@ -187,8 +195,8 @@ public class ClusterCommandExecutor implements DisposableBean {
* @throws ClusterCommandExecutionFailureException
* @throws IllegalArgumentException in case the node could not be resolved to a topology-known node
*/
public <S, T> java.util.Map<RedisClusterNode, T> executeCommandAsyncOnNodes(
final ClusterCommandCallback<S, T> callback, Iterable<RedisClusterNode> nodes) {
public <S, T> MulitNodeResult<T> executeCommandAsyncOnNodes(final ClusterCommandCallback<S, T> callback,
Iterable<RedisClusterNode> nodes) {
Assert.notNull(callback, "Callback must not be null!");
Assert.notNull(nodes, "Nodes must not be null!");
@@ -204,13 +212,13 @@ public class ClusterCommandExecutor implements DisposableBean {
}
}
Map<RedisClusterNode, Future<T>> futures = new LinkedHashMap<RedisClusterNode, Future<T>>();
Map<NodeExecution, Future<NodeResult<T>>> futures = new LinkedHashMap<NodeExecution, Future<NodeResult<T>>>();
for (final RedisClusterNode node : resolvedRedisClusterNodes) {
futures.put(node, executor.submit(new Callable<T>() {
futures.put(new NodeExecution(node), executor.submit(new Callable<NodeResult<T>>() {
@Override
public T call() throws Exception {
public NodeResult<T> call() throws Exception {
return executeCommandOnSingleNode(callback, node);
}
}));
@@ -219,35 +227,40 @@ public class ClusterCommandExecutor implements DisposableBean {
return collectResults(futures);
}
private <T> Map<RedisClusterNode, T> collectResults(Map<RedisClusterNode, Future<T>> futures) {
private <T> MulitNodeResult<T> collectResults(Map<NodeExecution, Future<NodeResult<T>>> futures) {
boolean done = false;
Map<RedisClusterNode, T> result = new HashMap<RedisClusterNode, T>();
MulitNodeResult<T> result = new MulitNodeResult<T>();
Map<RedisClusterNode, Throwable> exceptions = new HashMap<RedisClusterNode, Throwable>();
Set<String> saveGuard = new HashSet<String>();
while (!done) {
done = true;
for (Map.Entry<RedisClusterNode, Future<T>> entry : futures.entrySet()) {
for (Map.Entry<NodeExecution, Future<NodeResult<T>>> entry : futures.entrySet()) {
if (!entry.getValue().isDone() && !entry.getValue().isCancelled()) {
done = false;
} else {
if (!result.containsKey(entry.getKey()) && !exceptions.containsKey(entry.getKey())) {
try {
result.put(entry.getKey(), entry.getValue().get());
} catch (ExecutionException e) {
try {
RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause());
exceptions.put(entry.getKey(), ex != null ? ex : e.getCause());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause());
exceptions.put(entry.getKey(), ex != null ? ex : e.getCause());
break;
String futureId = ObjectUtils.getIdentityHexString(entry.getValue());
if (!saveGuard.contains(futureId)) {
result.add(entry.getValue().get());
saveGuard.add(futureId);
}
} catch (ExecutionException e) {
RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause());
exceptions.put(entry.getKey().getNode(), ex != null ? ex : e.getCause());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause());
exceptions.put(entry.getKey().getNode(), ex != null ? ex : e.getCause());
break;
}
}
}
@@ -268,12 +281,12 @@ public class ClusterCommandExecutor implements DisposableBean {
/**
* Run {@link MultiKeyClusterCommandCallback} with on a curated set of nodes serving one or more keys.
*
*
* @param cmd
* @return
* @throws ClusterCommandExecutionFailureException
*/
public <S, T> Map<RedisClusterNode, T> executeMuliKeyCommand(final MultiKeyClusterCommandCallback<S, T> cmd,
public <S, T> MulitNodeResult<T> executeMuliKeyCommand(final MultiKeyClusterCommandCallback<S, T> cmd,
Iterable<byte[]> keys) {
Map<RedisClusterNode, Set<byte[]>> nodeKeyMap = new HashMap<RedisClusterNode, Set<byte[]>>();
@@ -291,26 +304,28 @@ public class ClusterCommandExecutor implements DisposableBean {
}
}
Map<RedisClusterNode, Future<T>> futures = new LinkedHashMap<RedisClusterNode, Future<T>>();
Map<NodeExecution, Future<NodeResult<T>>> futures = new LinkedHashMap<NodeExecution, Future<NodeResult<T>>>();
for (final Entry<RedisClusterNode, Set<byte[]>> entry : nodeKeyMap.entrySet()) {
if (entry.getKey().isMaster()) {
for (final byte[] key : entry.getValue()) {
futures.put(entry.getKey(), executor.submit(new Callable<T>() {
futures.put(new NodeExecution(entry.getKey(), key), executor.submit(new Callable<NodeResult<T>>() {
@Override
public T call() throws Exception {
return (T) executeMultiKeyCommandOnSingleNode(cmd, entry.getKey(), key);
public NodeResult<T> call() throws Exception {
return executeMultiKeyCommandOnSingleNode(cmd, entry.getKey(), key);
}
}));
}
}
}
return collectResults(futures);
}
private <S, T> T executeMultiKeyCommandOnSingleNode(MultiKeyClusterCommandCallback<S, T> cmd, RedisClusterNode node,
byte[] key) {
private <S, T> NodeResult<T> executeMultiKeyCommandOnSingleNode(MultiKeyClusterCommandCallback<S, T> cmd,
RedisClusterNode node, byte[] key) {
Assert.notNull(cmd, "MultiKeyCommandCallback must not be null!");
Assert.notNull(node, "RedisClusterNode must not be null!");
@@ -320,7 +335,7 @@ public class ClusterCommandExecutor implements DisposableBean {
Assert.notNull(client, "Could not acquire resource for node. Is your cluster info up to date?");
try {
return cmd.doInCluster(client, key);
return new NodeResult<T>(node, cmd.doInCluster(client, key), key);
} catch (RuntimeException ex) {
RuntimeException translatedException = convertToDataAccessExeption(ex);
@@ -340,7 +355,7 @@ public class ClusterCommandExecutor implements DisposableBean {
/**
* Set the maximum number of redirects to follow on {@code MOVED} or {@code ASK}.
*
*
* @param maxRedirects set to zero to suspend redirects.
*/
public void setMaxRedirects(int maxRedirects) {
@@ -362,7 +377,7 @@ public class ClusterCommandExecutor implements DisposableBean {
/**
* Callback interface for Redis 'low level' code using the cluster client directly. To be used with
* {@link ClusterCommandExecutor} execution methods.
*
*
* @author Christoph Strobl
* @param <T> native driver connection
* @param <S>
@@ -374,7 +389,7 @@ public class ClusterCommandExecutor implements DisposableBean {
/**
* Callback interface for Redis 'low level' code using the cluster client to execute multi key commands.
*
*
* @author Christoph Strobl
* @param <T> native driver connection
* @param <S>
@@ -383,4 +398,232 @@ public class ClusterCommandExecutor implements DisposableBean {
S doInCluster(T client, byte[] key);
}
/**
* {@link NodeExecution} encapsulates the execution of a command on a specific node along with arguments, such as
* keys, involved.
*
* @author Christoph Strobl
* @since 1.7
*/
private static class NodeExecution {
private RedisClusterNode node;
private Object[] args;
public NodeExecution(RedisClusterNode node, Object... args) {
this.node = node;
this.args = args;
}
/**
* Get the {@link RedisClusterNode} the execution happens on.
*/
public RedisClusterNode getNode() {
return node;
}
/*
* (non-Javadoc)
* @see java.lang.Object#hashCode()
*/
@Override
public int hashCode() {
int result = ObjectUtils.nullSafeHashCode(node);
return result + ObjectUtils.nullSafeHashCode(args);
}
/*
* (non-Javadoc)
* @see java.lang.Object#equals(java.lang.Object)
*/
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj == null) {
return false;
}
if (!(obj instanceof NodeExecution)) {
return false;
}
NodeExecution that = (NodeExecution) obj;
if (!ObjectUtils.nullSafeEquals(this.node, that.node)) {
return false;
}
return ObjectUtils.nullSafeEquals(this.args, that.args);
}
}
/**
* {@link NodeResult} encapsules the actual value returned by a {@link ClusterCommandCallback} on a given
* {@link RedisClusterNode}.
*
* @author Christoph Strobl
* @param <T>
* @since 1.7
*/
public static class NodeResult<T> {
private RedisClusterNode node;
private T value;
private ByteArrayWrapper key;
/**
* Create new {@link NodeResult}.
*
* @param node
* @param value
*/
public NodeResult(RedisClusterNode node, T value) {
this(node, value, new byte[] {});
}
/**
* Create new {@link NodeResult}.
*
* @param node
* @param value
* @parm key
*/
public NodeResult(RedisClusterNode node, T value, byte[] key) {
this.node = node;
this.value = value;
this.key = new ByteArrayWrapper(key);
}
/**
* Get the actual value of the command execution.
*
* @return can be {@literal null}.
*/
public T getValue() {
return value;
}
/**
* Get the {@link RedisClusterNode} the command was executed on.
*
* @return never {@literal null}.
*/
public RedisClusterNode getNode() {
return node;
}
/**
* @return
*/
public byte[] getKey() {
return key.getArray();
}
}
/**
* {@link MulitNodeResult} holds all {@link NodeResult} of a command executed on multiple {@link RedisClusterNode}.
*
* @author Christoph Strobl
* @param <T>
* @since 1.7
*/
public static class MulitNodeResult<T> {
List<NodeResult<T>> nodeResults = new ArrayList<NodeResult<T>>();
private void add(NodeResult<T> result) {
nodeResults.add(result);
}
/**
* @return never {@literal null}.
*/
public List<NodeResult<T>> getResults() {
return Collections.unmodifiableList(nodeResults);
}
/**
* Get {@link List} of all individual {@link NodeResult#value}. <br />
* The resulting {@link List} may contain {@literal null} values.
*
* @return never {@literal null}.
*/
public List<T> resultsAsList() {
return toList(nodeResults);
}
/**
* Get {@link List} of all individual {@link NodeResult#value}. <br />
* The resulting {@link List} may contain {@literal null} values.
*
* @return never {@literal null}.
*/
public List<T> resultsAsListSortBy(byte[]... keys) {
ArrayList<NodeResult<T>> clone = new ArrayList<NodeResult<T>>(nodeResults);
Collections.sort(clone, new ResultByReferenceKeyPositionComperator(keys));
return toList(clone);
}
/**
* @param returnValue
* @return
*/
public T getFirstNonNullNotEmptyOrDefault(T returnValue) {
for (NodeResult<T> nodeResult : nodeResults) {
if (nodeResult.getValue() != null) {
if (nodeResult.getValue() instanceof Map) {
if (CollectionUtils.isEmpty((Map<?, ?>) nodeResult.getValue())) {
return nodeResult.getValue();
}
} else if (CollectionUtils.isEmpty((Collection<?>) nodeResult.getValue())) {
return nodeResult.getValue();
} else {
return nodeResult.getValue();
}
}
}
return returnValue;
}
private List<T> toList(Collection<NodeResult<T>> source) {
ArrayList<T> result = new ArrayList<T>();
for (NodeResult<T> nodeResult : source) {
result.add(nodeResult.getValue());
}
return result;
}
/**
* {@link Comparator} for sorting {@link NodeResult} by reference keys.
*
* @author Christoph Strobl
*/
private static class ResultByReferenceKeyPositionComperator implements Comparator<NodeResult<?>> {
List<ByteArrayWrapper> reference;
public ResultByReferenceKeyPositionComperator(byte[]... keys) {
reference = new ArrayList<ByteArrayWrapper>(new ByteArraySet(Arrays.asList(keys)));
}
@Override
public int compare(NodeResult<?> o1, NodeResult<?> o2) {
return Integer.compare(reference.indexOf(o1.key), reference.indexOf(o2.key));
}
}
}
}

293
src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java
View File

@@ -39,6 +39,7 @@ import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
import org.springframework.data.redis.connection.ClusterCommandExecutor;
import org.springframework.data.redis.connection.ClusterCommandExecutor.ClusterCommandCallback;
import org.springframework.data.redis.connection.ClusterCommandExecutor.MultiKeyClusterCommandCallback;
import org.springframework.data.redis.connection.ClusterCommandExecutor.NodeResult;
import org.springframework.data.redis.connection.ClusterInfo;
import org.springframework.data.redis.connection.ClusterNodeResourceProvider;
import org.springframework.data.redis.connection.ClusterSlotHashUtil;
@@ -110,8 +111,8 @@ public class JedisClusterConnection implements RedisClusterConnection {
closed = false;
topologyProvider = new JedisClusterTopologyProvider(cluster);
clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider,
new JedisClusterNodeResourceProvider(cluster), EXCEPTION_TRANSLATION);
clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, new JedisClusterNodeResourceProvider(cluster),
EXCEPTION_TRANSLATION);
try {
DirectFieldAccessor dfa = new DirectFieldAccessor(cluster);
@@ -168,14 +169,14 @@ public class JedisClusterConnection implements RedisClusterConnection {
}
}
return Long.valueOf(this.clusterCommandExecutor.executeMuliKeyCommand(
new JedisMultiKeyClusterCommandCallback<Long>() {
return Long
.valueOf(this.clusterCommandExecutor.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback<Long>() {
@Override
public Long doInCluster(Jedis client, byte[] key) {
return client.del(key);
}
}, Arrays.asList(keys)).size());
}, Arrays.asList(keys)).resultsAsList().size());
}
/*
@@ -201,14 +202,14 @@ public class JedisClusterConnection implements RedisClusterConnection {
Assert.notNull(pattern, "Pattern must not be null!");
Collection<Set<byte[]>> keysPerNode = clusterCommandExecutor.executeCommandOnAllNodes(
new JedisClusterCommandCallback<Set<byte[]>>() {
Collection<Set<byte[]>> keysPerNode = clusterCommandExecutor
.executeCommandOnAllNodes(new JedisClusterCommandCallback<Set<byte[]>>() {
@Override
public Set<byte[]> doInCluster(Jedis client) {
return client.keys(pattern);
}
}).values();
}).resultsAsList();
Set<byte[]> keys = new HashSet<byte[]>();
for (Set<byte[]> keySet : keysPerNode) {
@@ -232,7 +233,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
public Set<byte[]> doInCluster(Jedis client) {
return client.keys(pattern);
}
}, node);
}, node).getValue();
}
/*
@@ -251,8 +252,8 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public byte[] randomKey() {
List<RedisClusterNode> nodes = new ArrayList<RedisClusterNode>(topologyProvider.getTopology()
.getActiveMasterNodes());
List<RedisClusterNode> nodes = new ArrayList<RedisClusterNode>(
topologyProvider.getTopology().getActiveMasterNodes());
Set<RedisNode> inspectedNodes = new HashSet<RedisNode>(nodes.size());
do {
@@ -286,7 +287,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
public byte[] doInCluster(Jedis client) {
return client.randomBinaryKey();
}
}, node);
}, node).getValue();
}
/*
@@ -451,7 +452,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
public Long doInCluster(Jedis client) {
return client.pttl(key);
}
}, topologyProvider.getTopology().getKeyServingMasterNode(key));
}, topologyProvider.getTopology().getKeyServingMasterNode(key)).getValue();
}
/*
@@ -507,7 +508,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
public byte[] doInCluster(Jedis client) {
return client.dump(key);
}
}, topologyProvider.getTopology().getKeyServingMasterNode(key));
}, topologyProvider.getTopology().getKeyServingMasterNode(key)).getValue();
}
/*
@@ -571,16 +572,13 @@ public class JedisClusterConnection implements RedisClusterConnection {
return cluster.mget(keys);
}
Map<RedisClusterNode, byte[]> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand(
new JedisMultiKeyClusterCommandCallback<byte[]>() {
return this.clusterCommandExecutor.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback<byte[]>() {
@Override
public byte[] doInCluster(Jedis client, byte[] key) {
return client.get(key);
}
}, Arrays.asList(keys));
return new ArrayList<byte[]>(nodeResult.values());
@Override
public byte[] doInCluster(Jedis client, byte[] key) {
return client.get(key);
}
}, Arrays.asList(keys)).resultsAsListSortBy(keys);
}
/*
@@ -1120,22 +1118,13 @@ public class JedisClusterConnection implements RedisClusterConnection {
}
}
Map<RedisClusterNode, List<byte[]>> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand(
new JedisMultiKeyClusterCommandCallback<List<byte[]>>() {
return this.clusterCommandExecutor.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback<List<byte[]>>() {
@Override
public List<byte[]> doInCluster(Jedis client, byte[] key) {
return client.blpop(timeout, key);
}
}, Arrays.asList(keys));
for (List<byte[]> partial : nodeResult.values()) {
if (!partial.isEmpty()) {
return partial;
@Override
public List<byte[]> doInCluster(Jedis client, byte[] key) {
return client.blpop(timeout, key);
}
}
return Collections.emptyList();
}, Arrays.asList(keys)).getFirstNonNullNotEmptyOrDefault(Collections.<byte[]> emptyList());
}
/*
@@ -1145,22 +1134,13 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public List<byte[]> bRPop(final int timeout, byte[]... keys) {
Map<RedisClusterNode, List<byte[]>> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand(
new JedisMultiKeyClusterCommandCallback<List<byte[]>>() {
return this.clusterCommandExecutor.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback<List<byte[]>>() {
@Override
public List<byte[]> doInCluster(Jedis client, byte[] key) {
return client.brpop(timeout, key);
}
}, Arrays.asList(keys));
for (List<byte[]> partial : nodeResult.values()) {
if (!partial.isEmpty()) {
return partial;
@Override
public List<byte[]> doInCluster(Jedis client, byte[] key) {
return client.brpop(timeout, key);
}
}
return Collections.emptyList();
}, Arrays.asList(keys)).getFirstNonNullNotEmptyOrDefault(Collections.<byte[]> emptyList());
}
/*
@@ -1314,19 +1294,20 @@ public class JedisClusterConnection implements RedisClusterConnection {
}
}
Map<RedisClusterNode, Set<byte[]>> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand(
new JedisMultiKeyClusterCommandCallback<Set<byte[]>>() {
Collection<Set<byte[]>> resultList = this.clusterCommandExecutor
.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback<Set<byte[]>>() {
@Override
public Set<byte[]> doInCluster(Jedis client, byte[] key) {
return client.smembers(key);
}
}, Arrays.asList(keys));
}, Arrays.asList(keys)).resultsAsList();
ByteArraySet result = null;
for (Entry<RedisClusterNode, Set<byte[]>> entry : nodeResult.entrySet()) {
ByteArraySet tmp = new ByteArraySet(entry.getValue());
for (Set<byte[]> value : resultList) {
ByteArraySet tmp = new ByteArraySet(value);
if (result == null) {
result = tmp;
} else {
@@ -1383,18 +1364,18 @@ public class JedisClusterConnection implements RedisClusterConnection {
}
}
Map<RedisClusterNode, Set<byte[]>> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand(
new JedisMultiKeyClusterCommandCallback<Set<byte[]>>() {
Collection<Set<byte[]>> resultList = this.clusterCommandExecutor
.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback<Set<byte[]>>() {
@Override
public Set<byte[]> doInCluster(Jedis client, byte[] key) {
return client.smembers(key);
}
}, Arrays.asList(keys));
}, Arrays.asList(keys)).resultsAsList();
ByteArraySet result = new ByteArraySet();
for (Entry<RedisClusterNode, Set<byte[]>> entry : nodeResult.entrySet()) {
result.addAll(entry.getValue());
for (Set<byte[]> entry : resultList) {
result.addAll(entry);
}
if (result.isEmpty()) {
@@ -1447,21 +1428,21 @@ public class JedisClusterConnection implements RedisClusterConnection {
byte[][] others = Arrays.copyOfRange(keys, 1, keys.length - 1);
ByteArraySet values = new ByteArraySet(sMembers(source));
Map<RedisClusterNode, Set<byte[]>> nodeResult = clusterCommandExecutor.executeMuliKeyCommand(
new JedisMultiKeyClusterCommandCallback<Set<byte[]>>() {
Collection<Set<byte[]>> resultList = clusterCommandExecutor
.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback<Set<byte[]>>() {
@Override
public Set<byte[]> doInCluster(Jedis client, byte[] key) {
return client.smembers(key);
}
}, Arrays.asList(others));
}, Arrays.asList(others)).resultsAsList();
if (values.isEmpty()) {
return Collections.emptySet();
}
for (Set<byte[]> toSubstract : nodeResult.values()) {
values.removeAll(toSubstract);
for (Set<byte[]> singleNodeValue : resultList) {
values.removeAll(singleNodeValue);
}
return values.asRawSet();
@@ -1552,8 +1533,8 @@ public class JedisClusterConnection implements RedisClusterConnection {
redis.clients.jedis.ScanResult<String> result = cluster.sscan(JedisConverters.toString(key),
Long.toString(cursorId));
return new ScanIteration<byte[]>(Long.valueOf(result.getCursor()), JedisConverters.stringListToByteList()
.convert(result.getResult()));
return new ScanIteration<byte[]>(Long.valueOf(result.getCursor()),
JedisConverters.stringListToByteList().convert(result.getResult()));
}
}.open();
}
@@ -1672,8 +1653,8 @@ public class JedisClusterConnection implements RedisClusterConnection {
try {
if (limit != null) {
return JedisConverters.toTupleSet(cluster.zrangeByScoreWithScores(key, min, max, limit.getOffset(),
limit.getCount()));
return JedisConverters
.toTupleSet(cluster.zrangeByScoreWithScores(key, min, max, limit.getOffset(), limit.getCount()));
}
return JedisConverters.toTupleSet(cluster.zrangeByScoreWithScores(key, min, max));
} catch (Exception ex) {
@@ -1734,8 +1715,8 @@ public class JedisClusterConnection implements RedisClusterConnection {
try {
if (limit != null) {
return JedisConverters.toTupleSet(cluster.zrevrangeByScoreWithScores(key, max, min, limit.getOffset(),
limit.getCount()));
return JedisConverters
.toTupleSet(cluster.zrevrangeByScoreWithScores(key, max, min, limit.getOffset(), limit.getCount()));
}
return JedisConverters.toTupleSet(cluster.zrevrangeByScoreWithScores(key, max, min));
} catch (Exception ex) {
@@ -2018,8 +1999,8 @@ public class JedisClusterConnection implements RedisClusterConnection {
}
try {
return JedisConverters.toTupleSet(cluster.zrevrangeByScoreWithScores(key, max, min, Long.valueOf(offset)
.intValue(), Long.valueOf(count).intValue()));
return JedisConverters.toTupleSet(cluster.zrevrangeByScoreWithScores(key, max, min,
Long.valueOf(offset).intValue(), Long.valueOf(count).intValue()));
} catch (Exception ex) {
throw convertJedisAccessException(ex);
}
@@ -2206,8 +2187,8 @@ public class JedisClusterConnection implements RedisClusterConnection {
}
try {
return cluster.zrangeByScore(key, JedisConverters.toBytes(min), JedisConverters.toBytes(max), Long
.valueOf(offset).intValue(), Long.valueOf(count).intValue());
return cluster.zrangeByScore(key, JedisConverters.toBytes(min), JedisConverters.toBytes(max),
Long.valueOf(offset).intValue(), Long.valueOf(count).intValue());
} catch (Exception ex) {
throw convertJedisAccessException(ex);
}
@@ -2553,7 +2534,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
public String doInCluster(Jedis client) {
return client.ping();
}
}).isEmpty() ? "PONG" : null;
}).resultsAsList().isEmpty() ? "PONG" : null;
}
@@ -2570,7 +2551,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
public String doInCluster(Jedis client) {
return client.ping();
}
}, node);
}, node).getValue();
}
/*
@@ -2660,14 +2641,14 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public Long lastSave() {
List<Long> result = new ArrayList<Long>(clusterCommandExecutor.executeCommandOnAllNodes(
new JedisClusterCommandCallback<Long>() {
List<Long> result = new ArrayList<Long>(
clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback<Long>() {
@Override
public Long doInCluster(Jedis client) {
return client.lastsave();
}
}).values());
}).resultsAsList());
if (CollectionUtils.isEmpty(result)) {
return null;
@@ -2690,7 +2671,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
public Long doInCluster(Jedis client) {
return client.lastsave();
}
}, node);
}, node).getValue();
}
/*
@@ -2732,21 +2713,20 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public Long dbSize() {
Map<RedisClusterNode, Long> dbSizes = clusterCommandExecutor
.executeCommandOnAllNodes(new JedisClusterCommandCallback<Long>() {
Collection<Long> dbSizes = clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback<Long>() {
@Override
public Long doInCluster(Jedis client) {
return client.dbSize();
}
});
@Override
public Long doInCluster(Jedis client) {
return client.dbSize();
}
}).resultsAsList();
if (CollectionUtils.isEmpty(dbSizes)) {
return 0L;
}
Long size = 0L;
for (Long value : dbSizes.values()) {
for (Long value : dbSizes) {
size += value;
}
return size;
@@ -2765,7 +2745,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
public Long doInCluster(Jedis client) {
return client.dbSize();
}
}, node);
}, node).getValue();
}
/*
@@ -2841,13 +2821,20 @@ public class JedisClusterConnection implements RedisClusterConnection {
Properties infos = new Properties();
infos.putAll(clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback<Properties>() {
List<NodeResult<Properties>> nodeResults = clusterCommandExecutor
.executeCommandOnAllNodes(new JedisClusterCommandCallback<Properties>() {
@Override
public Properties doInCluster(Jedis client) {
return JedisConverters.toProperties(client.info());
@Override
public Properties doInCluster(Jedis client) {
return JedisConverters.toProperties(client.info());
}
}).getResults();
for (NodeResult<Properties> nodePorperties : nodeResults) {
for (Entry<Object, Object> entry : nodePorperties.getValue().entrySet()) {
infos.put(nodePorperties.getNode().asString() + "." + entry.getKey(), entry.getValue());
}
}));
}
return infos;
}
@@ -2859,14 +2846,14 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public Properties info(RedisClusterNode node) {
return JedisConverters.toProperties(clusterCommandExecutor.executeCommandOnSingleNode(
new JedisClusterCommandCallback<String>() {
return JedisConverters
.toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback<String>() {
@Override
public String doInCluster(Jedis client) {
return client.info();
}
}, node));
}, node).getValue());
}
/*
@@ -2878,13 +2865,20 @@ public class JedisClusterConnection implements RedisClusterConnection {
Properties infos = new Properties();
infos.putAll(clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback<Properties>() {
List<NodeResult<Properties>> nodeResults = clusterCommandExecutor
.executeCommandOnAllNodes(new JedisClusterCommandCallback<Properties>() {
@Override
public Properties doInCluster(Jedis client) {
return JedisConverters.toProperties(client.info(section));
@Override
public Properties doInCluster(Jedis client) {
return JedisConverters.toProperties(client.info(section));
}
}).getResults();
for (NodeResult<Properties> nodePorperties : nodeResults) {
for (Entry<Object, Object> entry : nodePorperties.getValue().entrySet()) {
infos.put(nodePorperties.getNode().asString() + "." + entry.getKey(), entry.getValue());
}
}));
}
return infos;
}
@@ -2896,14 +2890,14 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public Properties info(RedisClusterNode node, final String section) {
return JedisConverters.toProperties(clusterCommandExecutor.executeCommandOnSingleNode(
new JedisClusterCommandCallback<String>() {
return JedisConverters
.toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback<String>() {
@Override
public String doInCluster(Jedis client) {
return client.info(section);
}
}, node));
}, node).getValue());
}
/*
@@ -2961,19 +2955,19 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public List<String> getConfig(final String pattern) {
Map<RedisClusterNode, List<String>> mapResult = clusterCommandExecutor
List<NodeResult<List<String>>> mapResult = clusterCommandExecutor
.executeCommandOnAllNodes(new JedisClusterCommandCallback<List<String>>() {
@Override
public List<String> doInCluster(Jedis client) {
return client.configGet(pattern);
}
});
}).getResults();
List<String> result = new ArrayList<String>();
for (Entry<RedisClusterNode, List<String>> entry : mapResult.entrySet()) {
for (NodeResult<List<String>> entry : mapResult) {
String prefix = entry.getKey().asString();
String prefix = entry.getNode().asString();
int i = 0;
for (String value : entry.getValue()) {
result.add((i++ % 2 == 0 ? (prefix + ".") : "") + value);
@@ -2996,7 +2990,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
public List<String> doInCluster(Jedis client) {
return client.configGet(pattern);
}
}, node);
}, node).getValue();
}
/*
@@ -3070,14 +3064,14 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public Long time() {
return convertListOfStringToTime(clusterCommandExecutor
.executeCommandOnArbitraryNode(new JedisClusterCommandCallback<List<String>>() {
return convertListOfStringToTime(
clusterCommandExecutor.executeCommandOnArbitraryNode(new JedisClusterCommandCallback<List<String>>() {
@Override
public List<String> doInCluster(Jedis client) {
return client.time();
}
}));
}).getValue());
}
/*
@@ -3087,14 +3081,14 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public Long time(RedisClusterNode node) {
return convertListOfStringToTime(clusterCommandExecutor.executeCommandOnSingleNode(
new JedisClusterCommandCallback<List<String>>() {
return convertListOfStringToTime(
clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback<List<String>>() {
@Override
public List<String> doInCluster(Jedis client) {
return client.time();
}
}, node));
}, node).getValue());
}
private Long convertListOfStringToTime(List<String> serverTimeInformation) {
@@ -3149,17 +3143,16 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public List<RedisClientInfo> getClientList() {
Map<RedisClusterNode, String> map = clusterCommandExecutor
.executeCommandOnAllNodes(new JedisClusterCommandCallback<String>() {
Collection<String> map = clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback<String>() {
@Override
public String doInCluster(Jedis client) {
return client.clientList();
}
});
@Override
public String doInCluster(Jedis client) {
return client.clientList();
}
}).resultsAsList();
ArrayList<RedisClientInfo> result = new ArrayList<RedisClientInfo>();
for (String infos : map.values()) {
for (String infos : map) {
result.addAll(JedisConverters.toListOfRedisClientInformation(infos));
}
return result;
@@ -3172,14 +3165,14 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public List<RedisClientInfo> getClientList(RedisClusterNode node) {
return JedisConverters.toListOfRedisClientInformation(clusterCommandExecutor.executeCommandOnSingleNode(
new JedisClusterCommandCallback<String>() {
return JedisConverters.toListOfRedisClientInformation(
clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback<String>() {
@Override
public String doInCluster(Jedis client) {
return client.clientList();
}
}, node));
}, node).getValue());
}
/*
@@ -3385,8 +3378,8 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public List<byte[]> doInCluster(Jedis client) {
return JedisConverters.stringListToByteList().convert(
client.clusterGetKeysInSlot(slot, count != null ? count.intValue() : Integer.MAX_VALUE));
return JedisConverters.stringListToByteList()
.convert(client.clusterGetKeysInSlot(slot, count != null ? count.intValue() : Integer.MAX_VALUE));
}
}, node);
return null;
@@ -3438,7 +3431,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
return client.clusterCountKeysInSlot(slot);
}
}, node);
}, node).getValue();
}
@@ -3478,8 +3471,8 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public void clusterForget(final RedisClusterNode node) {
Set<RedisClusterNode> nodes = new LinkedHashSet<RedisClusterNode>(topologyProvider.getTopology()
.getActiveMasterNodes());
Set<RedisClusterNode> nodes = new LinkedHashSet<RedisClusterNode>(
topologyProvider.getTopology().getActiveMasterNodes());
final RedisClusterNode nodeToRemove = topologyProvider.getTopology().lookup(node);
nodes.remove(nodeToRemove);
@@ -3546,7 +3539,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
public Integer doInCluster(Jedis client) {
return client.clusterKeySlot(JedisConverters.toString(key)).intValue();
}
});
}).getValue();
}
/*
@@ -3585,14 +3578,14 @@ public class JedisClusterConnection implements RedisClusterConnection {
final RedisClusterNode nodeToUse = topologyProvider.getTopology().lookup(master);
return JedisConverters.toSetOfRedisClusterNodes(clusterCommandExecutor.executeCommandOnSingleNode(
new JedisClusterCommandCallback<List<String>>() {
return JedisConverters.toSetOfRedisClusterNodes(
clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback<List<String>>() {
@Override
public List<String> doInCluster(Jedis client) {
return client.clusterSlaves(nodeToUse.getId());
}
}, master));
}, master).getValue());
}
/*
@@ -3601,17 +3594,25 @@ public class JedisClusterConnection implements RedisClusterConnection {
*/
public Map<RedisClusterNode, Collection<RedisClusterNode>> clusterGetMasterSlaveMap() {
return clusterCommandExecutor.executeCommandAsyncOnNodes(
new JedisClusterCommandCallback<Collection<RedisClusterNode>>() {
List<NodeResult<Collection<RedisClusterNode>>> nodeResults = clusterCommandExecutor
.executeCommandAsyncOnNodes(new JedisClusterCommandCallback<Collection<RedisClusterNode>>() {
@Override
public Set<RedisClusterNode> doInCluster(Jedis client) {
// TODO: remove client.eval as soon as Jedis offers support for myid
return JedisConverters.toSetOfRedisClusterNodes(client.clusterSlaves((String) client.eval(
"return redis.call('cluster', 'myid')", 0)));
return JedisConverters.toSetOfRedisClusterNodes(
client.clusterSlaves((String) client.eval("return redis.call('cluster', 'myid')", 0)));
}
}, topologyProvider.getTopology().getActiveMasterNodes());
}, topologyProvider.getTopology().getActiveMasterNodes()).getResults();
Map<RedisClusterNode, Collection<RedisClusterNode>> result = new LinkedHashMap<RedisClusterNode, Collection<RedisClusterNode>>();
for (NodeResult<Collection<RedisClusterNode>> nodeResult : nodeResults) {
result.put(nodeResult.getNode(), nodeResult.getValue());
}
return result;
}
/*
@@ -3630,14 +3631,14 @@ public class JedisClusterConnection implements RedisClusterConnection {
@Override
public ClusterInfo clusterGetClusterInfo() {
return new ClusterInfo(JedisConverters.toProperties(clusterCommandExecutor
.executeCommandOnArbitraryNode(new JedisClusterCommandCallback<String>() {
return new ClusterInfo(JedisConverters
.toProperties(clusterCommandExecutor.executeCommandOnArbitraryNode(new JedisClusterCommandCallback<String>() {
@Override
public String doInCluster(Jedis client) {
return client.clusterInfo();
}
})));
}).getValue()));
}
/*

238
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java
View File

@@ -20,6 +20,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -36,6 +37,7 @@ import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
import org.springframework.data.redis.connection.ClusterCommandExecutor;
import org.springframework.data.redis.connection.ClusterCommandExecutor.ClusterCommandCallback;
import org.springframework.data.redis.connection.ClusterCommandExecutor.MultiKeyClusterCommandCallback;
import org.springframework.data.redis.connection.ClusterCommandExecutor.NodeResult;
import org.springframework.data.redis.connection.ClusterInfo;
import org.springframework.data.redis.connection.ClusterNodeResourceProvider;
import org.springframework.data.redis.connection.ClusterSlotHashUtil;
@@ -69,8 +71,8 @@ import com.lambdaworks.redis.codec.RedisCodec;
* @author Mark Paluch
* @since 1.7
*/
public class LettuceClusterConnection extends LettuceConnection implements
org.springframework.data.redis.connection.RedisClusterConnection {
public class LettuceClusterConnection extends LettuceConnection
implements org.springframework.data.redis.connection.RedisClusterConnection {
static final ExceptionTranslationStrategy exceptionConverter = new PassThroughExceptionTranslationStrategy(
new LettuceExceptionConverter());
@@ -93,8 +95,8 @@ public class LettuceClusterConnection extends LettuceConnection implements
this.clusterClient = clusterClient;
topologyProvider = new LettuceClusterTopologyProvider(clusterClient);
clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, new LettuceClusterNodeResourceProvider(
clusterClient), exceptionConverter);
clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider,
new LettuceClusterNodeResourceProvider(clusterClient), exceptionConverter);
}
/**
@@ -133,14 +135,14 @@ public class LettuceClusterConnection extends LettuceConnection implements
Assert.notNull(pattern, "Pattern must not be null!");
Collection<List<byte[]>> keysPerNode = clusterCommandExecutor.executeCommandOnAllNodes(
new LettuceClusterCommandCallback<List<byte[]>>() {
Collection<List<byte[]>> keysPerNode = clusterCommandExecutor
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<List<byte[]>>() {
@Override
public List<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> connection) {
return connection.keys(pattern);
}
}).values();
}).resultsAsList();
Set<byte[]> keys = new HashSet<byte[]>();
@@ -188,7 +190,7 @@ public class LettuceClusterConnection extends LettuceConnection implements
@Override
public Long dbSize() {
Map<RedisClusterNode, Long> dbSizes = clusterCommandExecutor
Collection<Long> dbSizes = clusterCommandExecutor
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<Long>() {
@Override
@@ -196,14 +198,14 @@ public class LettuceClusterConnection extends LettuceConnection implements
return client.dbsize();
}
});
}).resultsAsList();
if (CollectionUtils.isEmpty(dbSizes)) {
return 0L;
}
Long size = 0L;
for (Long value : dbSizes.values()) {
for (Long value : dbSizes) {
size += value;
}
return size;
@@ -218,13 +220,20 @@ public class LettuceClusterConnection extends LettuceConnection implements
Properties infos = new Properties();
infos.putAll(clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback<Properties>() {
List<NodeResult<Properties>> nodeResults = clusterCommandExecutor
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<Properties>() {
@Override
public Properties doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return LettuceConverters.toProperties(client.info());
@Override
public Properties doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return LettuceConverters.toProperties(client.info());
}
}).getResults();
for (NodeResult<Properties> nodePorperties : nodeResults) {
for (Entry<Object, Object> entry : nodePorperties.getValue().entrySet()) {
infos.put(nodePorperties.getNode().asString() + "." + entry.getKey(), entry.getValue());
}
}));
}
return infos;
}
@@ -233,14 +242,20 @@ public class LettuceClusterConnection extends LettuceConnection implements
public Properties info(final String section) {
Properties infos = new Properties();
List<NodeResult<Properties>> nodeResults = clusterCommandExecutor
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<Properties>() {
infos.putAll(clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback<Properties>() {
@Override
public Properties doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return LettuceConverters.toProperties(client.info(section));
}
}).getResults();
@Override
public Properties doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return LettuceConverters.toProperties(client.info(section));
for (NodeResult<Properties> nodePorperties : nodeResults) {
for (Entry<Object, Object> entry : nodePorperties.getValue().entrySet()) {
infos.put(nodePorperties.getNode().asString() + "." + entry.getKey(), entry.getValue());
}
}));
}
return infos;
}
@@ -252,14 +267,14 @@ public class LettuceClusterConnection extends LettuceConnection implements
@Override
public Properties info(RedisClusterNode node, final String section) {
return LettuceConverters.toProperties(clusterCommandExecutor.executeCommandOnSingleNode(
new LettuceClusterCommandCallback<String>() {
return LettuceConverters
.toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.info(section);
}
}, node));
}, node).getValue());
}
/*
@@ -301,17 +316,16 @@ public class LettuceClusterConnection extends LettuceConnection implements
Assert.notNull(master, "Master must not be null!");
final RedisClusterNode nodeToUse = topologyProvider.getTopology()
.lookup(master);
final RedisClusterNode nodeToUse = topologyProvider.getTopology().lookup(master);
return clusterCommandExecutor.executeCommandOnSingleNode(
new LettuceClusterCommandCallback<Set<RedisClusterNode>>() {
return clusterCommandExecutor
.executeCommandOnSingleNode(new LettuceClusterCommandCallback<Set<RedisClusterNode>>() {
@Override
public Set<RedisClusterNode> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return LettuceConverters.toSetOfRedisClusterNodes(client.clusterSlaves(nodeToUse.getId()));
}
}, master);
}, master).getValue();
}
/*
@@ -331,8 +345,8 @@ public class LettuceClusterConnection extends LettuceConnection implements
public RedisClusterNode clusterGetNodeForSlot(int slot) {
DirectFieldAccessor accessor = new DirectFieldAccessor(clusterClient);
return LettuceConverters.toRedisClusterNode(((Partitions) accessor.getPropertyValue("partitions"))
.getPartitionBySlot(slot));
return LettuceConverters
.toRedisClusterNode(((Partitions) accessor.getPropertyValue("partitions")).getPartitionBySlot(slot));
}
/*
@@ -357,7 +371,7 @@ public class LettuceClusterConnection extends LettuceConnection implements
public ClusterInfo doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return new ClusterInfo(LettuceConverters.toProperties(client.clusterInfo()));
}
});
}).getValue();
}
/*
@@ -542,14 +556,14 @@ public class LettuceClusterConnection extends LettuceConnection implements
*/
@Override
public String ping() {
Collection<String> ping = clusterCommandExecutor.executeCommandOnAllNodes(
new LettuceClusterCommandCallback<String>() {
Collection<String> ping = clusterCommandExecutor
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> connection) {
return doPing(connection);
}
}).values();
}).resultsAsList();
for (String result : ping) {
if (!ObjectUtils.nullSafeEquals("PONG", result)) {
@@ -573,7 +587,7 @@ public class LettuceClusterConnection extends LettuceConnection implements
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return doPing(client);
}
}, node);
}, node).getValue();
}
protected String doPing(RedisClusterConnection<byte[], byte[]> client) {
@@ -638,7 +652,7 @@ public class LettuceClusterConnection extends LettuceConnection implements
public Long doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.lastsave().getTime();
}
}, node);
}, node).getValue();
}
/*
@@ -671,7 +685,7 @@ public class LettuceClusterConnection extends LettuceConnection implements
public Long doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.dbsize();
}
}, node);
}, node).getValue();
}
/*
@@ -714,14 +728,14 @@ public class LettuceClusterConnection extends LettuceConnection implements
@Override
public Properties info(RedisClusterNode node) {
return LettuceConverters.toProperties(clusterCommandExecutor.executeCommandOnSingleNode(
new LettuceClusterCommandCallback<String>() {
return LettuceConverters
.toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.info();
}
}, node));
}, node).getValue());
}
/*
@@ -731,14 +745,14 @@ public class LettuceClusterConnection extends LettuceConnection implements
@Override
public Set<byte[]> keys(RedisClusterNode node, final byte[] pattern) {
return LettuceConverters.toBytesSet(clusterCommandExecutor.executeCommandOnSingleNode(
new LettuceClusterCommandCallback<List<byte[]>>() {
return LettuceConverters.toBytesSet(
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<List<byte[]>>() {
@Override
public List<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.keys(pattern);
}
}, node));
}, node).getValue());
}
/*
@@ -754,7 +768,7 @@ public class LettuceClusterConnection extends LettuceConnection implements
public byte[] doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.randomkey();
}
}, node);
}, node).getValue();
}
/*
@@ -886,16 +900,13 @@ public class LettuceClusterConnection extends LettuceConnection implements
return super.mGet(keys);
}
Map<RedisClusterNode, byte[]> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand(
new LettuceMultiKeyClusterCommandCallback<byte[]>() {
return this.clusterCommandExecutor.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback<byte[]>() {
@Override
public byte[] doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
return client.get(key);
}
}, Arrays.asList(keys));
return new ArrayList<byte[]>(nodeResult.values());
@Override
public byte[] doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
return client.get(key);
}
}, Arrays.asList(keys)).resultsAsListSortBy(keys);
}
/*
@@ -948,17 +959,19 @@ public class LettuceClusterConnection extends LettuceConnection implements
return super.bLPop(timeout, keys);
}
Map<RedisClusterNode, KeyValue<byte[], byte[]>> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand(
new LettuceMultiKeyClusterCommandCallback<KeyValue<byte[], byte[]>>() {
List<KeyValue<byte[], byte[]>> resultList = this.clusterCommandExecutor
.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback<KeyValue<byte[], byte[]>>() {
@Override
public KeyValue<byte[], byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
return client.blpop(timeout, key);
}
}, Arrays.asList(keys));
}, Arrays.asList(keys)).resultsAsList();
for (KeyValue<byte[], byte[]> partial : nodeResult.values()) {
return LettuceConverters.toBytesList(partial);
for (KeyValue<byte[], byte[]> kv : resultList) {
if (kv != null) {
return LettuceConverters.toBytesList(kv);
}
}
return Collections.emptyList();
@@ -975,17 +988,19 @@ public class LettuceClusterConnection extends LettuceConnection implements
return super.bRPop(timeout, keys);
}
Map<RedisClusterNode, KeyValue<byte[], byte[]>> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand(
new LettuceMultiKeyClusterCommandCallback<KeyValue<byte[], byte[]>>() {
List<KeyValue<byte[], byte[]>> resultList = this.clusterCommandExecutor
.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback<KeyValue<byte[], byte[]>>() {
@Override
public KeyValue<byte[], byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
return client.brpop(timeout, key);
}
}, Arrays.asList(keys));
}, Arrays.asList(keys)).resultsAsList();
for (KeyValue<byte[], byte[]> partial : nodeResult.values()) {
return LettuceConverters.toBytesList(partial);
for (KeyValue<byte[], byte[]> kv : resultList) {
if (kv != null) {
return LettuceConverters.toBytesList(kv);
}
}
return Collections.emptyList();
@@ -1057,19 +1072,19 @@ public class LettuceClusterConnection extends LettuceConnection implements
return super.sInter(keys);
}
Map<RedisClusterNode, Set<byte[]>> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand(
new LettuceMultiKeyClusterCommandCallback<Set<byte[]>>() {
Collection<Set<byte[]>> nodeResult = this.clusterCommandExecutor
.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback<Set<byte[]>>() {
@Override
public Set<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
return client.smembers(key);
}
}, Arrays.asList(keys));
}, Arrays.asList(keys)).resultsAsList();
ByteArraySet result = null;
for (Entry<RedisClusterNode, Set<byte[]>> entry : nodeResult.entrySet()) {
for (Set<byte[]> entry : nodeResult) {
ByteArraySet tmp = new ByteArraySet(entry.getValue());
ByteArraySet tmp = new ByteArraySet(entry);
if (result == null) {
result = tmp;
} else {
@@ -1118,18 +1133,18 @@ public class LettuceClusterConnection extends LettuceConnection implements
return super.sUnion(keys);
}
Map<RedisClusterNode, Set<byte[]>> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand(
new LettuceMultiKeyClusterCommandCallback<Set<byte[]>>() {
Collection<Set<byte[]>> nodeResult = this.clusterCommandExecutor
.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback<Set<byte[]>>() {
@Override
public Set<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
return client.smembers(key);
}
}, Arrays.asList(keys));
}, Arrays.asList(keys)).resultsAsList();
ByteArraySet result = new ByteArraySet();
for (Entry<RedisClusterNode, Set<byte[]>> entry : nodeResult.entrySet()) {
result.addAll(entry.getValue());
for (Set<byte[]> entry : nodeResult) {
result.addAll(entry);
}
if (result.isEmpty()) {
@@ -1174,20 +1189,20 @@ public class LettuceClusterConnection extends LettuceConnection implements
byte[][] others = Arrays.copyOfRange(keys, 1, keys.length - 1);
ByteArraySet values = new ByteArraySet(sMembers(source));
Map<RedisClusterNode, Set<byte[]>> nodeResult = clusterCommandExecutor.executeMuliKeyCommand(
new LettuceMultiKeyClusterCommandCallback<Set<byte[]>>() {
Collection<Set<byte[]>> nodeResult = clusterCommandExecutor
.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback<Set<byte[]>>() {
@Override
public Set<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
return client.smembers(key);
}
}, Arrays.asList(others));
}, Arrays.asList(others)).resultsAsList();
if (values.isEmpty()) {
return Collections.emptySet();
}
for (Set<byte[]> toSubstract : nodeResult.values()) {
for (Set<byte[]> toSubstract : nodeResult) {
values.removeAll(toSubstract);
}
@@ -1310,19 +1325,19 @@ public class LettuceClusterConnection extends LettuceConnection implements
@Override
public List<String> getConfig(final String pattern) {
Map<RedisClusterNode, List<String>> mapResult = clusterCommandExecutor
List<NodeResult<List<String>>> mapResult = clusterCommandExecutor
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<List<String>>() {
@Override
public List<String> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.configGet(pattern);
}
});
}).getResults();
List<String> result = new ArrayList<String>();
for (Entry<RedisClusterNode, List<String>> entry : mapResult.entrySet()) {
for (NodeResult<List<String>> entry : mapResult) {
String prefix = entry.getKey().asString();
String prefix = entry.getNode().asString();
int i = 0;
for (String value : entry.getValue()) {
result.add((i++ % 2 == 0 ? (prefix + ".") : "") + value);
@@ -1345,7 +1360,7 @@ public class LettuceClusterConnection extends LettuceConnection implements
public List<String> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.configGet(pattern);
}
}, node);
}, node).getValue();
}
/*
@@ -1420,14 +1435,14 @@ public class LettuceClusterConnection extends LettuceConnection implements
@Override
public Long time() {
return convertListOfStringToTime(clusterCommandExecutor
.executeCommandOnArbitraryNode(new LettuceClusterCommandCallback<List<byte[]>>() {
return convertListOfStringToTime(
clusterCommandExecutor.executeCommandOnArbitraryNode(new LettuceClusterCommandCallback<List<byte[]>>() {
@Override
public List<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.time();
}
}));
}).getValue());
}
/*
@@ -1437,14 +1452,14 @@ public class LettuceClusterConnection extends LettuceConnection implements
@Override
public Long time(RedisClusterNode node) {
return convertListOfStringToTime(clusterCommandExecutor.executeCommandOnSingleNode(
new LettuceClusterCommandCallback<List<byte[]>>() {
return convertListOfStringToTime(
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<List<byte[]>>() {
@Override
public List<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.time();
}
}, node));
}, node).getValue());
}
private Long convertListOfStringToTime(List<byte[]> serverTimeInformation) {
@@ -1464,17 +1479,16 @@ public class LettuceClusterConnection extends LettuceConnection implements
@Override
public List<RedisClientInfo> getClientList() {
Map<RedisClusterNode, String> map = clusterCommandExecutor
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<String>() {
List<String> map = clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.clientList();
}
});
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.clientList();
}
}).resultsAsList();
ArrayList<RedisClientInfo> result = new ArrayList<RedisClientInfo>();
for (String infos : map.values()) {
for (String infos : map) {
result.addAll(LettuceConverters.toListOfRedisClientInformation(infos));
}
return result;
@@ -1487,14 +1501,14 @@ public class LettuceClusterConnection extends LettuceConnection implements
@Override
public List<RedisClientInfo> getClientList(RedisClusterNode node) {
return LettuceConverters.toListOfRedisClientInformation(clusterCommandExecutor.executeCommandOnSingleNode(
new LettuceClusterCommandCallback<String>() {
return LettuceConverters.toListOfRedisClientInformation(
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return client.clientList();
}
}, node));
}, node).getValue());
}
/*
@@ -1504,14 +1518,22 @@ public class LettuceClusterConnection extends LettuceConnection implements
@Override
public Map<RedisClusterNode, Collection<RedisClusterNode>> clusterGetMasterSlaveMap() {
return clusterCommandExecutor.executeCommandAsyncOnNodes(
new LettuceClusterCommandCallback<Collection<RedisClusterNode>>() {
List<NodeResult<Collection<RedisClusterNode>>> nodeResults = clusterCommandExecutor
.executeCommandAsyncOnNodes(new LettuceClusterCommandCallback<Collection<RedisClusterNode>>() {
@Override
public Set<RedisClusterNode> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return Converters.toSetOfRedisClusterNodes(client.clusterSlaves(client.clusterMyId()));
}
}, topologyProvider.getTopology().getActiveMasterNodes());
}, topologyProvider.getTopology().getActiveMasterNodes()).getResults();
Map<RedisClusterNode, Collection<RedisClusterNode>> result = new LinkedHashMap<RedisClusterNode, Collection<RedisClusterNode>>();
for (NodeResult<Collection<RedisClusterNode>> nodeResult : nodeResults) {
result.put(nodeResult.getNode(), nodeResult.getValue());
}
return result;
}
/**
@@ -1521,8 +1543,8 @@ public class LettuceClusterConnection extends LettuceConnection implements
* @param <T>
* @since 1.7
*/
protected interface LettuceClusterCommandCallback<T> extends
ClusterCommandCallback<RedisClusterConnection<byte[], byte[]>, T> {}
protected interface LettuceClusterCommandCallback<T>
extends ClusterCommandCallback<RedisClusterConnection<byte[], byte[]>, T> {}
/**
* Lettuce specific implementation of {@link MultiKeyClusterCommandCallback}.
@@ -1531,8 +1553,8 @@ public class LettuceClusterConnection extends LettuceConnection implements
* @param <T>
* @since 1.7
*/
protected interface LettuceMultiKeyClusterCommandCallback<T> extends
MultiKeyClusterCommandCallback<RedisClusterConnection<byte[], byte[]>, T> {
protected interface LettuceMultiKeyClusterCommandCallback<T>
extends MultiKeyClusterCommandCallback<RedisClusterConnection<byte[], byte[]>, T> {
}
@@ -1606,8 +1628,8 @@ public class LettuceClusterConnection extends LettuceConnection implements
*/
@Override
public ClusterTopology getTopology() {
return new ClusterTopology(new LinkedHashSet<RedisClusterNode>(LettuceConverters.partitionsToClusterNodes(client
.getPartitions())));
return new ClusterTopology(
new LinkedHashSet<RedisClusterNode>(LettuceConverters.partitionsToClusterNodes(client.getPartitions())));
}
}

2
src/main/java/org/springframework/data/redis/core/RedisTemplate.java
View File

@@ -722,7 +722,7 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
}
}, true);
return keySerializer != null ? SerializationUtils.deserialize(rawKeys, keySerializer) : rawKeys;
return keySerializer != null ? SerializationUtils.deserialize(rawKeys, keySerializer) : (Set<K>) rawKeys;
}
public Boolean persist(K key) {