DATAREDIS-443 - Add Support for Spring 4.3 synchronized mode to RedisCache.

As of Spring Framework 4.3.RC1, the `Cache` interface has a new `<T> T get(Object key, Callable<T> valueLoader);` method (see SPR-9254).

If no entry for the given key is found, the `Callable` is invoked to compute/load the value that is then put into redis and returned. Additionally concurrent calls get synchronized so that the `Callable` is only called once. 

Using Spring Framework 4.3 failures result in `o.s.c.Cache$ValueRetrievalException` prior versions use `RedisSystemException`.

Author: christophstrobl

build.gradle

@@ -100,10 +100,10 @@ dependencies {
     testCompile "junit:junit:$junitVersion"
     testCompile "org.springframework:spring-test:$springVersion"
     testCompile "org.springframework:spring-jdbc:$springVersion"
-    testCompile 'org.testinfected.hamcrest-matchers:core-matchers:1.8'
+    testCompile 'org.testinfected.hamcrest-matchers:hamcrest-matchers:1.8'
     testCompile "org.mockito:mockito-core:$mockitoVersion"
     testCompile("javax.annotation:jsr250-api:1.0", optional)
-    testCompile("com.thoughtworks.xstream:xstream:1.4.4", optional)
+    testCompile("com.thoughtworks.xstream:xstream:1.4.8", optional)
     testCompile("javax.transaction:jta:1.1")
 
     sharedResources "org.springframework.data.build:spring-data-build-resources:$springDataBuildVersion@zip"

gradle.properties

@@ -2,7 +2,7 @@ slf4jVersion=1.7.12
 junitVersion=4.12
 jredisVersion=06052013
 jedisVersion=2.7.3
-springVersion=4.1.7.RELEASE
+springVersion=4.1.9.RELEASE
 springDataBuildVersion=1.8.0.BUILD-SNAPSHOT
 log4jVersion=1.2.17
 version=1.7.0.DATAREDIS-443-SNAPSHOT

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2015 the original author or authors.
+ * Copyright 2011-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.
@@ -19,18 +19,22 @@
 import static org.springframework.util.Assert.*;
 import static org.springframework.util.ObjectUtils.*;
 
+import java.lang.reflect.Constructor;
 import java.util.Arrays;
 import java.util.Set;
+import java.util.concurrent.Callable;
 
 import org.springframework.cache.Cache;
 import org.springframework.cache.support.SimpleValueWrapper;
 import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.ReturnType;
 import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.RedisOperations;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.util.ClassUtils;
 
 /**
  * Cache implementation on top of Redis.
@@ -91,6 +95,32 @@ public ValueWrapper get(Object key) {
 				redisOperations.getKeySerializer()));
 	}
 
+	/*
+	 * @see  org.springframework.cache.Cache#get(java.lang.Object, java.util.concurrent.Callable)
+	 * introduced in springframework 4.3.0.RC1
+	 */
+	public <T> T get(final Object key, final Callable<T> valueLoader) {
+
+		BinaryRedisCacheElement rce = new BinaryRedisCacheElement(new RedisCacheElement(new RedisCacheKey(key).usePrefix(
+				cacheMetadata.getKeyPrefix()).withKeySerializer(redisOperations.getKeySerializer()), valueLoader),
+				cacheValueAccessor);
+
+		ValueWrapper val = get(key);
+		if (val != null) {
+			return (T) val.get();
+		}
+
+		RedisWriteThroughCallback callback = new RedisWriteThroughCallback(rce, cacheMetadata);
+
+		try {
+			byte[] result = (byte[]) redisOperations.execute(callback);
+			return (T) (result == null ? null : cacheValueAccessor.deserializeIfNecessary(result));
+		} catch (RuntimeException e) {
+			throw CacheValueRetrievalExceptionFactory.INSTANCE.create(key, valueLoader, e);
+		}
+
+	}
+
 	/**
 	 * Return the value to which this cache maps the specified key.
 	 * 
@@ -361,13 +391,18 @@ static class BinaryRedisCacheElement extends RedisCacheElement {
 		private byte[] keyBytes;
 		private byte[] valueBytes;
 		private RedisCacheElement element;
+		private boolean lazyLoad;
+		private CacheValueAccessor accessor;
 
 		public BinaryRedisCacheElement(RedisCacheElement element, CacheValueAccessor accessor) {
 
 			super(element.getKey(), element.get());
 			this.element = element;
 			this.keyBytes = element.getKeyBytes();
-			this.valueBytes = accessor.convertToBytesIfNecessary(element.get());
+			this.accessor = accessor;
+
+			lazyLoad = element.get() instanceof Callable;
+			this.valueBytes = lazyLoad ? null : accessor.convertToBytesIfNecessary(element.get());
 		}
 
 		@Override
@@ -393,9 +428,16 @@ public RedisCacheElement expireAfter(long seconds) {
 
 		@Override
 		public byte[] get() {
+
+			if (lazyLoad && valueBytes == null) {
+				try {
+					valueBytes = accessor.convertToBytesIfNecessary(((Callable<?>) element.get()).call());
+				} catch (Exception e) {
+					throw e instanceof RuntimeException ? (RuntimeException) e : new RuntimeException(e.getMessage(), e);
+				}
+			}
 			return valueBytes;
 		}
-
 	}
 
 	/**
@@ -470,6 +512,15 @@ protected boolean waitForLock(RedisConnection connection) {
 
 			return foundLock;
 		}
+
+		protected void lock(RedisConnection connection) {
+			waitForLock(connection);
+			connection.set(cacheMetadata.getCacheLockKey(), "locked".getBytes());
+		}
+
+		protected void unlock(RedisConnection connection) {
+			connection.del(cacheMetadata.getCacheLockKey());
+		}
 	}
 
 	/**
@@ -666,4 +717,86 @@ private byte[] put(BinaryRedisCacheElement element, RedisConnection connection)
 		}
 	}
 
+	/**
+	 * @author Christoph Strobl
+	 * @since 1.7
+	 */
+	static class RedisWriteThroughCallback extends AbstractRedisCacheCallback<byte[]> {
+
+		public RedisWriteThroughCallback(BinaryRedisCacheElement element, RedisCacheMetadata metadata) {
+			super(element, metadata);
+		}
+
+		@Override
+		public byte[] doInRedis(BinaryRedisCacheElement element, RedisConnection connection) throws DataAccessException {
+
+			try {
+
+				lock(connection);
+
+				try {
+
+					byte[] value = connection.get(element.getKeyBytes());
+
+					if (value != null) {
+						return value;
+					}
+
+					connection.watch(element.getKeyBytes());
+					connection.multi();
+
+					value = element.get();
+					connection.set(element.getKeyBytes(), value);
+
+					processKeyExpiration(element, connection);
+					maintainKnownKeys(element, connection);
+
+					connection.exec();
+
+					return value;
+				} catch (RuntimeException e) {
+
+					connection.discard();
+					throw e;
+				}
+			} finally {
+				unlock(connection);
+			}
+		}
+	};
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 1.7 (TODO: remove when upgrading to spring 4.3)
+	 */
+	private static enum CacheValueRetrievalExceptionFactory {
+
+		INSTANCE;
+
+		private static boolean isSpring43;
+
+		static {
+			isSpring43 = ClassUtils.isPresent("org.springframework.cache.Cache$ValueRetrievalException",
+					ClassUtils.getDefaultClassLoader());
+		}
+
+		public RuntimeException create(Object key, Callable<?> valueLoader, Throwable cause) {
+
+			if (isSpring43) {
+				try {
+					Class<?> execption = ClassUtils.forName("org.springframework.cache.Cache$ValueRetrievalException", this
+							.getClass().getClassLoader());
+					Constructor<?> c = ClassUtils.getConstructorIfAvailable(execption, Object.class, Callable.class,
+							Throwable.class);
+					return (RuntimeException) c.newInstance(key, valueLoader, cause);
+				} catch (Exception ex) {
+					// ignore
+				}
+			}
+
+			return new RedisSystemException(String.format("Value for key '%s' could not be loaded using '%s'.", key,
+					valueLoader), cause);
+		}
+	}
+
 }

src/test/java/org/springframework/data/redis/cache/RedisCacheTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-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.
@@ -26,10 +26,15 @@
 import static org.springframework.data.redis.matcher.RedisTestMatchers.*;
 
 import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 
+import org.hamcrest.core.IsEqual;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.Test;
@@ -39,6 +44,7 @@
 import org.springframework.cache.Cache;
 import org.springframework.cache.Cache.ValueWrapper;
 import org.springframework.data.redis.ConnectionFactoryTracker;
+import org.springframework.data.redis.LongObjectFactory;
 import org.springframework.data.redis.ObjectFactory;
 import org.springframework.data.redis.core.AbstractOperationsTestParams;
 import org.springframework.data.redis.core.RedisTemplate;
@@ -274,4 +280,53 @@ public void putIfAbsentShouldSetValueOnlyIfNotPresent() {
 
 		assertThat(wrapper.get(), equalTo(value));
 	}
+
+	/**
+	 * @see DATAREDIS-443
+	 */
+	@Test
+	public void testCacheGetSynchronized() throws InterruptedException {
+
+		assumeThat(cache, instanceOf(RedisCache.class));
+		assumeThat(valueFactory, instanceOf(LongObjectFactory.class));
+
+		int threadCount = 10;
+		final AtomicLong counter = new AtomicLong();
+		final List<Object> results = new CopyOnWriteArrayList<Object>();
+		final CountDownLatch latch = new CountDownLatch(threadCount);
+
+		final RedisCache redisCache = (RedisCache) cache;
+
+		final Object key = getKey();
+
+		Runnable run = new Runnable() {
+			@Override
+			public void run() {
+				try {
+					Long value = redisCache.get(key, new Callable<Long>() {
+						@Override
+						public Long call() throws Exception {
+
+							Thread.sleep(333); // make sure the thread will overlap
+							return counter.incrementAndGet();
+						}
+					});
+					results.add(value);
+				} finally {
+					latch.countDown();
+				}
+			}
+		};
+
+		for (int i = 0; i < threadCount; i++) {
+			new Thread(run).start();
+			Thread.sleep(100);
+		}
+		latch.await();
+
+		assertThat(results.size(), IsEqual.equalTo(threadCount));
+		for (Object result : results) {
+			assertThat((Long) result, equalTo(1L));
+		}
+	}
 }