DATAREDIS-468 - Polishing.

Add DecoratedRedisConnection interface to unwrap decorated connections. When running on Redis Cluster, use keys/del instead of lua script because cluster connections do not support lua script execution. Original pull request: spring-projects#173.

Author: mp911de

src/main/java/org/springframework/data/redis/cache/RedisCache.java

@@ -28,6 +28,7 @@
 import org.springframework.cache.support.SimpleValueWrapper;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.RedisSystemException;
+import org.springframework.data.redis.connection.DecoratedRedisConnection;
 import org.springframework.data.redis.connection.RedisClusterConnection;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.ReturnType;
@@ -39,10 +40,11 @@
 
 /**
  * Cache implementation on top of Redis.
- * 
+ *
  * @author Costin Leau
  * @author Christoph Strobl
  * @author Thomas Darimont
+ * @author Mark Paluch
  */
 @SuppressWarnings("unchecked")
 public class RedisCache implements Cache {
@@ -54,7 +56,7 @@ public class RedisCache implements Cache {
 
 /**
  * Constructs a new <code>RedisCache</code> instance.
- * 
+ *
  * @param name cache name
  * @param prefix
  * @param redisOperations
@@ -624,7 +626,16 @@ public Void doInLock(RedisConnection connection) throws DataAccessException {
 byte[] prefixToUse = Arrays.copyOf(metadata.getKeyPrefix(), metadata.getKeyPrefix().length + WILD_CARD.length);
 System.arraycopy(WILD_CARD, 0, prefixToUse, metadata.getKeyPrefix().length, WILD_CARD.length);
 
-connection.eval(REMOVE_KEYS_BY_PATTERN_LUA, ReturnType.INTEGER, 0, prefixToUse);
+if(isClusterConnection(connection)) {
+
+// load keys to the client because currently Redis Cluster connections do not allow eval of lua scripts.
+Set<byte[]> keys = connection.keys(prefixToUse);
+if (!keys.isEmpty()) {
+connection.del(keys.toArray(new byte[keys.size()][]));
+}
+} else {
+connection.eval(REMOVE_KEYS_BY_PATTERN_LUA, ReturnType.INTEGER, 0, prefixToUse);
+}
 
 return null;
 }
@@ -765,8 +776,9 @@ public byte[] doInRedis(BinaryRedisCacheElement element, RedisConnection connect
 
 return value;
 } catch (RuntimeException e) {
-
-connection.discard();
+if (!isClusterConnection(connection)) {
+connection.discard();
+}
 throw e;
 }
 } finally {
@@ -810,6 +822,11 @@ public RuntimeException create(Object key, Callable<?> valueLoader, Throwable ca
 }
 
 private static boolean isClusterConnection(RedisConnection connection) {
+
+while (connection instanceof DecoratedRedisConnection) {
+connection = ((DecoratedRedisConnection) connection).getDelegate();
+}
+
 return connection instanceof RedisClusterConnection;
 }
 

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2016 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;
+
+/**
+ * Specifies that the connection decorates another {@link RedisConnection}.
+ *
+ * @author Mark Paluch
+ * @since 1.7
+ */
+public interface DecoratedRedisConnection {
+
+/**
+ * Gets the underlying {@link RedisConnection}.
+ *
+ * @return never {@literal null}.
+ */
+RedisConnection getDelegate();
+
+}

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

@@ -49,8 +49,9 @@
  * @author Jennifer Hickey
  * @author Christoph Strobl
  * @author Thomas Darimont
+ * @author Mark Paluch
  */
-public class DefaultStringRedisConnection implements StringRedisConnection {
+public class DefaultStringRedisConnection implements StringRedisConnection, DecoratedRedisConnection {
 
 private static final byte[][] EMPTY_2D_BYTE_ARRAY = new byte[0][];
 
@@ -2724,4 +2725,13 @@ public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption opt
 delegate.migrate(key, target, dbIndex, option, timeout);
 }
 
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.DecoratedRedisConnection#getDelegate()
+ */
+@Override
+public RedisConnection getDelegate() {
+return delegate;
+}
+
 }