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`. Original pull request: #162.
2016-01-08 14:49:35 +01:00
parent e7bfa2f26e
commit a03c444298
4 changed files with 271 additions and 7 deletions
--- a/build.gradle
+++ b/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"
--- a/src/main/java/org/springframework/data/redis/cache/RedisCache.java
+++ b/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 @@ package org.springframework.data.redis.cache;
 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 class RedisCache implements Cache {
 				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 @@ public class RedisCache implements Cache {
 		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 class RedisCache implements Cache {

 		@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 @@ public class RedisCache implements Cache {

 			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 @@ public class RedisCache implements Cache {
 		}
 	}

+	/**
+	 * @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);
+		}
+	}
+
 }
--- a/src/test/java/org/springframework/data/redis/cache/RedisCacheTest.java
+++ b/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.junit.Assume.*;
 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.junit.runners.Parameterized.Parameters;
 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 class RedisCacheTest extends AbstractNativeCacheTest<RedisTemplate> {

 		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));
+		}
+	}
 }
--- a/src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -15,14 +15,22 @@
 */
 package org.springframework.data.redis.cache;

+import static org.hamcrest.core.IsEqual.*;
+import static org.junit.Assert.*;
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
+import static org.springframework.util.ClassUtils.*;
+
+import java.util.concurrent.Callable;

 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
+import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.ReturnType;
@@ -59,6 +67,8 @@ public class RedisCacheUnitTests {

 	RedisCache cache;

+	public @Rule ExpectedException exception = ExpectedException.none();
+
 	@SuppressWarnings("unchecked")
 	@Before
 	public void setUp() {
@@ -75,6 +85,7 @@ public class RedisCacheUnitTests {

 		when(keySerializerMock.serialize(any(byte[].class))).thenReturn(KEY_BYTES);
 		when(valueSerializerMock.serialize(any(byte[].class))).thenReturn(VALUE_BYTES);
+		when(valueSerializerMock.deserialize(eq(VALUE_BYTES))).thenReturn(VALUE);
 	}

 	/**
@@ -141,4 +152,69 @@ public class RedisCacheUnitTests {

 		verify(connectionMock, never()).expire(eq(KNOWN_KEYS_SET_NAME_BYTES), anyLong());
 	}
+
+	/**
+	 * @see DATAREDIS-443
+	 */
+	@Test
+	@SuppressWarnings("unchecked")
+	public void getWithCallable() throws ClassNotFoundException, LinkageError {
+
+		if (isPresent("org.springframework.cache.Cache$ValueRetrievalException", getDefaultClassLoader())) {
+			exception.expect((Class<? extends Throwable>) forName("org.springframework.cache.Cache$ValueRetrievalException",
+					getDefaultClassLoader()));
+		} else {
+			exception.expect(RedisSystemException.class);
+		}
+
+		exception.expectMessage("Value for key 'key' could not be loaded");
+
+		cache = new RedisCache(CACHE_NAME, NO_PREFIX_BYTES, templateSpy, 0L);
+
+		cache.get(KEY, new Callable<Object>() {
+			@Override
+			public Object call() throws Exception {
+				throw new UnsupportedOperationException("Expected exception");
+			}
+		});
+	}
+
+	/**
+	 * @see DATAREDIS-443
+	 */
+	@Test
+	public void getWithCallableShouldReadValueFromCallableAddToCache() {
+
+		cache = new RedisCache(CACHE_NAME, NO_PREFIX_BYTES, templateSpy, 0L);
+
+		cache.get(KEY, new Callable<Object>() {
+			@Override
+			public Object call() throws Exception {
+				return VALUE;
+			}
+		});
+
+		verify(connectionMock, times(2)).get(eq(KEY_BYTES));
+		verify(connectionMock, times(1)).multi();
+		verify(connectionMock, times(1)).set(eq(KEY_BYTES), eq(VALUE_BYTES));
+		verify(connectionMock, times(1)).exec();
+	}
+
+	/**
+	 * @see DATAREDIS-443
+	 */
+	@Test
+	@SuppressWarnings("unchecked")
+	public void getWithCallableShouldNotReadValueFromCallableWhenAlreadyPresent() {
+
+		cache = new RedisCache(CACHE_NAME, NO_PREFIX_BYTES, templateSpy, 0L);
+		Callable<Object> callableMock = mock(Callable.class);
+
+		when(connectionMock.exists(KEY_BYTES)).thenReturn(true);
+		when(connectionMock.get(KEY_BYTES)).thenReturn(null).thenReturn(VALUE_BYTES);
+
+		assertThat((String) cache.get(KEY, callableMock), equalTo(VALUE));
+		verifyZeroInteractions(callableMock);
+	}
+
 }