DATAREDIS-481 - Polishing.

Reduce visibility of DefaultRedisCacheWriter as trivial implementation. Replace ConnectionCallback with Java 8 functions. Increase visibility of the RedisCache constructor to allow subclassing. Move factory methods to create DefaultRedisCacheWriter to RedisCacheWriter interface.

Refactor RedisCacheManagerConfigurator to stateful RedisCacheManagerBuilder. Remove RedisCache.setTransactionAware override in RedisCacheManagerConfigurator removing behavior not compliant with AbstractTransactionSupportingCacheManager. Terminate interrupted cache wait loop with exception. Javadoc, formatting, reorder methods.

Allow ConversionService configuration via RedisCacheConfiguration. Accept cache keys that are either convertible to String or that override toString to obtain the String representation of the cache key.

Original pull request: #252.

Author: mp911de

Diff for: src/main/asciidoc/new-features.adoc

@@ -11,6 +11,7 @@ New and noteworthy in the latest releases.
 * Upgrade to `Lettuce` 5.0.
 * Reactive connection support using https://github.com/lettuce-io/lettuce-core[lettuce-io/lettuce-core].
 * Introduce Redis feature-specific interfaces for `RedisConnection`.
+* Revised `RedisCache` implementation.
 
 
 [[new-in-1.8.0]]

Diff for: src/main/asciidoc/reference/redis.adoc

@@ -473,19 +473,19 @@ Spring Redis provides an implementation for Spring http://docs.spring.io/spring/
 ----
 @Bean
 public RedisCacheManager cacheManager(RedisConnectionFactory connectionFactory) {
-	return RedisCacheManager.defaultCacheManager(connectionFactory);
+	return RedisCacheManager.create(connectionFactory);
 }
 ----
 
-`RedisCacheManager` behavior can be configured via `RedisCacheManagerConfigurator` allowing to set the default `RedisCacheConfiguration`, transaction behaviour and predefined caches.
+`RedisCacheManager` behavior can be configured via `RedisCacheManagerBuilder` allowing to set the default `RedisCacheConfiguration`, transaction behaviour and predefined caches.
 
 [source,java]
 ----
-RedisCacheManager cm = RedisCacheManager.usingRawConnectionFactory(connectionFactory)
-	.withCacheDefaults(defaultCacheConfig())
-	.withInitialCacheConfigurations(singletonMap("predefined", defaultCacheConfig().disableCachingNullValues()))
+RedisCacheManager cm = RedisCacheManager.builder(connectionFactory)
+	.cacheDefaults(defaultCacheConfig())
+	.initialCacheConfigurations(singletonMap("predefined", defaultCacheConfig().disableCachingNullValues()))
 	.transactionAware()
-	.createAndGet();
+	.build();
 ----
 
 Behavior of `RedisCache` created via `RedisCacheManager` is defined via `RedisCacheConfiguration`. The configuration allows to set key expiration times, prefixes and ``RedisSerializer``s for converting to and from the binary storage format.
@@ -498,16 +498,15 @@ RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
 	.disableCachingNullValues();
 ----
 
-Using default `RedisCacheManager` uses a non locking `RedisCacheWriter` for reading & writing bits.
-While this ensures a max of performance the lack of entry locking can lead to overlapping, non atomic commands, for `putIfAbsent` and `clean` as those methods combine a series of commands sent to Redis.
-The locking counterpart prevents command overlap by setting an explicit lock key and checking against presence of this key which leads to additional requests and potential command wait times.
+`RedisCacheManager` defaults to a lock-free `RedisCacheWriter` for reading & writing binary values. Lock-free caching improves throughput. The lack of entry locking can lead to overlapping, non atomic commands, for `putIfAbsent` and `clean` methods as those require multiple commands sent to Redis.
+The locking counterpart prevents command overlap by setting an explicit lock key and checking against presence of this key, which leads to additional requests and potential command wait times.
 
 It is possible to opt in to the locking behavior as follows:
 
 [source,java]
 ----
-RedisCacheManager cm = RedisCacheManager.usingCacheWriter(DefaultRedisCacheWriter.lockingRedisCacheWriter())
-	.withCacheDefaults(defaultCacheConfig())
+RedisCacheManager cm = RedisCacheManager.build(RedisCacheWriter.lockingRedisCacheWriter())
+	.cacheDefaults(defaultCacheConfig())
 	...
 ----
 
@@ -521,7 +520,7 @@ RedisCacheManager cm = RedisCacheManager.usingCacheWriter(DefaultRedisCacheWrite
 |non locking
 
 |Cache Configuration
-|`RedisCacheConfiguraiton#defaultConfiguration`
+|`RedisCacheConfiguration#defaultConfiguration`
 
 |Initial Caches
 |none
@@ -550,5 +549,8 @@ RedisCacheManager cm = RedisCacheManager.usingCacheWriter(DefaultRedisCacheWrite
 
 |Value Serializer
 |`JdkSerializationRedisSerializer`
+
+|Conversion Service
+|`DefaultFormattingConversionService` with default cache key converters
 |====
 

Diff for: src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java

@@ -15,36 +15,36 @@
  */
 package org.springframework.data.redis.cache;
 
-import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.Function;
 
-import org.springframework.data.redis.connection.RedisClusterConnection;
+import org.springframework.dao.PessimisticLockingFailureException;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
-import org.springframework.data.redis.connection.ReturnType;
 import org.springframework.data.redis.core.types.Expiration;
 import org.springframework.util.Assert;
 
 /**
  * {@link RedisCacheWriter} implementation capable of reading/writing binary data from/to Redis in {@literal standalone}
  * and {@literal cluster} environments. Works upon a given {@link RedisConnectionFactory} to obtain the actual
  * {@link RedisConnection}. <br />
- * {@link DefaultRedisCacheWriter} can be used in {@link #lockingRedisCacheWriter(RedisConnectionFactory) locking} or
- * {@link #nonLockingRedisCacheWriter(RedisConnectionFactory) non-locking} mode. While {@literal non-locking} aims for
- * maximum performance it may result in overlapping, non atomic, command execution for operations spanning multiple
- * Redis interactions like {@code putIfAbsent}. The {@literal locking} counterpart prevents command overlap by setting
- * an explicit lock key and checking against presence of this key which leads to additional requests and potential
- * command wait times.
+ * {@link DefaultRedisCacheWriter} can be used in
+ * {@link RedisCacheWriter#lockingRedisCacheWriter(RedisConnectionFactory) locking} or
+ * {@link RedisCacheWriter#nonLockingRedisCacheWriter(RedisConnectionFactory) non-locking} mode. While
+ * {@literal non-locking} aims for maximum performance it may result in overlapping, non atomic, command execution for
+ * operations spanning multiple Redis interactions like {@code putIfAbsent}. The {@literal locking} counterpart prevents
+ * command overlap by setting an explicit lock key and checking against presence of this key which leads to additional
+ * requests and potential command wait times.
  *
  * @author Christoph Strobl
+ * @author Mark Paluch
  * @since 2.0
  */
-public class DefaultRedisCacheWriter implements RedisCacheWriter {
-
-	private static final byte[] CLEAN_SCRIPT = "local keys = redis.call('KEYS', ARGV[1]); local keysCount = table.getn(keys); if(keysCount > 0) then for _, key in ipairs(keys) do redis.call('del', key); end; end; return keysCount;"
-			.getBytes(Charset.forName("UTF-8"));
+class DefaultRedisCacheWriter implements RedisCacheWriter {
 
 	private final RedisConnectionFactory connectionFactory;
 	private final Duration sleepTime;
@@ -70,30 +70,10 @@ public class DefaultRedisCacheWriter implements RedisCacheWriter {
 		this.sleepTime = sleepTime;
 	}
 
-	/**
-	 * Create new {@link RedisCacheWriter} without locking behavior.
-	 *
-	 * @param connectionFactory must not be {@literal null}.
-	 * @return new instance of {@link DefaultRedisCacheWriter}.
-	 */
-	public static DefaultRedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connectionFactory) {
-
-		Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
-		return new DefaultRedisCacheWriter(connectionFactory);
-	}
-
-	/**
-	 * Create new {@link RedisCacheWriter} with locking behavior.
-	 *
-	 * @param connectionFactory must not be {@literal null}.
-	 * @return new instance of {@link DefaultRedisCacheWriter}.
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.RedisCacheWriter#put(java.lang.String, byte[], byte[], java.time.Duration)
 	 */
-	public static DefaultRedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory) {
-
-		Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
-		return new DefaultRedisCacheWriter(connectionFactory, Duration.ofMillis(50));
-	}
-
 	@Override
 	public void put(String name, byte[] key, byte[] value, Duration ttl) {
 
@@ -114,6 +94,10 @@ public void put(String name, byte[] key, byte[] value, Duration ttl) {
 		});
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.RedisCacheWriter#get(java.lang.String, byte[])
+	 */
 	@Override
 	public byte[] get(String name, byte[] key) {
 
@@ -123,6 +107,10 @@ public byte[] get(String name, byte[] key) {
 		return execute(name, connection -> connection.get(key));
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.RedisCacheWriter#putIfAbsent(java.lang.String, byte[], byte[], java.time.Duration)
+	 */
 	@Override
 	public byte[] putIfAbsent(String name, byte[] key, byte[] value, Duration ttl) {
 
@@ -156,6 +144,10 @@ public byte[] putIfAbsent(String name, byte[] key, byte[] value, Duration ttl) {
 		});
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.RedisCacheWriter#remove(java.lang.String, byte[])
+	 */
 	@Override
 	public void remove(String name, byte[] key) {
 
@@ -165,46 +157,10 @@ public void remove(String name, byte[] key) {
 		execute(name, connection -> connection.del(key));
 	}
 
-	/**
-	 * Explicitly set a write lock on a cache.
-	 *
-	 * @param name the name of the cache to lock.
-	 */
-	public void lock(String name) {
-		execute(name, connection -> doLock(name, connection));
-	}
-
-	private Boolean doLock(String name, RedisConnection connection) {
-		return connection.setNX(createCacheLockKey(name), new byte[] {});
-	}
-
-	/**
-	 * Explicitly remove a write lock from a cache.
-	 *
-	 * @param name the name of the cache to unlock.
-	 */
-	public void unlock(String name) {
-		executeWithoutLockCheck(connection -> doUnlock(name, connection));
-	}
-
-	private Long doUnlock(String name, RedisConnection connection) {
-		return connection.del(createCacheLockKey(name));
-	}
-
-	/**
-	 * Check if a cache has set a lock.
-	 *
-	 * @param name the name of the cache to check for presence of a lock.
-	 * @return {@literal true} if lock found.
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.RedisCacheWriter#clean(java.lang.String, byte[])
 	 */
-	public boolean isLoked(String name) {
-		return executeWithoutLockCheck(connection -> doCheckLock(name, connection));
-	}
-
-	private boolean doCheckLock(String name, RedisConnection connection) {
-		return connection.exists(createCacheLockKey(name));
-	}
-
 	@Override
 	public void clean(String name, byte[] pattern) {
 
@@ -216,17 +172,16 @@ public void clean(String name, byte[] pattern) {
 			boolean wasLocked = false;
 
 			try {
-				if (connection instanceof RedisClusterConnection) {
 
-					if (isLockingCacheWriter()) {
-						doLock(name, connection);
-						wasLocked = true;
-					}
+				if (isLockingCacheWriter()) {
+					doLock(name, connection);
+					wasLocked = true;
+				}
+
+				byte[][] keys = connection.keys(pattern).toArray(new byte[0][]);
 
-					byte[][] keys = connection.keys(pattern).stream().toArray(size -> new byte[size][]);
+				if (keys.length > 0) {
 					connection.del(keys);
-				} else {
-					connection.eval(CLEAN_SCRIPT, ReturnType.INTEGER, 0, pattern);
 				}
 			} finally {
 
@@ -239,67 +194,92 @@ public void clean(String name, byte[] pattern) {
 		});
 	}
 
+	/**
+	 * Explicitly set a write lock on a cache.
+	 *
+	 * @param name the name of the cache to lock.
+	 */
+	void lock(String name) {
+		execute(name, connection -> doLock(name, connection));
+	}
+
+	/**
+	 * Explicitly remove a write lock from a cache.
+	 *
+	 * @param name the name of the cache to unlock.
+	 */
+	void unlock(String name) {
+		executeLockFree(connection -> doUnlock(name, connection));
+	}
+
+	private Boolean doLock(String name, RedisConnection connection) {
+		return connection.setNX(createCacheLockKey(name), new byte[0]);
+	}
+
+	private Long doUnlock(String name, RedisConnection connection) {
+		return connection.del(createCacheLockKey(name));
+	}
+
+	boolean doCheckLock(String name, RedisConnection connection) {
+		return connection.exists(createCacheLockKey(name));
+	}
+
 	/**
 	 * @return {@literal true} if {@link RedisCacheWriter} uses locks.
 	 */
-	public boolean isLockingCacheWriter() {
+	private boolean isLockingCacheWriter() {
 		return !sleepTime.isZero() && !sleepTime.isNegative();
 	}
 
-	<T> T execute(String name, ConnectionCallback<T> callback) {
+	private <T> T execute(String name, Function<RedisConnection, T> callback) {
 
 		RedisConnection connection = connectionFactory.getConnection();
 		try {
 
 			checkAndPotentiallyWaitUntilUnlocked(name, connection);
-			return callback.doWithConnection(connection);
+			return callback.apply(connection);
 		} finally {
 			connection.close();
 		}
 	}
 
-	private <T> T executeWithoutLockCheck(ConnectionCallback<T> callback) {
+	private void executeLockFree(Consumer<RedisConnection> callback) {
 
 		RedisConnection connection = connectionFactory.getConnection();
 
 		try {
-			return callback.doWithConnection(connection);
+			callback.accept(connection);
 		} finally {
 			connection.close();
 		}
 	}
 
 	private void checkAndPotentiallyWaitUntilUnlocked(String name, RedisConnection connection) {
 
-		if (isLockingCacheWriter()) {
+		if (!isLockingCacheWriter()) {
+			return;
+		}
 
-			long timeout = sleepTime.toMillis();
+		try {
 
 			while (doCheckLock(name, connection)) {
-				try {
-					Thread.sleep(timeout);
-				} catch (InterruptedException ex) {
-					Thread.currentThread().interrupt();
-				}
+				Thread.sleep(sleepTime.toMillis());
 			}
+		} catch (InterruptedException ex) {
+
+			// Re-interrupt current thread, to allow other participants to react.
+			Thread.currentThread().interrupt();
+
+			throw new PessimisticLockingFailureException(String.format("Interrupted while waiting to unlock cache %s", name),
+					ex);
 		}
 	}
 
-	private boolean shouldExpireWithin(Duration ttl) {
+	private static boolean shouldExpireWithin(Duration ttl) {
 		return ttl != null && !ttl.isZero() && !ttl.isNegative();
 	}
 
-	byte[] createCacheLockKey(String name) {
-		return (name + "~lock").getBytes(Charset.forName("UTF-8"));
-	}
-
-	/**
-	 * @param <T>
-	 * @author Christoph Strobl
-	 * @since 2.0
-	 */
-	interface ConnectionCallback<T> {
-
-		T doWithConnection(RedisConnection connection);
+	private static byte[] createCacheLockKey(String name) {
+		return (name + "~lock").getBytes(StandardCharsets.UTF_8);
 	}
 }