From be40b0853a43eae4f3020b0a2c278c21bb5574ab Mon Sep 17 00:00:00 2001
From: Mark Paluch <mpaluch@pivotal.io>
Date: Fri, 5 Feb 2016 08:25:16 +0100
Subject: [PATCH] DATAREDIS-452 - Improve thread synchronization in
 RedisCacheTest.testCacheGetSynchronized.

Original Pull Request: #165
---
 pom.xml                                       |  8 ++
 .../data/redis/cache/RedisCacheTest.java      | 89 +++++++++++--------
 2 files changed, 61 insertions(+), 36 deletions(-)

diff --git a/pom.xml b/pom.xml
index 281705c37..eb5622e21 100644
--- a/pom.xml
+++ b/pom.xml
@@ -28,6 +28,7 @@
 		<jedis>2.8.0</jedis>
 		<srp>0.7</srp>
 		<jredis>06052013</jredis>
+		<multithreadedtc>1.01</multithreadedtc>
 	</properties>
 
 	<dependencyManagement>
@@ -181,6 +182,13 @@
 			<version>${xstream}</version>
 			<scope>test</scope>
 		</dependency>
+
+		<dependency>
+			<groupId>edu.umd.cs.mtc</groupId>
+			<artifactId>multithreadedtc</artifactId>
+			<version>${multithreadedtc}</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<build>
diff --git a/src/test/java/org/springframework/data/redis/cache/RedisCacheTest.java b/src/test/java/org/springframework/data/redis/cache/RedisCacheTest.java
index 78c6cee49..b8824b8c6 100644
--- a/src/test/java/org/springframework/data/redis/cache/RedisCacheTest.java
+++ b/src/test/java/org/springframework/data/redis/cache/RedisCacheTest.java
@@ -16,6 +16,7 @@
 
 package org.springframework.data.redis.cache;
 
+import static org.hamcrest.core.Is.is;
 import static org.hamcrest.core.IsEqual.*;
 import static org.hamcrest.core.IsInstanceOf.*;
 import static org.hamcrest.core.IsNot.*;
@@ -26,14 +27,13 @@ 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 edu.umd.cs.mtc.MultithreadedTestCase;
+import edu.umd.cs.mtc.TestFramework;
 import org.hamcrest.core.IsEqual;
 import org.junit.AfterClass;
 import org.junit.Before;
@@ -44,8 +44,8 @@ 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.StringObjectFactory;
 import org.springframework.data.redis.core.AbstractOperationsTestParams;
 import org.springframework.data.redis.core.RedisTemplate;
 
@@ -53,6 +53,7 @@ import org.springframework.data.redis.core.RedisTemplate;
  * @author Costin Leau
  * @author Jennifer Hickey
  * @author Christoph Strobl
+ * @author Mark Paluch
  */
 @SuppressWarnings("rawtypes")
 @RunWith(Parameterized.class)
@@ -283,50 +284,66 @@ public class RedisCacheTest extends AbstractNativeCacheTest<RedisTemplate> {
 
 	/**
 	 * @see DATAREDIS-443
+	 * @see DATAREDIS-452
 	 */
 	@Test
-	public void testCacheGetSynchronized() throws InterruptedException {
+	public void testCacheGetSynchronized() throws Throwable {
 
 		assumeThat(cache, instanceOf(RedisCache.class));
-		assumeThat(valueFactory, instanceOf(LongObjectFactory.class));
+		assumeThat(valueFactory, instanceOf(StringObjectFactory.class));
 
-		int threadCount = 10;
-		final AtomicLong counter = new AtomicLong();
-		final List<Object> results = new CopyOnWriteArrayList<Object>();
-		final CountDownLatch latch = new CountDownLatch(threadCount);
+		TestFramework.runOnce(new CacheGetWithValueLoaderIsThreadSafe((RedisCache) cache));
+	}
 
-		final RedisCache redisCache = (RedisCache) cache;
+	static class CacheGetWithValueLoaderIsThreadSafe extends MultithreadedTestCase {
 
-		final Object key = getKey();
+		RedisCache redisCache;
+		TestCacheLoader<String> cacheLoader;
 
-		Runnable run = new Runnable() {
-			@Override
-			public void run() {
-				try {
-					Long value = redisCache.get(key, new Callable<Long>() {
-						@Override
-						public Long call() throws Exception {
+		public CacheGetWithValueLoaderIsThreadSafe(RedisCache redisCache) {
+			this.redisCache = redisCache;
 
-							Thread.sleep(333); // make sure the thread will overlap
-							return counter.incrementAndGet();
-						}
-					});
-					results.add(value);
-				} finally {
-					latch.countDown();
+			cacheLoader = new TestCacheLoader<String>("test"){
+
+				@Override
+				public String call() throws Exception {
+					waitForTick(2);
+					return super.call();
 				}
-			}
-		};
-
-		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));
+		public void thread1(){
+			assertTick(0);
+ 			Thread.currentThread().setName(getClass().getSimpleName() + " Cache Loader Thread");
+
+			String result = redisCache.get("key", cacheLoader);
+
+			assertThat(result, is("test"));
+		}
+
+		public void thread2(){
+			waitForTick(1);
+			Thread.currentThread().setName(getClass().getSimpleName() + " Cache Reader Thread");
+
+			String result = redisCache.get("key", new TestCacheLoader<String>("illegal value"));
+
+			assertThat(result, is("test"));
+			assertTick(2);
 		}
 	}
+
+	protected static class TestCacheLoader<T> implements Callable<T> {
+
+ 		private final T value;
+
+ 		public TestCacheLoader(T value) {
+ 			this.value = value;
+ 		}
+
+ 		@Override
+ 		public T call() throws Exception {
+ 			return value;
+ 		}
+ 	}
 }