DATAREDIS-330 - Add support for SENTINEL commands.

add support for commands:
  SENTINEL FAILOVER master
  SENTINEL SLAVES master
  SENTINEL REMOVE name
  SENTINEL MONITOR name ip port quorum

Original pull request: #92.

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

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2014 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.io.IOException;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
+import org.springframework.data.redis.RedisSystemException;
+
+/**
+ * @author Christoph Strobl
+ * @since 1.4
+ */
+public abstract class AbstractRedisConnection implements RedisConnection {
+
+	private RedisSentinelConfiguration sentinelConfiguration;
+	private ConcurrentHashMap<RedisNode, RedisSentinelConnection> connectionCache = new ConcurrentHashMap<RedisNode, RedisSentinelConnection>();
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisConnection#getSentinelCommands()
+	 */
+	@Override
+	public RedisSentinelConnection getSentinelConnection() {
+
+		if (!hasRedisSentinelConfigured()) {
+			throw new InvalidDataAccessResourceUsageException("No sentinels configured.");
+		}
+
+		RedisNode node = selectActiveSentinel();
+		RedisSentinelConnection connection = connectionCache.get(node);
+		if (connection == null || !connection.isOpen()) {
+			connection = getSentinelConnection(node);
+			connectionCache.putIfAbsent(node, connection);
+		}
+		return connection;
+	}
+
+	public void setSentinelConfiguration(RedisSentinelConfiguration sentinelConfiguration) {
+		this.sentinelConfiguration = sentinelConfiguration;
+	}
+
+	public boolean hasRedisSentinelConfigured() {
+		return this.sentinelConfiguration != null;
+	}
+
+	private RedisNode selectActiveSentinel() {
+
+		for (RedisNode node : this.sentinelConfiguration.getSentinels()) {
+			if (isActive(node)) {
+				return node;
+			}
+		}
+
+		throw new InvalidDataAccessApiUsageException("Could not find any active sentinels");
+	}
+
+	/**
+	 * Check if node is active by sending ping.
+	 * 
+	 * @param node
+	 * @return
+	 */
+	protected boolean isActive(RedisNode node) {
+		return false;
+	}
+
+	/**
+	 * Get {@link RedisSentinelCommands} connected to given node.
+	 * 
+	 * @param sentinel
+	 * @return
+	 */
+	protected RedisSentinelConnection getSentinelConnection(RedisNode sentinel) {
+		throw new UnsupportedOperationException("Sentinel is not supported by this client.");
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisConnection#close()
+	 */
+	@Override
+	public void close() throws DataAccessException {
+
+		if (!connectionCache.isEmpty()) {
+			for (RedisNode node : connectionCache.keySet()) {
+				RedisSentinelConnection connection = connectionCache.remove(node);
+				if (connection.isOpen()) {
+					try {
+						connection.close();
+					} catch (IOException e) {
+						throw new RedisSystemException("Failed to close sentinel connection", e);
+					}
+				}
+			}
+		}
+	}
+
+}

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

@@ -2380,4 +2380,9 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
 				new TupleConverter());
 	}
 
+	@Override
+	public RedisSentinelConnection getSentinelConnection() {
+		return delegate.getSentinelConnection();
+	}
+
 }

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

@@ -92,4 +92,10 @@ public interface RedisConnection extends RedisCommands {
 	 * @return the result of the executed commands.
 	 */
 	List<Object> closePipeline() throws RedisPipelineException;
+
+	/**
+	 * @return
+	 * @since 1.4
+	 */
+	RedisSentinelConnection getSentinelConnection();
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2013 the original author or authors.
+ * Copyright 2011-2014 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.
@@ -22,6 +22,7 @@ import org.springframework.dao.support.PersistenceExceptionTranslator;
  * Thread-safe factory of Redis connections.
  * 
  * @author Costin Leau
+ * @author Christoph Strobl
  */
 public interface RedisConnectionFactory extends PersistenceExceptionTranslator {
 
@@ -41,4 +42,10 @@ public interface RedisConnectionFactory extends PersistenceExceptionTranslator {
 	 * @return Whether or not to convert pipeline and tx results
 	 */
 	boolean getConvertPipelineAndTxResults();
+
+	/**
+	 * @return
+	 * @since 1.4
+	 */
+	RedisSentinelConnection getSentinelConnection();
 }

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

@@ -15,22 +15,37 @@
  */
 package org.springframework.data.redis.connection;
 
+import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
 /**
  * @author Christoph Strobl
+ * @author Thomas Darimont
+ * 
  * @since 1.4
  */
-public class RedisNode {
+public class RedisNode implements NamedNode {
 
-	private final String host;
-	private final int port;
+	private String name;
+	private String host;
+	private int port;
 
+	/**
+	 * Creates a new {@link RedisNode} with the given {@code host}, {@code port}.
+	 * 
+	 * @param host must not be {@literal null}
+	 * @param port
+	 */
 	public RedisNode(String host, int port) {
+		
+		Assert.notNull(host,"host must not be null!");
+		
 		this.host = host;
 		this.port = port;
 	}
 
+	protected RedisNode() {}
+
 	public String getHost() {
 		return host;
 	}
@@ -43,6 +58,15 @@ public class RedisNode {
 		return host + ":" + port;
 	}
 
+	@Override
+	public String getName() {
+		return this.name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
 	@Override
 	public String toString() {
 		return asString();
@@ -77,7 +101,42 @@ public class RedisNode {
 			return false;
 		}
 
+		if (!ObjectUtils.nullSafeEquals(this.name, other.name)) {
+			return false;
+		}
+
 		return true;
 	}
 
+	/**
+	 * @author Christoph Strobl
+	 * 
+	 * @since 1.4
+	 */
+	public static class RedisNodeBuilder {
+
+		private RedisNode node;
+
+		public RedisNodeBuilder() {
+			node = new RedisNode();
+		}
+
+		public RedisNodeBuilder withName(String name) {
+			node.name = name;
+			return this;
+		}
+
+		public RedisNodeBuilder listeningAt(String host, int port) {
+
+			Assert.hasText(host, "Hostname must not be empty or null.");
+			node.host = host;
+			node.port = port;
+			return this;
+		}
+
+		public RedisNode build() {
+			return this.node;
+		}
+	}
+
 }

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

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2014 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;
+
+/**
+ * @author Christoph Strobl
+ * @since 1.4
+ */
+public interface RedisSentinelCommands {
+
+	/**
+	 * Force a failover as if the {@literal master} was not reachable.
+	 * 
+	 * @param master must not be {@literal null}.
+	 */
+	void failover(NamedNode master);
+
+	/**
+	 * Get a {@link Collection} of monitored masters and their state.
+	 * 
+	 * @return Collection of {@link RedisServer}s. Never {@literal null}.
+	 */
+	Collection<RedisServer> masters();
+
+	/**
+	 * Show list of slaves for given {@literal master}.
+	 * 
+	 * @param master must not be {@literal null}.
+	 * @return Collection of {@link RedisServer}s. Never {@literal null}.
+	 */
+	Collection<RedisServer> slaves(NamedNode master);
+
+	/**
+	 * Removes given {@literal master}. The server will no longer be monitored and will no longer be returned by
+	 * {@link #masters()}.
+	 * 
+	 * @param master must not be {@literal null}.
+	 */
+	void remove(NamedNode master);
+
+	/**
+	 * Tell sentinel to start monitoring a new {@literal master} with the specified {@link RedisServer#getName()},
+	 * {@link RedisServer#getHost()}, {@link RedisServer#getPort()}, and {@link RedisServer#getQuorum()}.
+	 * 
+	 * @param master must not be {@literal null}.
+	 */
+	void monitor(RedisServer master);
+
+}

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

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2014 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.io.Closeable;
+
+/**
+ * @author Christoph Strobl
+ * @since 1.4
+ */
+public interface RedisSentinelConnection extends RedisSentinelCommands, Closeable {
+
+	/**
+	 * @return true if connected to server
+	 */
+	boolean isOpen();
+
+}

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

@@ -0,0 +1,213 @@
+/*
+ * Copyright 2014 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.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Christoph Strobl
+ * @author Thomas Darimont
+ * 
+ * @since 1.4
+ */
+public class RedisServer extends RedisNode {
+
+	public static enum INFO {
+
+		NAME("name"), //
+		HOST("ip"), //
+		PORT("port"), //
+		RUN_ID("runid"), //
+		FLAGS("flags"), //
+		PENDING_COMMANDS("pending-commands"), //
+		LAST_PING_SENT("last-ping-sent"), //
+		LAST_OK_PING_REPLY("last-ok-ping-reply"), //
+		DOWN_AFTER_MILLISECONDS("down-after-milliseconds"), //
+		INFO_REFRESH("info-refresh"), //
+		ROLE_REPORTED("role-reported"), //
+		ROLE_REPORTED_TIME("role-reported-time"), //
+		CONFIG_EPOCH("config-epoch"), //
+		NUMBER_SLAVES("num-slaves"), //
+		NUMBER_OTHER_SENTINELS("multi"), //
+		BUFFER_LENGTH("qbuf"), //
+		BUFFER_FREE_SPACE("qbuf-free"), //
+		OUTPUT_BUFFER_LENGTH("obl"), //
+		OUTPUT_LIST_LENGTH("number-other-sentinels"), //
+		QUORUM("quorum"), //
+		FAILOVER_TIMEOUT("failover-timeout"), //
+		PARALLEL_SYNCS("parallel-syncs"); //
+
+		String key;
+
+		INFO(String key) {
+			this.key = key;
+		}
+	}
+
+	private Properties properties;
+
+	/**
+	 * Creates a new {@link RedisServer} with the given {@code host}, {@code port}.
+	 * 
+	 * @param host must not be {@literal null}
+	 * @param port
+	 */
+	public RedisServer(String host, int port) {
+		this(host, port, new Properties());
+	}
+
+	/**
+	 * Creates a new {@link RedisServer} with the given {@code host}, {@code port} and {@code properties}.
+	 * 
+	 * @param host must not be {@literal null}
+	 * @param port
+	 * @param properties may be {@literal null}
+	 */
+	public RedisServer(String host, int port, Properties properties) {
+
+		super(host, port);
+		this.properties = properties;
+
+		String name = host + ":" + port;
+		if (properties != null && properties.containsKey(INFO.NAME.key)) {
+			name = get(INFO.NAME);
+		}
+
+		setName(name);
+	}
+
+	/**
+	 * Creates a new {@link RedisServer} from the given properties.
+	 * 
+	 * @param properties
+	 * @return
+	 */
+	public static RedisServer newServerFrom(Properties properties) {
+
+		String host = properties.getProperty(INFO.HOST.key, "127.0.0.1");
+		int port = Integer.parseInt(properties.getProperty(INFO.PORT.key, "26379"));
+
+		return new RedisServer(host, port, properties);
+	}
+
+	public void setQuorum(Long quorum) {
+
+		if (quorum == null) {
+			this.properties.remove(INFO.QUORUM.key);
+			return;
+		}
+
+		this.properties.put(INFO.QUORUM.key, quorum.toString());
+	}
+
+	public String getRunId() {
+		return get(INFO.RUN_ID);
+	}
+
+	public String getFlags() {
+		return get(INFO.FLAGS);
+	}
+
+	public boolean isMaster() {
+
+		String role = getRoleReported();
+		if (!StringUtils.hasText(role)) {
+			return false;
+		}
+		return role.equalsIgnoreCase("master");
+	}
+
+	public Long getPendingCommands() {
+		return getLongValueOf(INFO.PENDING_COMMANDS);
+	}
+
+	public Long getLastPingSent() {
+		return getLongValueOf(INFO.LAST_PING_SENT);
+	}
+
+	public Long getLastOkPingReply() {
+		return getLongValueOf(INFO.LAST_OK_PING_REPLY);
+	}
+
+	public Long getDownAfterMilliseconds() {
+		return getLongValueOf(INFO.DOWN_AFTER_MILLISECONDS);
+	}
+
+	public Long getInfoRefresh() {
+		return getLongValueOf(INFO.INFO_REFRESH);
+	}
+
+	public String getRoleReported() {
+		return get(INFO.ROLE_REPORTED);
+	}
+
+	public Long roleReportedTime() {
+		return getLongValueOf(INFO.ROLE_REPORTED_TIME);
+	}
+
+	public Long getConfigEpoch() {
+		return getLongValueOf(INFO.CONFIG_EPOCH);
+	}
+
+	public Long getNumberSlaves() {
+		return getLongValueOf(INFO.NUMBER_SLAVES);
+	}
+
+	public Long getNumberOtherSentinels() {
+		return getLongValueOf(INFO.NUMBER_OTHER_SENTINELS);
+	}
+
+	public Long getQuorum() {
+		return getLongValueOf(INFO.QUORUM);
+	}
+
+	public Long getFailoverTimeout() {
+		return getLongValueOf(INFO.FAILOVER_TIMEOUT);
+	}
+
+	public Long getParallelSyncs() {
+		return getLongValueOf(INFO.PARALLEL_SYNCS);
+	}
+
+	/**
+	 * @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 client 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 information for 'empty' / 'null' key.");
+		return this.properties.getProperty(key);
+	}
+
+	private Long getLongValueOf(INFO info) {
+
+		String value = get(info);
+		return value == null ? null : Long.valueOf(value);
+	}
+}

src/main/java/org/springframework/data/redis/connection/convert/Converters.java

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.convert;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 
@@ -38,6 +39,7 @@ abstract public class Converters {
 	private static final Converter<String, Properties> STRING_TO_PROPS = new StringToPropertiesConverter();
 	private static final Converter<Long, Boolean> LONG_TO_BOOLEAN = new LongToBooleanConverter();
 	private static final Converter<String, DataType> STRING_TO_DATA_TYPE = new StringToDataTypeConverter();
+	private static final Converter<Map<?, ?>, Properties> MAP_TO_PROPERTIES = MapToPropertiesConverter.INSTANCE;
 
 	public static Converter<String, Properties> stringToProps() {
 		return STRING_TO_PROPS;
@@ -55,6 +57,10 @@ abstract public class Converters {
 		return STRING_TO_PROPS.convert(source);
 	}
 
+	public static Properties toProperties(Map<?, ?> source) {
+		return MAP_TO_PROPERTIES.convert(source);
+	}
+
 	public static Boolean toBoolean(Long source) {
 		return LONG_TO_BOOLEAN.convert(source);
 	}
@@ -76,7 +82,6 @@ abstract public class Converters {
 		return tupleArgs;
 	}
 
-
 	/**
 	 * Returns the timestamp constructed from the given {@code seconds} and {@code microseconds}.
 	 *

src/main/java/org/springframework/data/redis/connection/convert/MapToPropertiesConverter.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2014 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.convert;
+
+import java.util.Map;
+import java.util.Properties;
+
+import org.springframework.core.convert.converter.Converter;
+
+/**
+ * @author Christoph Strobl
+ * @since 1.4
+ */
+public enum MapToPropertiesConverter implements Converter<Map<?, ?>, Properties> {
+	INSTANCE;
+
+	@Override
+	public Properties convert(Map<?, ?> source) {
+
+		Properties p = new Properties();
+		if (source != null) {
+			p.putAll(source);
+		}
+		return p;
+	}
+
+}

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -34,10 +34,11 @@ import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.ExceptionTranslationStrategy;
 import org.springframework.data.redis.FallbackExceptionTranslationStrategy;
 import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.connection.AbstractRedisConnection;
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.FutureResult;
 import org.springframework.data.redis.connection.MessageListener;
-import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisNode;
 import org.springframework.data.redis.connection.RedisPipelineException;
 import org.springframework.data.redis.connection.RedisSubscribedConnectionException;
 import org.springframework.data.redis.connection.RedisZSetCommands;
@@ -85,7 +86,7 @@ import redis.clients.util.Pool;
  * @author Thomas Darimont
  * @author Jungtaek Lim
  */
-public class JedisConnection implements RedisConnection {
+public class JedisConnection extends AbstractRedisConnection {
 
 	private static final Field CLIENT_FIELD;
 	private static final Method SEND_COMMAND;
@@ -234,6 +235,7 @@ public class JedisConnection implements RedisConnection {
 	}
 
 	public void close() throws DataAccessException {
+		super.close();
 		// return the connection to the pool
 		if (pool != null) {
 			if (!broken) {
@@ -3114,4 +3116,35 @@ public class JedisConnection implements RedisConnection {
 
 		return args;
 	}
+
+	@Override
+	protected boolean isActive(RedisNode node) {
+
+		if (node == null) {
+			return false;
+		}
+
+		Jedis temp = null;
+		try {
+			temp = getJedis(node);
+			temp.connect();
+			return temp.ping().equalsIgnoreCase("pong");
+		} catch (Exception e) {
+			return false;
+		} finally {
+			if (temp != null) {
+				temp.disconnect();
+				temp.close();
+			}
+		}
+	}
+
+	@Override
+	protected JedisSentinelConnection getSentinelConnection(RedisNode sentinel) {
+		return new JedisSentinelConnection(getJedis(sentinel));
+	}
+
+	protected Jedis getJedis(RedisNode node) {
+		return new Jedis(node.getHost(), node.getPort());
+	}
 }

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java

@@ -26,6 +26,7 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
 import org.springframework.data.redis.ExceptionTranslationStrategy;
 import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
 import org.springframework.data.redis.RedisConnectionFailureException;
@@ -33,6 +34,7 @@ import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisNode;
 import org.springframework.data.redis.connection.RedisSentinelConfiguration;
+import org.springframework.data.redis.connection.RedisSentinelConnection;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
@@ -413,6 +415,32 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 		return sentinelConfig != null;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisConnectionFactory#getSentinelConnection()
+	 */
+	@Override
+	public RedisSentinelConnection getSentinelConnection() {
+
+		if (!isRedisSentinelAware()) {
+			throw new InvalidDataAccessResourceUsageException("No Sentinels configured");
+		}
+		
+		return new JedisSentinelConnection(getActiveSentinel());
+	}
+
+	private Jedis getActiveSentinel() {
+
+		Assert.notNull(this.sentinelConfig);
+		for (RedisNode node : this.sentinelConfig.getSentinels()) {
+			Jedis jedis = new Jedis(node.getHost(), node.getPort());
+			if (jedis.ping().equalsIgnoreCase("pong")) {
+				return jedis;
+			}
+		}
+
+		throw new InvalidDataAccessResourceUsageException("no sentinel found");
+	}
+
 	private Set<String> convertToJedisSentinelSet(Collection<RedisNode> nodes) {
 
 		if (CollectionUtils.isEmpty(nodes)) {

src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java

@@ -15,15 +15,18 @@
  */
 package org.springframework.data.redis.connection.jedis;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.connection.DefaultTuple;
 import org.springframework.data.redis.connection.RedisListCommands.Position;
+import org.springframework.data.redis.connection.RedisServer;
 import org.springframework.data.redis.connection.RedisStringCommands.BitOperation;
 import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
 import org.springframework.data.redis.connection.SortParameters;
@@ -36,6 +39,7 @@ import org.springframework.data.redis.connection.convert.SetConverter;
 import org.springframework.data.redis.connection.convert.StringToRedisClientInfoConverter;
 import org.springframework.data.redis.core.types.RedisClientInfo;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
 import redis.clients.jedis.BinaryClient.LIST_POSITION;
@@ -61,6 +65,7 @@ abstract public class JedisConverters extends Converters {
 	private static final Converter<Exception, DataAccessException> EXCEPTION_CONVERTER = new JedisExceptionConverter();
 	private static final Converter<String[], List<RedisClientInfo>> STRING_TO_CLIENT_INFO_CONVERTER = new StringToRedisClientInfoConverter();
 	private static final Converter<redis.clients.jedis.Tuple, Tuple> TUPLE_CONVERTER;
+	private static final Converter<Properties, RedisServer> PROPERTIES_TO_SENTINEL;
 	private static final ListConverter<redis.clients.jedis.Tuple, Tuple> TUPLE_LIST_TO_TUPLE_LIST_CONVERTER;
 
 	static {
@@ -80,6 +85,14 @@ abstract public class JedisConverters extends Converters {
 		};
 		TUPLE_SET_TO_TUPLE_SET = new SetConverter<redis.clients.jedis.Tuple, Tuple>(TUPLE_CONVERTER);
 		TUPLE_LIST_TO_TUPLE_LIST_CONVERTER = new ListConverter<redis.clients.jedis.Tuple, Tuple>(TUPLE_CONVERTER);
+		PROPERTIES_TO_SENTINEL = new Converter<Properties, RedisServer>() {
+
+			@Override
+			public RedisServer convert(Properties source) {
+				return source != null ? RedisServer.newServerFrom(source) : null;
+
+			}
+		};
 	}
 
 	public static Converter<String, byte[]> stringToBytes() {
@@ -157,6 +170,28 @@ abstract public class JedisConverters extends Converters {
 		return STRING_TO_CLIENT_INFO_CONVERTER.convert(source.split("\\r?\\n"));
 	}
 
+	/**
+	 * @param source
+	 * @return
+	 * @since 1.4
+	 */
+	public static List<RedisServer> toListOfRedisServer(List<Map<String, String>> source) {
+
+		if (CollectionUtils.isEmpty(source)) {
+			return Collections.emptyList();
+		}
+
+		List<RedisServer> sentinels = new ArrayList<RedisServer>();
+		for (Map<String, String> info : source) {
+			sentinels.add(RedisServer.newServerFrom(Converters.toProperties(info)));
+		}
+		return sentinels;
+	}
+
+	public static DataAccessException toDataAccessException(Exception ex) {
+		return EXCEPTION_CONVERTER.convert(ex);
+	}
+
 	public static LIST_POSITION toListPosition(Position source) {
 		Assert.notNull("list positions are mandatory");
 		return (Position.AFTER.equals(source) ? LIST_POSITION.AFTER : LIST_POSITION.BEFORE);

src/main/java/org/springframework/data/redis/connection/jedis/JedisSentinelConnection.java

@@ -0,0 +1,162 @@
+/*
+ * Copyright 2014 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.jedis;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.springframework.data.redis.connection.NamedNode;
+import org.springframework.data.redis.connection.RedisNode;
+import org.springframework.data.redis.connection.RedisSentinelCommands;
+import org.springframework.data.redis.connection.RedisSentinelConnection;
+import org.springframework.data.redis.connection.RedisServer;
+import org.springframework.util.Assert;
+
+import redis.clients.jedis.Jedis;
+
+/**
+ * @author Christoph Strobl
+ * @since 1.4
+ */
+public class JedisSentinelConnection implements RedisSentinelConnection {
+
+	private Jedis jedis;
+
+	public JedisSentinelConnection(RedisNode sentinel) {
+		this(sentinel.getHost(), sentinel.getPort());
+	}
+
+	public JedisSentinelConnection(String host, int port) {
+		this(new Jedis(host, port));
+	}
+
+	public JedisSentinelConnection(Jedis jedis) {
+
+		Assert.notNull(jedis, "Cannot created JedisSentinelConnection using 'null' as client.");
+		this.jedis = jedis;
+		init();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisSentinelCommands#failover(org.springframework.data.redis.connection.NamedNode)
+	 */
+	@Override
+	public void failover(NamedNode master) {
+
+		Assert.notNull(master, "Redis node master must not be 'null' for failover.");
+		Assert.hasText(master.getName(), "Redis master name must not be 'null' or empty for failover.");
+		jedis.sentinelFailover(master.getName());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisSentinelCommands#masters()
+	 */
+	@Override
+	public List<RedisServer> masters() {
+		return JedisConverters.toListOfRedisServer(jedis.sentinelMasters());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisSentinelCommands#slaves(org.springframework.data.redis.connection.NamedNode)
+	 */
+	@Override
+	public List<RedisServer> slaves(NamedNode master) {
+
+		Assert.notNull(master, "Master node cannot be 'null' when loading slaves.");
+		return slaves(master.getName());
+	}
+
+	/**
+	 * @param masterName
+	 * @see RedisSentinelCommands#slaves(NamedNode)
+	 * @return
+	 */
+	public List<RedisServer> slaves(String masterName) {
+
+		Assert.hasText(masterName, "Name of redis master cannot be 'null' or empty when loading slaves.");
+		return JedisConverters.toListOfRedisServer(jedis.sentinelSlaves(masterName));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisSentinelCommands#remove(org.springframework.data.redis.connection.NamedNode)
+	 */
+	@Override
+	public void remove(NamedNode master) {
+
+		Assert.notNull(master, "Master node cannot be 'null' when trying to remove.");
+		remove(master.getName());
+	}
+
+	/**
+	 * @param masterName
+	 * @see RedisSentinelCommands#remove(NamedNode)
+	 */
+	public void remove(String masterName) {
+
+		Assert.hasText(masterName, "Name of redis master cannot be 'null' or empty when trying to remove.");
+		jedis.sentinelRemove(masterName);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisSentinelCommands#monitor(org.springframework.data.redis.connection.RedisServer)
+	 */
+	@Override
+	public void monitor(RedisServer server) {
+
+		Assert.notNull(server, "Cannot monitor 'null' server.");
+		Assert.hasText(server.getName(), "Name of server to monitor must not be 'null' or empty.");
+		Assert.hasText(server.getHost(), "Host must not be 'null' for server to monitor.");
+		Assert.notNull(server.getPort(), "Port must not be 'null' for server to monitor.");
+		Assert.notNull(server.getQuorum(), "Quorum must not be 'null' for server to monitor.");
+		jedis.sentinelMonitor(server.getName(), server.getHost(), server.getPort().intValue(), server.getQuorum()
+				.intValue());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.io.Closeable#close()
+	 */
+	@Override
+	public void close() throws IOException {
+		jedis.close();
+	}
+
+	private void init() {
+		if (!jedis.isConnected()) {
+			doInit(jedis);
+		}
+	}
+
+	/**
+	 * Do what ever is required to establish the connection to redis.
+	 * 
+	 * @param jedis
+	 */
+	protected void doInit(Jedis jedis) {
+		jedis.connect();
+	}
+
+	@Override
+	public boolean isOpen() {
+		return jedis != null && jedis.isConnected();
+	}
+
+}

src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java

@@ -37,10 +37,10 @@ import org.jredis.protocol.Command;
 import org.jredis.ri.alphazero.JRedisSupport;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.RedisSystemException;
+import org.springframework.data.redis.connection.AbstractRedisConnection;
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.data.redis.connection.Pool;
-import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.ReturnType;
 import org.springframework.data.redis.connection.SortParameters;
 import org.springframework.data.redis.connection.Subscription;
@@ -59,7 +59,7 @@ import org.springframework.util.ReflectionUtils;
  * @author Christoph Strobl
  * @author Thomas Darimont
  */
-public class JredisConnection implements RedisConnection {
+public class JredisConnection extends AbstractRedisConnection {
 
 	private static final Method SERVICE_REQUEST;
 
@@ -120,6 +120,8 @@ public class JredisConnection implements RedisConnection {
 	}
 
 	public void close() throws RedisSystemException {
+		super.close();
+
 		if (isClosed()) {
 			return;
 		}
@@ -1275,4 +1277,5 @@ public class JredisConnection implements RedisConnection {
 	public Cursor<Entry<byte[], byte[]>> hScan(byte[] key, ScanOptions options) {
 		throw new UnsupportedOperationException("'HSCAN' command is not uspported for jredis");
 	}
+
 }

src/main/java/org/springframework/data/redis/connection/jredis/JredisConnectionFactory.java

@@ -28,6 +28,7 @@ import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.connection.Pool;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisSentinelConnection;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -201,4 +202,9 @@ public class JredisConnectionFactory implements InitializingBean, DisposableBean
 	public boolean getConvertPipelineAndTxResults() {
 		return false;
 	}
+
+	@Override
+	public RedisSentinelConnection getSentinelConnection() {
+		throw new UnsupportedOperationException();
+	}
 }

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

@@ -42,12 +42,12 @@ import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.dao.QueryTimeoutException;
 import org.springframework.data.redis.ExceptionTranslationStrategy;
-import org.springframework.data.redis.RedisConnectionFailureException;
 import org.springframework.data.redis.FallbackExceptionTranslationStrategy;
+import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.connection.AbstractRedisConnection;
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.FutureResult;
 import org.springframework.data.redis.connection.MessageListener;
-import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisPipelineException;
 import org.springframework.data.redis.connection.RedisSubscribedConnectionException;
 import org.springframework.data.redis.connection.ReturnType;
@@ -100,7 +100,7 @@ import com.lambdaworks.redis.pubsub.RedisPubSubConnection;
  * @author Christoph Strobl
  * @author Thomas Darimont
  */
-public class LettuceConnection implements RedisConnection {
+public class LettuceConnection extends AbstractRedisConnection {
 
 	private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy(
 			LettuceConverters.exceptionConverter());
@@ -358,6 +358,8 @@ public class LettuceConnection implements RedisConnection {
 	}
 
 	public void close() throws DataAccessException {
+		super.close();
+
 		isClosed = true;
 
 		if (asyncDedicatedConn != null) {

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

@@ -29,6 +29,7 @@ import org.springframework.data.redis.RedisConnectionFailureException;
 import org.springframework.data.redis.connection.Pool;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisSentinelConnection;
 import org.springframework.util.Assert;
 
 import com.lambdaworks.redis.RedisAsyncConnection;
@@ -338,4 +339,9 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 		client.setDefaultTimeout(timeout, TimeUnit.MILLISECONDS);
 		return client;
 	}
+
+	@Override
+	public RedisSentinelConnection getSentinelConnection() {
+		throw new UnsupportedOperationException();
+	}
 }

src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java

@@ -32,12 +32,12 @@ import java.util.concurrent.Future;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.ExceptionTranslationStrategy;
-import org.springframework.data.redis.RedisConnectionFailureException;
 import org.springframework.data.redis.FallbackExceptionTranslationStrategy;
+import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.connection.AbstractRedisConnection;
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.FutureResult;
 import org.springframework.data.redis.connection.MessageListener;
-import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisPipelineException;
 import org.springframework.data.redis.connection.RedisSubscribedConnectionException;
 import org.springframework.data.redis.connection.ReturnType;
@@ -69,7 +69,7 @@ import com.google.common.util.concurrent.ListenableFuture;
  * @author Christoph Strobl
  * @author Thomas Darimont
  */
-public class SrpConnection implements RedisConnection {
+public class SrpConnection extends AbstractRedisConnection {
 
 	private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy(
 			SrpConverters.exceptionConverter());
@@ -265,6 +265,8 @@ public class SrpConnection implements RedisConnection {
 	}
 
 	public void close() throws DataAccessException {
+
+		super.close();
 		isClosed = true;
 		queue.remove(this);
 

src/main/java/org/springframework/data/redis/connection/srp/SrpConnectionFactory.java

@@ -26,6 +26,7 @@ import org.springframework.data.redis.ExceptionTranslationStrategy;
 import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisSentinelConnection;
 
 /**
  * Connection factory creating <a href="http://github.com/spullara/redis-protocol">Redis Protocol</a> based connections.
@@ -159,4 +160,9 @@ public class SrpConnectionFactory implements InitializingBean, DisposableBean, R
 	public void setConvertPipelineAndTxResults(boolean convertPipelineAndTxResults) {
 		this.convertPipelineAndTxResults = convertPipelineAndTxResults;
 	}
+
+	@Override
+	public RedisSentinelConnection getSentinelConnection() {
+		throw new UnsupportedOperationException();
+	}
 }