diff --git a/pom.xml b/pom.xml
index 137d6a729..281705c37 100644
--- a/pom.xml
+++ b/pom.xml
@@ -24,7 +24,7 @@
1.9.21.4.82.2
- 3.3.1.Final
+ 3.4.Final2.8.00.706052013
diff --git a/src/asciidoc/reference/redis-cluster.adoc b/src/asciidoc/reference/redis-cluster.adoc
new file mode 100644
index 000000000..281458962
--- /dev/null
+++ b/src/asciidoc/reference/redis-cluster.adoc
@@ -0,0 +1,136 @@
+[[cluster]]
+= Redis Cluster
+
+Working with http://redis.io/topics/cluster-spec[Redis Cluster] requires a Redis Server version 3.0+ and provides a very own set of features and capabilities. Please refer to the http://redis.io/topics/cluster-tutorial[Cluster Tutorial] for more information.
+
+TIP: Redis Cluster is only supported by <> and <>.
+
+== Enabling Redis Cluster
+
+Cluster support is based on the very same building blocks as non clustered communication. `RedisClusterConnection` and extension to `RedisConnection` handles the communication with the Redis Cluster and translates errors into the Spring DAO exception hierarchy.
+`RedisClusterConnection` 's are created via the `RedisConnectionFactory` which has to be set up with the according `RedisClusterConfiguration`.
+
+.Sample RedisConnectionFactory Configuration for Redis Cluster
+====
+[source,java]
+----
+@Configuration
+public class AppConfig {
+
+ /*
+ * spring.redis.cluster.nodes[0] = 127.0.0.1:7379
+ * spring.redis.cluster.nodes[1] = 127.0.0.1:7380
+ * ...
+ */
+ @Value("${spring.redis.cluster.nodes}")
+ private Collection initialClusterNodes;
+
+ public @Bean RedisConnectionFactory connectionFactory() {
+
+ return new JedisConnectionFactory(
+ new RedisClusterConfiguration(initialClusterNodes));
+ }
+}
+----
+====
+
+NOTE: The initial configuration points driver libraries to an initial set of cluster nodes. Changes resulting from live cluster reconfiguration will only be kept in the native driver and not be written back to the configuration.
+
+== Working With Redis Cluster Connection
+
+As mentioned above Redis Cluster behaves different from single node Redis or even a Sentinel monitored master slave environment. This is reasoned by the automatic sharding that maps a key to one of 16384 slots which are distributed across the nodes. Therefore commands that involve more than one key must assert that all keys map to the exact same slot in order to avoid cross slot execution errors.
+Further on, hence a single cluster node, only servers a dedicated set of keys, commands issued against one particular server only return results for those keys served by the server. As a very simple example take the `KEYS` command. When issued to a server in cluster environment it only returns the keys served by the node the request is sent to and not necessarily all keys within the cluster. So to get all keys in cluster environment it is necessary to read the keys from at least all known master nodes.
+
+While redirects for to a specific keys to the corresponding slot serving node are handled by the driver libraries, higher level functions like collecting information across nodes, or sending commands to all nodes in the cluster that are covered by `RedisClusterConnection`. Picking up the keys example from just before, this means, that the `keys(pattern)` method picks up every master node in cluster and simultaneously executes the `KEYS` command on every single one, while picking up the results and returning the cumulated set of keys. To just request the keys of a single node `RedisClusterConnection` provides overloads for those (like `keys(node, pattern)` ).
+
+.Sample of Running Commands Across the Cluster
+====
+[source,text]
+----
+redis-cli@127.0.0.1:7379 > cluster nodes
+
+6b38bb... 127.0.0.1:7379 master - 0 0 25 connected 0-5460 <1>
+7bb78c... 127.0.0.1:7380 master - 0 1449730618304 2 connected 5461-10922 <2>
+164888... 127.0.0.1:7381 master - 0 1449730618304 3 connected 10923-16383 <3>
+b8b5ee... 127.0.0.1:7382 slave 6b38bb... 0 1449730618304 25 connected <4>
+----
+
+[source,java]
+----
+RedisClusterConnection connection = connectionFactory.getClusterConnnection();
+
+connection.set("foo", value); <5>
+connection.set("bar", value); <6>
+
+connection.keys("*"); <7>
+
+connection.keys(NODE_7379, "*"); <8>
+connection.keys(NODE_7380, "*"); <9>
+connection.keys(NODE_7381, "*"); <10>
+connection.keys(NODE_7382, "*"); <11>
+----
+<1> Master node serving slots 0 to 5460 replicated to slave at 7382
+<2> Master node serving slots 5461 to 10922
+<3> Master node serving slots 10923 to 16383
+<4> Slave node holding replicates of master at 7379
+<5> Request routed to node at 7381 serving slot 12182
+<6> Request routed to node at 7379 serving slot 5061
+<7> Request routed to nodes at 7379, 7380, 7381 -> [foo, bar]
+<8> Request routed to node at 7379 -> [bar]
+<9> Request routed to node at 7380 -> []
+<10> Request routed to node at 7381 -> [foo]
+<11> Request routed to node at 7382 -> [bar]
+====
+
+Cross slot requests such as `MGET` are automatically served by the native driver library when all keys map to the same slot. However once this is not the case `RedisClusterConnection` executes multiple parallel `GET` commands against the slot serving nodes and again returns a cumulated result. Obviously this is less performing than the single slot execution and therefore should be used with care. In doubt please consider pinning keys to the same slot by providing a prefix in curly brackets like `{my-prefix}.foo` and `{my-prefix}.bar` which will both map to the same slot number.
+
+.Sample of Cross Slot Request Handling
+====
+[source,text]
+----
+redis-cli@127.0.0.1:7379 > cluster nodes
+
+6b38bb... 127.0.0.1:7379 master - 0 0 25 connected 0-5460 <1>
+7bb...
+----
+
+[source,java]
+----
+RedisClusterConnection connection = connectionFactory.getClusterConnnection();
+
+connection.set("foo", value); // slot: 12182
+connection.set("{foo}.bar", value); // slot: 12182
+connection.set("bar", value); // slot: 5461
+
+connection.mGet("foo", "{foo}.bar"); <2>
+
+connection.mGet("foo", "bar"); <3>
+----
+<1> Same Configuration as in the sample before.
+<2> Keys map to same slot -> 127.0.0.1:7381 MGET foo {foo}.bar
+<3> Keys map to different slots and get split up into single slot ones routed to the according nodes +
+ -> 127.0.0.1:7379 GET bar +
+ -> 127.0.0.1:7381 GET foo
+====
+
+TIP: The above provides simple examples to demonstrate the general strategy followed by Spring Data Redis. Be aware that some operations might require loading huge amounts of data into memory in order to compute the desired command. Additionally not all cross slot requests can safely be ported to multiple single slot requests and will error if misused (eg. `PFCOUNT` ).
+
+== Working With RedisTemplate and ClusterOperations
+
+Please refer to the section <> to read about general purpose, configuration and usage of `RedisTemplate`.
+
+WARNING: Please be careful when setting up `RedisTemplate#keySerializer` using any of the Json `RedisSerializers` as changing json structure has immediate influence on hash slot calculation.
+
+`RedisTemplate` provides access to cluster specific operations via the `ClusterOperations` interface that can be obtained via `RedisTemplate.opsForCluster()`. This allows to execute commands explicitly on a single node within the cluster while retaining de-/serialization features configured for the template and provides administrative commands such as `CLUSTER MEET` or more high level operations for eg. resharding.
+
+
+.Accessing RedisClusterConnection via RedisTemplate
+====
+[source,text]
+----
+ClusterOperations clusterOps = redisTemplate.opsForCluster();
+clusterOps.shutdown(NODE_7379); <1>
+----
+<1> Shut down node at 7379 and cross fingers there is a slave in place that can take over.
+====
+
diff --git a/src/main/asciidoc/index.adoc b/src/main/asciidoc/index.adoc
index cade2e5b4..f3997a7f1 100644
--- a/src/main/asciidoc/index.adoc
+++ b/src/main/asciidoc/index.adoc
@@ -33,6 +33,7 @@ include::introduction/getting-started.adoc[]
:leveloffset: +1
include::reference/introduction.adoc[]
include::reference/redis.adoc[]
+include::reference/redis-cluster.adoc[]
:leveloffset: -1
[[appendixes]]
@@ -43,4 +44,4 @@ include::reference/redis.adoc[]
include::appendix/introduction.adoc[]
include::appendix/appendix-schema.adoc[]
include::appendix/appendix-command-reference.adoc[]
-:leveloffset: -1
\ No newline at end of file
+:leveloffset: -1
diff --git a/src/main/asciidoc/reference/redis.adoc b/src/main/asciidoc/reference/redis.adoc
index ec93bb711..b366c2788 100644
--- a/src/main/asciidoc/reference/redis.adoc
+++ b/src/main/asciidoc/reference/redis.adoc
@@ -427,8 +427,4 @@ NOTE: By default `RedisCacheManager` will not participate in any ongoing transac
NOTE: By default `RedisCacheManager` does not prefix keys for cache regions, which can lead to an unexpected growth of a `ZSET` used to maintain known keys. It's highly recommended to enable the usage of prefixes in order to avoid this unexpected growth and potential key clashes using more than one cache region.
-[[redis:future]]
-== Roadmap ahead
-
-Spring Data Redis project is in its early stages. We are interested in feedback, knowing what your use cases are, what are the common patters you encounter so that the Redis module better serves your needs. Do contact us using the channels <> above, we are interested in hearing from you!
diff --git a/src/main/java/org/springframework/data/redis/ClusterRedirectException.java b/src/main/java/org/springframework/data/redis/ClusterRedirectException.java
new file mode 100644
index 000000000..a8cb91553
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/ClusterRedirectException.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis;
+
+import org.springframework.dao.DataRetrievalFailureException;
+
+/**
+ * {@link ClusterRedirectException} indicates that a requested slot is not served by the targeted server but can be
+ * obtained on another one.
+ *
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterRedirectException extends DataRetrievalFailureException {
+
+ private static final long serialVersionUID = -857075813794333965L;
+
+ private final int slot;
+ private final String host;
+ private final int port;
+
+ /**
+ * Creates new {@link ClusterRedirectException}.
+ *
+ * @param slot the slot to redirect to.
+ * @param targetHost the host to redirect to.
+ * @param targetPort the port on the host.
+ * @param e the root cause from the data access API in use
+ */
+ public ClusterRedirectException(int slot, String targetHost, int targetPort, Throwable e) {
+
+ super(String.format("Redirect: slot %s to %s:%s.", slot, targetHost, targetPort), e);
+
+ this.slot = slot;
+ this.host = targetHost;
+ this.port = targetPort;
+ }
+
+ /**
+ * Get slot to go for.
+ *
+ * @return
+ */
+ public int getSlot() {
+ return slot;
+ }
+
+ /**
+ * Get host serving the slot.
+ *
+ * @return
+ */
+ public String getTargetHost() {
+ return host;
+ }
+
+ /**
+ * Get port on host serving the slot.
+ *
+ * @return
+ */
+ public int getTargetPort() {
+ return port;
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/redis/ClusterStateFailureExeption.java b/src/main/java/org/springframework/data/redis/ClusterStateFailureExeption.java
new file mode 100644
index 000000000..35a766cbf
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/ClusterStateFailureExeption.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis;
+
+import org.springframework.dao.DataAccessResourceFailureException;
+
+/**
+ * {@link DataAccessResourceFailureException} indicating the current local snapshot of cluster state does no longer
+ * represent the actual remote state. This can happen nodes are removed from cluster, slots get migrated to other nodes
+ * and so on.
+ *
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterStateFailureExeption extends DataAccessResourceFailureException {
+
+ private static final long serialVersionUID = 333399051713240852L;
+
+ /**
+ * Creates new {@link ClusterStateFailureExeption}.
+ *
+ * @param msg
+ */
+ public ClusterStateFailureExeption(String msg) {
+ super(msg);
+ }
+
+ /**
+ * Creates new {@link ClusterStateFailureExeption}.
+ *
+ * @param msg
+ * @param cause
+ */
+ public ClusterStateFailureExeption(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/redis/TooManyClusterRedirectionsException.java b/src/main/java/org/springframework/data/redis/TooManyClusterRedirectionsException.java
new file mode 100644
index 000000000..3d9df7076
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/TooManyClusterRedirectionsException.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis;
+
+import org.springframework.dao.DataRetrievalFailureException;
+
+/**
+ * {@link DataRetrievalFailureException} thrown when following cluster redirects exceeds the max number of edges.
+ *
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class TooManyClusterRedirectionsException extends DataRetrievalFailureException {
+
+ private static final long serialVersionUID = -2818933672669154328L;
+
+ /**
+ * Creates new {@link TooManyClusterRedirectionsException}.
+ *
+ * @param msg
+ */
+ public TooManyClusterRedirectionsException(String msg) {
+ super(msg);
+ }
+
+ /**
+ * Creates new {@link TooManyClusterRedirectionsException}.
+ *
+ * @param msg
+ * @param cause
+ */
+ public TooManyClusterRedirectionsException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutionFailureException.java b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutionFailureException.java
new file mode 100644
index 000000000..6f32b8289
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutionFailureException.java
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.dao.UncategorizedDataAccessException;
+
+/**
+ * Exception thrown when at least one call to a clustered redis environment fails.
+ *
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterCommandExecutionFailureException extends UncategorizedDataAccessException {
+
+ private static final long serialVersionUID = 5727044227040368955L;
+
+ private final Collection causes;
+
+ /**
+ * Creates new {@link ClusterCommandExecutionFailureException}.
+ *
+ * @param cause must not be {@literal null}.
+ */
+ public ClusterCommandExecutionFailureException(Throwable cause) {
+ this(Collections.singletonList(cause));
+ }
+
+ /**
+ * Creates new {@link ClusterCommandExecutionFailureException}.
+ *
+ * @param causes must not be {@literal empty}.
+ */
+ public ClusterCommandExecutionFailureException(List causes) {
+
+ super(causes.get(0).getMessage(), causes.get(0));
+ this.causes = causes;
+ }
+
+ /**
+ * Get the collected errors.
+ *
+ * @return never {@literal null}.
+ */
+ public Collection getCauses() {
+ return causes;
+ }
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java
new file mode 100644
index 000000000..bd70718fb
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java
@@ -0,0 +1,352 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.core.task.AsyncTaskExecutor;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.ClusterRedirectException;
+import org.springframework.data.redis.ExceptionTranslationStrategy;
+import org.springframework.data.redis.TooManyClusterRedirectionsException;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.util.Assert;
+
+/**
+ * {@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
+ * @since 1.7
+ */
+public class ClusterCommandExecutor implements DisposableBean {
+
+ private AsyncTaskExecutor executor;
+ private final ClusterTopologyProvider topologyProvider;
+ private final ClusterNodeResourceProvider resourceProvider;
+ private final ExceptionTranslationStrategy exceptionTranslationStrategy;
+ private int maxRedirects = 5;
+
+ /**
+ * 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}.
+ */
+ public ClusterCommandExecutor(ClusterTopologyProvider topologyProvider, ClusterNodeResourceProvider resourceProvider,
+ ExceptionTranslationStrategy exceptionTranslation) {
+
+ Assert.notNull(topologyProvider);
+ Assert.notNull(resourceProvider);
+ Assert.notNull(exceptionTranslation);
+
+ this.topologyProvider = topologyProvider;
+ this.resourceProvider = resourceProvider;
+ this.exceptionTranslationStrategy = exceptionTranslation;
+ }
+
+ /**
+ * @param topologyProvider must not be {@literal null}.
+ * @param resourceProvider must not be {@literal null}.
+ * @param exceptionTranslation must not be {@literal null}.
+ * @param executor can be {@literal null}.
+ */
+ public ClusterCommandExecutor(ClusterTopologyProvider topologyProvider, ClusterNodeResourceProvider resourceProvider,
+ ExceptionTranslationStrategy exceptionTranslation, AsyncTaskExecutor executor) {
+
+ this(topologyProvider, resourceProvider, exceptionTranslation);
+ this.executor = executor;
+ }
+
+ {
+ if (executor == null) {
+ ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
+ threadPoolTaskExecutor.initialize();
+ this.executor = threadPoolTaskExecutor;
+ }
+ }
+
+ /**
+ * Run {@link ClusterCommandCallback} on a random node.
+ *
+ * @param cmd must not be {@literal null}.
+ * @return
+ */
+ public T executeCommandOnArbitraryNode(ClusterCommandCallback cmd) {
+
+ Assert.notNull(cmd, "ClusterCommandCallback must not be null!");
+ List nodes = new ArrayList(getClusterTopology().getActiveNodes());
+ return executeCommandOnSingleNode(cmd, nodes.get(new Random().nextInt(nodes.size())));
+ }
+
+ /**
+ * 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 T executeCommandOnSingleNode(ClusterCommandCallback cmd, RedisClusterNode node) {
+ return executeCommandOnSingleNode(cmd, node, 0);
+ }
+
+ private T executeCommandOnSingleNode(ClusterCommandCallback 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));
+ }
+
+ S client = this.resourceProvider.getResourceForSpecificNode(node);
+ Assert.notNull(client, "Could not acquire resource for node. Is your cluster info up to date?");
+
+ try {
+ return cmd.doInCluster(client);
+ } catch (RuntimeException ex) {
+
+ RuntimeException translatedException = convertToDataAccessExeption(ex);
+ if (translatedException instanceof ClusterRedirectException) {
+ ClusterRedirectException cre = (ClusterRedirectException) translatedException;
+ return executeCommandOnSingleNode(cmd,
+ topologyProvider.getTopology().lookup(cre.getTargetHost(), cre.getTargetPort()), redirectCount + 1);
+ } else {
+ throw translatedException != null ? translatedException : ex;
+ }
+ } finally {
+ this.resourceProvider.returnResourceForSpecificNode(node, client);
+ }
+ }
+
+ /**
+ * Run {@link ClusterCommandCallback} on all reachable master nodes.
+ *
+ * @param cmd
+ * @return
+ * @throws ClusterCommandExecutionFailureException
+ */
+ public Map executeCommandOnAllNodes(final ClusterCommandCallback cmd) {
+ return executeCommandAsyncOnNodes(cmd, getClusterTopology().getActiveMasterNodes());
+ }
+
+ /**
+ * @param callback
+ * @param nodes
+ * @return
+ * @throws ClusterCommandExecutionFailureException
+ */
+ public java.util.Map executeCommandAsyncOnNodes(
+ final ClusterCommandCallback callback, Iterable nodes) {
+
+ Assert.notNull(callback, "Callback must not be null!");
+ Assert.notNull(nodes, "Nodes must not be null!");
+
+ Map> futures = new LinkedHashMap>();
+ for (final RedisClusterNode node : nodes) {
+
+ futures.put(node, executor.submit(new Callable() {
+
+ @Override
+ public T call() throws Exception {
+ return executeCommandOnSingleNode(callback, node);
+ }
+ }));
+ }
+
+ return collectResults(futures);
+ }
+
+ private Map collectResults(Map> futures) {
+
+ boolean done = false;
+
+ Map result = new HashMap();
+ Map exceptions = new HashMap();
+ while (!done) {
+
+ done = true;
+ for (Map.Entry> 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) {
+
+ RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause());
+ exceptions.put(entry.getKey(), ex != null ? ex : e.getCause());
+ } catch (InterruptedException e) {
+
+ RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause());
+ exceptions.put(entry.getKey(), ex != null ? ex : e.getCause());
+ }
+ }
+ }
+ }
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+
+ done = true;
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ if (!exceptions.isEmpty()) {
+ throw new ClusterCommandExecutionFailureException(new ArrayList(exceptions.values()));
+ }
+ return result;
+ }
+
+ /**
+ * Run {@link MultiKeyClusterCommandCallback} with on a curated set of nodes serving one or more keys.
+ *
+ * @param cmd
+ * @return
+ * @throws ClusterCommandExecutionFailureException
+ */
+ public Map executeMuliKeyCommand(final MultiKeyClusterCommandCallback cmd,
+ Iterable keys) {
+
+ Map> nodeKeyMap = new HashMap>();
+
+ for (byte[] key : keys) {
+ for (RedisClusterNode node : getClusterTopology().getKeyServingNodes(key)) {
+
+ if (nodeKeyMap.containsKey(node)) {
+ nodeKeyMap.get(node).add(key);
+ } else {
+ Set keySet = new LinkedHashSet();
+ keySet.add(key);
+ nodeKeyMap.put(node, keySet);
+ }
+ }
+ }
+
+ Map> futures = new LinkedHashMap>();
+ for (final Entry> entry : nodeKeyMap.entrySet()) {
+
+ if (entry.getKey().isMaster()) {
+ for (final byte[] key : entry.getValue()) {
+ futures.put(entry.getKey(), executor.submit(new Callable() {
+
+ @Override
+ public T call() throws Exception {
+ return (T) executeMultiKeyCommandOnSingleNode(cmd, entry.getKey(), key);
+ }
+ }));
+ }
+ }
+ }
+ return collectResults(futures);
+ }
+
+ private T executeMultiKeyCommandOnSingleNode(MultiKeyClusterCommandCallback cmd, RedisClusterNode node,
+ byte[] key) {
+
+ Assert.notNull(cmd, "MultiKeyCommandCallback must not be null!");
+ Assert.notNull(node, "RedisClusterNode must not be null!");
+ Assert.notNull(key, "Keys for execution must not be null!");
+
+ S client = this.resourceProvider.getResourceForSpecificNode(node);
+ Assert.notNull(client, "Could not acquire resource for node. Is your cluster info up to date?");
+
+ try {
+ return cmd.doInCluster(client, key);
+ } catch (RuntimeException ex) {
+
+ RuntimeException translatedException = convertToDataAccessExeption(ex);
+ throw translatedException != null ? translatedException : ex;
+ } finally {
+ this.resourceProvider.returnResourceForSpecificNode(node, client);
+ }
+ }
+
+ private ClusterTopology getClusterTopology() {
+ return this.topologyProvider.getTopology();
+ }
+
+ private DataAccessException convertToDataAccessExeption(Exception e) {
+ return exceptionTranslationStrategy.translate(e);
+ }
+
+ /**
+ * 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) {
+ this.maxRedirects = maxRedirects;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.beans.factory.DisposableBean#destroy()
+ */
+ @Override
+ public void destroy() throws Exception {
+
+ if (executor instanceof DisposableBean) {
+ ((DisposableBean) executor).destroy();
+ }
+ }
+
+ /**
+ * Callback interface for Redis 'low level' code using the cluster client directly. To be used with
+ * {@link ClusterCommandExecutor} execution methods.
+ *
+ * @author Christoph Strobl
+ * @param native driver connection
+ * @param
+ * @since 1.7
+ */
+ public static interface ClusterCommandCallback {
+ S doInCluster(T client);
+ }
+
+ /**
+ * Callback interface for Redis 'low level' code using the cluster client to execute multi key commands.
+ *
+ * @author Christoph Strobl
+ * @param native driver connection
+ * @param
+ */
+ public static interface MultiKeyClusterCommandCallback {
+ S doInCluster(T client, byte[] key);
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterInfo.java b/src/main/java/org/springframework/data/redis/connection/ClusterInfo.java
new file mode 100644
index 000000000..354719b52
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ClusterInfo.java
@@ -0,0 +1,173 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.Properties;
+
+import org.springframework.data.redis.core.types.RedisClientInfo.INFO;
+import org.springframework.util.Assert;
+
+/**
+ * {@link ClusterInfo} gives access to cluster information such as {@code cluster_state} and
+ * {@code cluster_slots_assigned} provided by the {@code CLUSTER INFO} command.
+ *
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterInfo {
+
+ public static enum Info {
+ STATE("cluster_state"), SLOTS_ASSIGNED("cluster_slots_assigned"), SLOTS_OK("cluster_slots_ok"), SLOTS_PFAIL(
+ "cluster_slots_pfail"), SLOTS_FAIL("cluster_slots_fail"), KNOWN_NODES("cluster_known_nodes"), SIZE(
+ "cluster_size"), CURRENT_EPOCH("cluster_current_epoch"), MY_EPOCH("cluster_my_epoch"), MESSAGES_SENT(
+ "cluster_stats_messages_sent"), MESSAGES_RECEIVED("cluster_stats_messages_received");
+
+ String key;
+
+ Info(String key) {
+ this.key = key;
+ }
+ }
+
+ private final Properties clusterProperties;
+
+ /**
+ * Creates new {@link ClusterInfo} for given {@link Properties}.
+ *
+ * @param clusterProperties must not be {@literal null}.
+ */
+ public ClusterInfo(Properties clusterProperties) {
+
+ Assert.notNull(clusterProperties, "ClusterProperties must not be null!");
+ this.clusterProperties = clusterProperties;
+ }
+
+ /**
+ * @see Info#STATE
+ * @return
+ */
+ public String getState() {
+ return get(Info.STATE);
+ }
+
+ /**
+ * @see Info#SLOTS_ASSIGNED
+ * @return
+ */
+ public Long getSlotsAssigned() {
+ return getLongValueOf(Info.SLOTS_ASSIGNED);
+ }
+
+ /**
+ * @see Info#SLOTS_OK
+ * @return
+ */
+ public Long getSlotsOk() {
+ return getLongValueOf(Info.SLOTS_OK);
+ }
+
+ /**
+ * @see Info#SLOTS_PFAIL
+ * @return
+ */
+ public Long getSlotsPfail() {
+ return getLongValueOf(Info.SLOTS_PFAIL);
+ }
+
+ /**
+ * @see Info#SLOTS_FAIL
+ * @return
+ */
+ public Long getSlotsFail() {
+ return getLongValueOf(Info.SLOTS_FAIL);
+ }
+
+ /**
+ * @see Info#KNOWN_NODES
+ * @return
+ */
+ public Long getKnownNodes() {
+ return getLongValueOf(Info.KNOWN_NODES);
+ }
+
+ /**
+ * @see Info#SIZE
+ * @return
+ */
+ public Long getClusterSize() {
+ return getLongValueOf(Info.SIZE);
+ }
+
+ /**
+ * @see Info#CURRENT_EPOCH
+ * @return
+ */
+ public Long getCurrentEpoch() {
+ return getLongValueOf(Info.CURRENT_EPOCH);
+ }
+
+ /**
+ * @see Info#MESSAGES_SENT
+ * @return
+ */
+ public Long getMessagesSent() {
+ return getLongValueOf(Info.MESSAGES_SENT);
+ }
+
+ /**
+ * @see Info#MESSAGES_RECEIVED
+ * @return
+ */
+ public Long getMessagesReceived() {
+ return getLongValueOf(Info.MESSAGES_RECEIVED);
+ }
+
+ /**
+ * @param info must not be null
+ * @return {@literal null} if no entry found for requested {@link INFO}.
+ */
+ public String get(Info info) {
+
+ Assert.notNull(info, "Cannot retrieve cluster information for 'null'.");
+ return get(info.key);
+ }
+
+ /**
+ * @param key must not be {@literal null} or {@literal empty}.
+ * @return {@literal null} if no entry found for requested {@code key}.
+ */
+ public String get(String key) {
+
+ Assert.hasText(key, "Cannot get cluster information for 'empty' / 'null' key.");
+ return this.clusterProperties.getProperty(key);
+ }
+
+ private Long getLongValueOf(Info info) {
+
+ String value = get(info);
+ return value == null ? null : Long.valueOf(value);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see java.lang.Object#toString()
+ */
+ @Override
+ public String toString() {
+ return this.clusterProperties.toString();
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterNodeResourceProvider.java b/src/main/java/org/springframework/data/redis/connection/ClusterNodeResourceProvider.java
new file mode 100644
index 000000000..379b74b9d
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ClusterNodeResourceProvider.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+/**
+ * {@link ClusterNodeResourceProvider} provides access to low level client api to directly execute operations against a
+ * Redis instance.
+ *
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public interface ClusterNodeResourceProvider {
+
+ /**
+ * Get the client resource for the given node.
+ *
+ * @param node must not be {@literal null}.
+ * @return
+ */
+ S getResourceForSpecificNode(RedisClusterNode node);
+
+ /**
+ * Return the resource object for the given node. This can mean free up resources or return elements back to a pool.
+ *
+ * @param node must not be {@literal null}.
+ * @param resource must not be {@literal null}.
+ */
+ void returnResourceForSpecificNode(RedisClusterNode node, Object resource);
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterSlotHashUtil.java b/src/main/java/org/springframework/data/redis/connection/ClusterSlotHashUtil.java
new file mode 100644
index 000000000..2ddc22412
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ClusterSlotHashUtil.java
@@ -0,0 +1,137 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import org.springframework.util.Assert;
+
+/**
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public final class ClusterSlotHashUtil {
+
+ private static final int SLOT_COUNT = 16384;
+
+ private static final byte SUBKEY_START = '{';
+ private static final byte SUBKEY_END = '}';
+
+ private static final int[] LOOKUP_TABLE = { 0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50A5, 0x60C6, 0x70E7, 0x8108,
+ 0x9129, 0xA14A, 0xB16B, 0xC18C, 0xD1AD, 0xE1CE, 0xF1EF, 0x1231, 0x0210, 0x3273, 0x2252, 0x52B5, 0x4294, 0x72F7,
+ 0x62D6, 0x9339, 0x8318, 0xB37B, 0xA35A, 0xD3BD, 0xC39C, 0xF3FF, 0xE3DE, 0x2462, 0x3443, 0x0420, 0x1401, 0x64E6,
+ 0x74C7, 0x44A4, 0x5485, 0xA56A, 0xB54B, 0x8528, 0x9509, 0xE5EE, 0xF5CF, 0xC5AC, 0xD58D, 0x3653, 0x2672, 0x1611,
+ 0x0630, 0x76D7, 0x66F6, 0x5695, 0x46B4, 0xB75B, 0xA77A, 0x9719, 0x8738, 0xF7DF, 0xE7FE, 0xD79D, 0xC7BC, 0x48C4,
+ 0x58E5, 0x6886, 0x78A7, 0x0840, 0x1861, 0x2802, 0x3823, 0xC9CC, 0xD9ED, 0xE98E, 0xF9AF, 0x8948, 0x9969, 0xA90A,
+ 0xB92B, 0x5AF5, 0x4AD4, 0x7AB7, 0x6A96, 0x1A71, 0x0A50, 0x3A33, 0x2A12, 0xDBFD, 0xCBDC, 0xFBBF, 0xEB9E, 0x9B79,
+ 0x8B58, 0xBB3B, 0xAB1A, 0x6CA6, 0x7C87, 0x4CE4, 0x5CC5, 0x2C22, 0x3C03, 0x0C60, 0x1C41, 0xEDAE, 0xFD8F, 0xCDEC,
+ 0xDDCD, 0xAD2A, 0xBD0B, 0x8D68, 0x9D49, 0x7E97, 0x6EB6, 0x5ED5, 0x4EF4, 0x3E13, 0x2E32, 0x1E51, 0x0E70, 0xFF9F,
+ 0xEFBE, 0xDFDD, 0xCFFC, 0xBF1B, 0xAF3A, 0x9F59, 0x8F78, 0x9188, 0x81A9, 0xB1CA, 0xA1EB, 0xD10C, 0xC12D, 0xF14E,
+ 0xE16F, 0x1080, 0x00A1, 0x30C2, 0x20E3, 0x5004, 0x4025, 0x7046, 0x6067, 0x83B9, 0x9398, 0xA3FB, 0xB3DA, 0xC33D,
+ 0xD31C, 0xE37F, 0xF35E, 0x02B1, 0x1290, 0x22F3, 0x32D2, 0x4235, 0x5214, 0x6277, 0x7256, 0xB5EA, 0xA5CB, 0x95A8,
+ 0x8589, 0xF56E, 0xE54F, 0xD52C, 0xC50D, 0x34E2, 0x24C3, 0x14A0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405, 0xA7DB,
+ 0xB7FA, 0x8799, 0x97B8, 0xE75F, 0xF77E, 0xC71D, 0xD73C, 0x26D3, 0x36F2, 0x0691, 0x16B0, 0x6657, 0x7676, 0x4615,
+ 0x5634, 0xD94C, 0xC96D, 0xF90E, 0xE92F, 0x99C8, 0x89E9, 0xB98A, 0xA9AB, 0x5844, 0x4865, 0x7806, 0x6827, 0x18C0,
+ 0x08E1, 0x3882, 0x28A3, 0xCB7D, 0xDB5C, 0xEB3F, 0xFB1E, 0x8BF9, 0x9BD8, 0xABBB, 0xBB9A, 0x4A75, 0x5A54, 0x6A37,
+ 0x7A16, 0x0AF1, 0x1AD0, 0x2AB3, 0x3A92, 0xFD2E, 0xED0F, 0xDD6C, 0xCD4D, 0xBDAA, 0xAD8B, 0x9DE8, 0x8DC9, 0x7C26,
+ 0x6C07, 0x5C64, 0x4C45, 0x3CA2, 0x2C83, 0x1CE0, 0x0CC1, 0xEF1F, 0xFF3E, 0xCF5D, 0xDF7C, 0xAF9B, 0xBFBA, 0x8FD9,
+ 0x9FF8, 0x6E17, 0x7E36, 0x4E55, 0x5E74, 0x2E93, 0x3EB2, 0x0ED1, 0x1EF0 };
+
+ private ClusterSlotHashUtil() {
+
+ }
+
+ /**
+ * @param keys must not be {@literal null}.
+ * @return
+ */
+ public static boolean isSameSlotForAllKeys(byte[]... keys) {
+
+ Assert.notNull(keys, "Keys must not be null!");
+
+ if (keys.length <= 1) {
+ return true;
+ }
+
+ int slot = calculateSlot(keys[0]);
+ for (int i = 1; i < keys.length; i++) {
+ if (slot != calculateSlot(keys[i])) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Calculate the slot from the given key.
+ *
+ * @param key must not be {@literal null} or empty.
+ * @return
+ */
+ public static int calculateSlot(String key) {
+
+ Assert.hasText(key, "Key must not be null or empty!");
+ return calculateSlot(key.getBytes());
+ }
+
+ /**
+ * Calculate the slot from the given key.
+ *
+ * @param key must not be {@literal null}.
+ * @return
+ */
+ public static int calculateSlot(byte[] key) {
+
+ Assert.notNull(key, "Key must not be null!");
+
+ byte[] finalKey = key;
+ int start = indexOf(key, SUBKEY_START);
+ if (start != -1) {
+ int end = indexOf(key, start + 1, SUBKEY_END);
+ if (end != -1 && end != start + 1) {
+
+ finalKey = new byte[end - (start + 1)];
+ System.arraycopy(key, start + 1, finalKey, 0, finalKey.length);
+ }
+ }
+ return crc16(finalKey) % SLOT_COUNT;
+ }
+
+ private static int indexOf(byte[] haystack, byte needle) {
+ return indexOf(haystack, 0, needle);
+ }
+
+ private static int indexOf(byte[] haystack, int start, byte needle) {
+
+ for (int i = start; i < haystack.length; i++) {
+
+ if (haystack[i] == needle) {
+ return i;
+ }
+ }
+
+ return -1;
+ }
+
+ private static int crc16(byte[] bytes) {
+
+ int crc = 0x0000;
+
+ for (byte b : bytes) {
+ crc = ((crc << 8) ^ LOOKUP_TABLE[((crc >>> 8) ^ (b & 0xFF)) & 0xFF]);
+ }
+ return crc & 0xFFFF;
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterTopology.java b/src/main/java/org/springframework/data/redis/connection/ClusterTopology.java
new file mode 100644
index 000000000..70b5f3483
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ClusterTopology.java
@@ -0,0 +1,169 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import org.springframework.data.redis.ClusterStateFailureExeption;
+import org.springframework.util.Assert;
+
+/**
+ * {@link ClusterTopology} holds snapshot like information about {@link RedisClusterNode}s.
+ *
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterTopology {
+
+ private final Set nodes;
+
+ /**
+ * Creates new instance of {@link ClusterTopology}.
+ *
+ * @param nodes can be {@literal null}.
+ */
+ public ClusterTopology(Set nodes) {
+ this.nodes = nodes != null ? nodes : Collections. emptySet();
+ }
+
+ /**
+ * Get all {@link RedisClusterNode}s.
+ *
+ * @return never {@literal null}.
+ */
+ public Set getNodes() {
+ return Collections.unmodifiableSet(nodes);
+ }
+
+ /**
+ * Get all nodes (master and slave) in cluster where {@code link-state} is {@literal connected} and {@code flags} does
+ * not contain {@literal fail} or {@literal fail?}.
+ *
+ * @return never {@literal null}.
+ */
+ public Set getActiveNodes() {
+
+ Set activeNodes = new LinkedHashSet(nodes.size());
+ for (RedisClusterNode node : nodes) {
+ if (node.isConnected() && !node.isMarkedAsFail()) {
+ activeNodes.add(node);
+ }
+ }
+ return activeNodes;
+ }
+
+ /**
+ * Get all master nodes in cluster where {@code link-state} is {@literal connected} and {@code flags} does not contain
+ * {@literal fail} or {@literal fail?}.
+ *
+ * @return never {@literal null}.
+ */
+ public Set getActiveMasterNodes() {
+
+ Set activeMasterNodes = new LinkedHashSet(nodes.size());
+ for (RedisClusterNode node : nodes) {
+ if (node.isMaster() && node.isConnected() && !node.isMarkedAsFail()) {
+ activeMasterNodes.add(node);
+ }
+ }
+ return activeMasterNodes;
+ }
+
+ /**
+ * Get all master nodes in cluster.
+ *
+ * @return never {@literal null}.
+ */
+ public Set getMasterNodes() {
+
+ Set masterNodes = new LinkedHashSet(nodes.size());
+ for (RedisClusterNode node : nodes) {
+ if (node.isMaster()) {
+ masterNodes.add(node);
+ }
+ }
+ return masterNodes;
+ }
+
+ /**
+ * Get the {@link RedisClusterNode}s (master and slave) serving s specific slot.
+ *
+ * @param slot
+ * @return never {@literal null}.
+ */
+ public Set getSlotServingNodes(int slot) {
+
+ Set slotServingNodes = new LinkedHashSet(nodes.size());
+ for (RedisClusterNode node : nodes) {
+ if (node.servesSlot(slot)) {
+ slotServingNodes.add(node);
+ }
+ }
+ return slotServingNodes;
+ }
+
+ /**
+ * Get the {@link RedisClusterNode} that is the current master serving the given key.
+ *
+ * @param key must not be {@literal null}.
+ * @return
+ * @throws ClusterStateFailureExeption
+ */
+ public RedisClusterNode getKeyServingMasterNode(byte[] key) {
+
+ Assert.notNull(key, "Key for node lookup must not be null!");
+
+ int slot = ClusterSlotHashUtil.calculateSlot(key);
+ for (RedisClusterNode node : nodes) {
+ if (node.isMaster() && node.servesSlot(slot)) {
+ return node;
+ }
+ }
+ throw new ClusterStateFailureExeption(String.format("Could not find master node serving slot %s for key '%s',",
+ slot, key));
+ }
+
+ /**
+ * Get the {@link RedisClusterNode} matching given {@literal host} and {@literal port}.
+ *
+ * @param host must not be {@literal null}.
+ * @param port
+ * @return
+ * @throws ClusterStateFailureExeption
+ */
+ public RedisClusterNode lookup(String host, int port) {
+
+ for (RedisClusterNode node : nodes) {
+ if (host.equals(node.getHost()) && port == node.getPort()) {
+ return node;
+ }
+ }
+ throw new ClusterStateFailureExeption(String.format(
+ "Could not find node at %s:%s. Is your cluster info up to date?", host, port));
+ }
+
+ /**
+ * @param key must not be {@literal null}.
+ * @return {@literal null}.
+ */
+ public Set getKeyServingNodes(byte[] key) {
+
+ Assert.notNull(key, "Key must not be null for Cluster Node lookup.");
+ return getSlotServingNodes(ClusterSlotHashUtil.calculateSlot(key));
+ }
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterTopologyProvider.java b/src/main/java/org/springframework/data/redis/connection/ClusterTopologyProvider.java
new file mode 100644
index 000000000..b3fb569d4
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ClusterTopologyProvider.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+/**
+ * {@link ClusterTopologyProvider} manages the current cluster topology and makes sure to refresh cluster information.
+ *
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public interface ClusterTopologyProvider {
+
+ /**
+ * Get the current known {@link ClusterTopology}.
+ *
+ * @return never {@null}.
+ */
+ ClusterTopology getTopology();
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
index 751d00d9c..702948003 100644
--- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
+++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
@@ -2688,4 +2688,22 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
return byteSetToStringSet.convert(results);
}
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption)
+ */
+ @Override
+ public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option) {
+ delegate.migrate(key, target, dbIndex, option);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption, long)
+ */
+ @Override
+ public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option, long timeout) {
+ delegate.migrate(key, target, dbIndex, option, timeout);
+ }
+
}
diff --git a/src/main/java/org/springframework/data/redis/connection/RedisClusterCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisClusterCommands.java
new file mode 100644
index 000000000..ef6df7ff7
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/RedisClusterCommands.java
@@ -0,0 +1,167 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.data.redis.connection.RedisClusterNode.SlotRange;
+
+/**
+ * Interface for the {@literal cluster} commands supported by Redis.
+ *
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public interface RedisClusterCommands {
+
+ /**
+ * Retrieve cluster node information such as {@literal id}, {@literal host}, {@literal port} and {@literal slots}.
+ *
+ * @return never {@literal null}.
+ */
+ Iterable clusterGetClusterNodes();
+
+ /**
+ * Retrieve information about connected slaves for given master node.
+ *
+ * @param node must not be {@literal null}.
+ * @return never {@literal null}.
+ */
+ Collection clusterGetSlaves(RedisClusterNode master);
+
+ /**
+ * Retrieve information about masters and their connected slaves.
+ *
+ * @return never {@literal null}.
+ */
+ Map> clusterGetMasterSlaveMap();
+
+ /**
+ * Find the slot for a given {@code key}.
+ *
+ * @param key must not be {@literal null}.
+ * @return
+ */
+ Integer clusterGetSlotForKey(byte[] key);
+
+ /**
+ * Find the {@link RedisClusterNode} serving given {@literal slot}.
+ *
+ * @param slot
+ * @return
+ */
+ RedisClusterNode clusterGetNodeForSlot(int slot);
+
+ /**
+ * Find the {@link RedisClusterNode} serving given {@literal key}.
+ *
+ * @param key must not be {@literal null}.
+ * @return
+ */
+ RedisClusterNode clusterGetNodeForKey(byte[] key);
+
+ /**
+ * Get cluster information.
+ *
+ * @return
+ */
+ ClusterInfo clusterGetClusterInfo();
+
+ /**
+ * Assign slots to given {@link RedisClusterNode}.
+ *
+ * @param node must not be {@literal null}.
+ * @param slots
+ */
+ 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}.
+ */
+ void clusterAddSlots(RedisClusterNode node, SlotRange range);
+
+ /**
+ * Count the number of keys assigned to one {@literal slot}.
+ *
+ * @param slot
+ * @return
+ */
+ Long clusterCountKeysInSlot(int slot);
+
+ /**
+ * Remove slots from {@link RedisClusterNode}.
+ *
+ * @param node must not be {@literal null}.
+ * @param slots
+ */
+ 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}.
+ */
+ void clusterDeleteSlotsInRange(RedisClusterNode node, SlotRange range);
+
+ /**
+ * Remove given {@literal node} from cluster.
+ *
+ * @param node must not be {@literal null}.
+ */
+ void clusterForget(RedisClusterNode node);
+
+ /**
+ * Add given {@literal node} to cluster.
+ *
+ * @param node must not be {@literal null}.
+ */
+ void clusterMeet(RedisClusterNode node);
+
+ /**
+ * @param node must not be {@literal null}.
+ * @param slot
+ * @param mode must not be{@literal null}.
+ */
+ void clusterSetSlot(RedisClusterNode node, int slot, AddSlots mode);
+
+ /**
+ * Get {@literal keys} served by slot.
+ *
+ * @param slot
+ * @param count must not be {@literal null}.
+ * @return
+ */
+ List clusterGetKeysInSlot(int slot, Integer count);
+
+ /**
+ * Assign a {@literal slave} to given {@literal master}.
+ *
+ * @param master must not be {@literal null}.
+ * @param slave must not be {@literal null}.
+ */
+ void clusterReplicate(RedisClusterNode master, RedisClusterNode slave);
+
+ public enum AddSlots {
+ MIGRATING, IMPORTING, STABLE, NODE
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/RedisClusterConfiguration.java b/src/main/java/org/springframework/data/redis/connection/RedisClusterConfiguration.java
new file mode 100644
index 000000000..43cf45d98
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/RedisClusterConfiguration.java
@@ -0,0 +1,211 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import static org.springframework.util.Assert.*;
+import static org.springframework.util.StringUtils.*;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.springframework.core.env.MapPropertySource;
+import org.springframework.core.env.PropertySource;
+import org.springframework.util.NumberUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * Configuration class used for setting up {@link RedisConnection} via {@link RedisConnectionFactory} using connecting
+ * to Redis Cluster. Useful when setting up a high availability Redis
+ * environment.
+ *
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class RedisClusterConfiguration {
+
+ private static final String REDIS_CLUSTER_NODES_CONFIG_PROPERTY = "spring.redis.cluster.nodes";
+ private static final String REDIS_CLUSTER_TIMEOUT_CONFIG_PROPERTY = "spring.redis.cluster.timeout";
+ private static final String REDIS_CLUSTER_MAX_REDIRECTS_CONFIG_PROPERTY = "spring.redis.cluster.max-redirects";
+
+ private Set clusterNodes;
+ private Long clusterTimeout;
+ private Integer maxRedirects;
+
+ /**
+ * Creates new {@link RedisClusterConfiguration}.
+ */
+ public RedisClusterConfiguration() {
+ this(new MapPropertySource("RedisClusterConfiguration", Collections. emptyMap()));
+ }
+
+ /**
+ * Creates {@link RedisClusterConfiguration} for given hostPort combinations.
+ *
+ *