+ introduce Session callback to allow the same connection to be reused

+ various improvements to the underlying connection handling infrastructure
+ updated AtomicCounterInteger to take advantage of thew new infrastructure
+ improved tests by refactoring some common functionality

spring-data-redis/pom.xml

@@ -133,7 +133,7 @@
 			</plugin>
 		</plugins>
 	</build>
-	
+<!--	
 	<repositories>
    <repository>
       <id>oss-snapshots</id>
@@ -144,5 +144,5 @@
       </snapshots>
    </repository>
   </repositories>
-
+-->
 </project>

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java

@@ -478,8 +478,12 @@ public class JedisConnection implements RedisConnection {
 
 	@Override
 	public void multi() {
+		if (isQueueing()) {
+			return;
+		}
+
 		try {
-			client.multi();
+			jedis.multi();
 		} catch (Exception ex) {
 			throw convertJedisAccessException(ex);
 		}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisConnectionUtils.java

@@ -33,6 +33,16 @@ public abstract class RedisConnectionUtils {
 
 	private static final Log log = LogFactory.getLog(RedisConnectionUtils.class);
 
+	/**
+	 * Binds a new Redis connection (from the given factory) to the current thread, if none is already bound. 
+	 * 
+	 * @param factory connection factory
+	 * @return a new Redis connection
+	 */
+	public static RedisConnection bindConnection(RedisConnectionFactory factory) {
+		return doGetConnection(factory, true, true);
+	}
+
 	/**
 	 * Gets a Redis connection from the given factory. Is aware of and will return any existing corresponding connections bound to the current thread, 
 	 * for example when using a transaction manager. Will always create a new connection otherwise.
@@ -40,8 +50,8 @@ public abstract class RedisConnectionUtils {
 	 * @param factory connection factory for creating the connection
 	 * @return an active Redis connection
 	 */
-	public static RedisConnection getRedisConnection(RedisConnectionFactory factory) {
-		return doGetRedisConnection(factory, true);
+	public static RedisConnection getConnection(RedisConnectionFactory factory) {
+		return doGetConnection(factory, true, false);
 	}
 
 	/**
@@ -49,10 +59,11 @@ public abstract class RedisConnectionUtils {
 	 * for example when using a transaction manager. Will create a new Connection otherwise, if {@code allowCreate} is <tt>true</tt>.
 	 * 
 	 * @param factory connection factory for creating the connection
-	 * @param allowCreate whether a new (unbound) connection should be created when no connection can be found for the current thread 
+	 * @param allowCreate whether a new (unbound) connection should be created when no connection can be found for the current thread
+	 * @param bind binds the connection to the thread, in case one was created
 	 * @return an active Redis connection
 	 */
-	public static RedisConnection doGetRedisConnection(RedisConnectionFactory factory, boolean allowCreate) {
+	public static RedisConnection doGetConnection(RedisConnectionFactory factory, boolean allowCreate, boolean bind) {
 		Assert.notNull(factory, "No RedisConnectionFactory specified");
 
 		RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory);
@@ -70,10 +81,14 @@ public abstract class RedisConnectionUtils {
 
 		RedisConnection conn = factory.getConnection();
 
-		if (TransactionSynchronizationManager.isSynchronizationActive()) {
+		boolean synchronizationActive = TransactionSynchronizationManager.isSynchronizationActive();
+
+		if (bind || synchronizationActive) {
 			connHolder = new RedisConnectionHolder(conn);
-			TransactionSynchronizationManager.registerSynchronization(new RedisConnectionSynchronization(connHolder,
-					factory, true));
+			if (synchronizationActive) {
+				TransactionSynchronizationManager.registerSynchronization(new RedisConnectionSynchronization(
+						connHolder, factory, true));
+			}
 			TransactionSynchronizationManager.bindResource(factory, connHolder);
 			return connHolder.getConnection();
 		}
@@ -92,11 +107,26 @@ public abstract class RedisConnectionUtils {
 		}
 		// Only release non-transactional/non-bound connections.
 		if (!isConnectionTransactional(conn, factory)) {
-			log.debug("Closing Redis Connection");
+			if (log.isDebugEnabled()) {
+				log.debug("Closing Redis Connection");
+			}
 			conn.close();
 		}
 	}
 
+	/**
+	 * Unbinds and closes the connection (if any) associated with the given factory.
+	 * 
+	 * @param factory Redis factory
+	 */
+	public static void unbindConnection(RedisConnectionFactory factory) {
+		RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.unbindResourceIfPossible(factory);
+		if (connHolder != null) {
+			RedisConnection connection = connHolder.getConnection();
+			connection.close();
+		}
+	}
+
 	/**
 	 * Return whether the given Redis connection is transactional, that is, bound to the current thread by Spring's transaction facilities. 
 	 * 

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisOperations.java

@@ -50,6 +50,19 @@ public interface RedisOperations<K, V> {
 	 */
 	<T> T execute(RedisCallback<T> action);
 
+
+	/**
+	 * Executes a Redis session.
+	 * 
+	 * Allows multiple operations to be executed in the same session enabling 'transactional' capabilities through {@link #multi()} 
+	 * and {@link #watch(Collection)} operations.
+	 *  
+	 * @param <T> return type
+	 * @param session session callback
+	 * @return result object returned by the action or <tt>null</tt>
+	 */
+	<T> T execute(SessionCallback<T> session);
+
 	Boolean hasKey(K key);
 
 	void delete(Collection<K> key);
@@ -76,12 +89,15 @@ public interface RedisOperations<K, V> {
 
 	void unwatch();
 
+	/**'
+	 * 
+	 */
 	void multi();
 
 	void discard();
 
 	Object exec();
-	
+
 	List<V> sort(K key, SortParameters params);
 
 	Long sort(K key, SortParameters params, K destination);

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java

@@ -99,6 +99,7 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 		afterPropertiesSet();
 	}
 
+	@Override
 	public <T> T execute(RedisCallback<T> action) {
 		return execute(action, isExposeConnection());
 	}
@@ -115,6 +116,19 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 		return execute(action, exposeConnection, valueSerializer);
 	}
 
+	/**
+	 * Executes the given action object within a connection, that can be pipelined or not and which can be exposed or not.
+	 * 
+	 * @param <T> return type
+	 * @param action callback object to execute
+	 * @param exposeConnection whether to enforce exposure of the native Redis Connection to callback code
+	 * @param pipeline whether to pipeline or not the connection for the execution duration 
+	 * @return object returned by the action
+	 */
+	public <T> T execute(RedisCallback<T> action, boolean exposeConnection, boolean pipeline) {
+		return execute(action, exposeConnection, pipeline, valueSerializer);
+	}
+
 	/**
 	 * Executes the given action object within a connection, which can be exposed or not. Allows a custom serializer
 	 * to be specified for the returned object.
@@ -126,10 +140,30 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 	 * @return returned by the action
 	 */
 	public <T> T execute(RedisCallback<T> action, boolean exposeConnection, RedisSerializer<?> returnSerializer) {
+		return execute(action, exposeConnection, false, returnSerializer);
+	}
+
+	/**
+	 * Executes the given action object within a connection, which can be exposed or not. Allows a custom serializer
+	 * to be specified for the returned object.
+	 * 
+	 * @param <T> return type
+	 * @param action action callback object that specifies the Redis action
+	 * @param exposeConnection whether to enforce exposure of the native Redis Connection to callback code
+	 * @param pipeline whether to pipeline or not the connection for the execution duration
+	 * @param returnSerializer serializer used for converting the binary data to the custom return type
+	 * @return returned by the action
+	 */
+	public <T> T execute(RedisCallback<T> action, boolean exposeConnection, boolean pipeline, RedisSerializer<?> returnSerializer) {
 		Assert.notNull(action, "Callback object must not be null");
 
 		RedisConnectionFactory factory = getConnectionFactory();
-		RedisConnection conn = RedisConnectionUtils.getRedisConnection(factory);
+		RedisConnection conn = RedisConnectionUtils.getConnection(factory);
+
+		boolean pipelineStatus = conn.isPipelined();
+		if (pipeline && !pipelineStatus) {
+			conn.openPipeline();
+		}
 
 		boolean existingConnection = TransactionSynchronizationManager.hasResource(factory);
 
@@ -139,7 +173,13 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 			// TODO: should do flush?
 			return postProcessResult(result, conn, existingConnection);
 		} finally {
-			RedisConnectionUtils.releaseConnection(conn, factory);
+			try {
+				if (pipeline && !pipelineStatus) {
+					conn.closePipeline();
+				}
+			} finally {
+				RedisConnectionUtils.releaseConnection(conn, factory);
+			}
 		}
 	}
 
@@ -153,6 +193,18 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 		return result;
 	}
 
+	@Override
+	public <T> T execute(SessionCallback<T> session) {
+		RedisConnectionFactory factory = getConnectionFactory();
+		// bind connection
+		RedisConnectionUtils.bindConnection(factory);
+		try {
+			return session.execute(this);
+		} finally {
+			RedisConnectionUtils.unbindConnection(factory);
+		}
+	}
+
 	/**
 	 * Returns whether to expose the native Redis connection to RedisCallback code, or rather a connection proxy (the default). 
 	 *

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/SessionCallback.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.core;
+
+/**
+ * Callback executing all operations against a surrogate 'session' (basically against the same underlying Redis connection).
+ * Allows 'transactions' to take place through the use of multi/discard/exec/watch/unwatch commands. 
+ * 
+ * @author Costin Leau
+ */
+public interface SessionCallback<T> {
+
+	/**
+	 * Executes all the given operations inside the same session.
+	 * 
+	 * @param <T> return type
+	 * @param operations Redis operations
+	 * @return return value
+	 */
+	<K, V> T execute(RedisOperations<K, V> operations);
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/support/atomic/CASUtils.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.support.atomic;
+
+import java.util.Collections;
+import java.util.concurrent.Callable;
+
+import org.springframework.data.keyvalue.redis.core.RedisOperations;
+import org.springframework.data.keyvalue.redis.core.SessionCallback;
+
+/**
+ * Check-And-Set (CAS) utility. Performs the CAS loop until successful pattern using
+ * Redis watch/exec operations. 
+ * 
+ * The given callback can contain one or multiple reads followed by a multi call
+ * and one or multiple writes:
+ * 
+ * <pre>
+ * return CASUtils.execute(ops, key, new Callable<Integer>() {
+ *  @Override
+ *  public Integer call() throws Exception {
+ *    // check
+ *    int value = get();
+ *    // start MULTI
+ *    ops.multi();
+ *    // set
+ *    ops.increment(key, 1);
+ *    return value;
+ *  }
+ * });
+ * </pre>
+ * 
+ * @author Costin Leau
+ */
+abstract class CASUtils {
+
+	public static <T, K, V> T execute(final RedisOperations<K, V> ops, final K key, final Callable<T> callback) {
+		return ops.execute(new SessionCallback<T>() {
+			@Override
+			public T execute(RedisOperations operations) {
+				try {
+					for (;;) {
+						operations.watch(Collections.singleton(key));
+						T result = callback.call();
+						if (operations.exec() != null) {
+							return result;
+						}
+					}
+				} catch (Exception ex) {
+					// includes DataAccessException
+					if (ex instanceof RuntimeException) {
+						throw (RuntimeException) ex;
+					}
+					throw new RuntimeException("Callback threw exception", ex);
+				}
+			}
+		});
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/support/atomic/RedisAtomicInteger.java

@@ -17,9 +17,13 @@ package org.springframework.data.keyvalue.redis.support.atomic;
 
 import java.io.Serializable;
 import java.util.Collections;
+import java.util.concurrent.Callable;
 
+import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory;
 import org.springframework.data.keyvalue.redis.core.KeyBound;
 import org.springframework.data.keyvalue.redis.core.RedisOperations;
+import org.springframework.data.keyvalue.redis.core.RedisTemplate;
+import org.springframework.data.keyvalue.redis.core.SessionCallback;
 import org.springframework.data.keyvalue.redis.core.ValueOperations;
 
 /**
@@ -35,6 +39,40 @@ public class RedisAtomicInteger extends Number implements Serializable, KeyBound
 	private ValueOperations<String, Integer> operations;
 	private RedisOperations<String, Integer> generalOps;
 
+
+	/**
+	 * Constructs a new <code>RedisAtomicInteger</code> instance.
+	 *
+	 * @param redisCounter redis counter
+	 * @param factory connection factory
+	 */
+	public RedisAtomicInteger(String redisCounter, RedisConnectionFactory factory) {
+		RedisTemplate<String, Integer> redisTemplate = new RedisTemplate<String, Integer>(factory);
+		redisTemplate.setExposeConnection(true);
+		this.key = redisCounter;
+		this.generalOps = redisTemplate;
+		this.operations = generalOps.opsForValue();
+		if (this.operations.get(redisCounter) == null) {
+			set(0);
+		}
+	}
+
+	/**
+	 * Constructs a new <code>RedisAtomicInteger</code> instance.
+	 *
+	 * @param redisCounter
+	 * @param factory
+	 * @param initialValue
+	 */
+	public RedisAtomicInteger(String redisCounter, RedisConnectionFactory factory, int initialValue) {
+		RedisTemplate<String, Integer> redisTemplate = new RedisTemplate<String, Integer>(factory);
+		redisTemplate.setExposeConnection(true);
+		this.key = redisCounter;
+		this.generalOps = redisTemplate;
+		this.operations = generalOps.opsForValue();
+		this.operations.set(redisCounter, initialValue);
+	}
+
 	/**
 	 * Constructs a new <code>RedisAtomicInteger</code> instance. Uses as initial value
 	 * the data from the backing store (sets the counter to 0 if no value is found).
@@ -109,20 +147,26 @@ public class RedisAtomicInteger extends Number implements Serializable, KeyBound
 	 * @return true if successful. False return indicates that
 	 * the actual value was not equal to the expected value.
 	 */
-	public boolean compareAndSet(int expect, int update) {
-		for (;;) {
-			generalOps.watch(Collections.singleton(key));
-			if (expect == get()) {
-				generalOps.multi();
-				set(update);
-				if (generalOps.exec() != null) {
-					return true;
+	public boolean compareAndSet(final int expect, final int update) {
+		return generalOps.execute(new SessionCallback<Boolean>() {
+
+			@Override
+			public Boolean execute(RedisOperations operations) {
+				for (;;) {
+					operations.watch(Collections.singleton(key));
+					if (expect == get()) {
+						generalOps.multi();
+						set(update);
+						if (operations.exec() != null) {
+							return true;
+						}
+					}
+					{
+						return false;
+					}
 				}
 			}
-			else {
-				return false;
-			}
-		}
+		});
 	}
 
 	/**
@@ -130,15 +174,15 @@ public class RedisAtomicInteger extends Number implements Serializable, KeyBound
 	 * @return the previous value
 	 */
 	public int getAndIncrement() {
-		for (;;) {
-			generalOps.watch(Collections.singleton(key));
-			int value = get();
-			generalOps.multi();
-			operations.increment(key, 1);
-			if (generalOps.exec() != null) {
+		return CASUtils.execute(generalOps, key, new Callable<Integer>() {
+			@Override
+			public Integer call() throws Exception {
+				int value = get();
+				generalOps.multi();
+				operations.increment(key, 1);
 				return value;
 			}
-		}
+		});
 	}
 
 
@@ -147,15 +191,15 @@ public class RedisAtomicInteger extends Number implements Serializable, KeyBound
 	 * @return the previous value
 	 */
 	public int getAndDecrement() {
-		for (;;) {
-			generalOps.watch(Collections.singleton(key));
-			int value = get();
-			generalOps.multi();
-			operations.increment(key, -1);
-			if (generalOps.exec() != null) {
+		return CASUtils.execute(generalOps, key, new Callable<Integer>() {
+			@Override
+			public Integer call() throws Exception {
+				int value = get();
+				generalOps.multi();
+				operations.increment(key, -1);
 				return value;
 			}
-		}
+		});
 	}
 
 
@@ -164,16 +208,16 @@ public class RedisAtomicInteger extends Number implements Serializable, KeyBound
 	 * @param delta the value to add
 	 * @return the previous value
 	 */
-	public int getAndAdd(int delta) {
-		for (;;) {
-			generalOps.watch(Collections.singleton(key));
-			int value = get();
-			generalOps.multi();
-			set(value + delta);
-			if (generalOps.exec() != null) {
+	public int getAndAdd(final int delta) {
+		return CASUtils.execute(generalOps, key, new Callable<Integer>() {
+			@Override
+			public Integer call() throws Exception {
+				int value = get();
+				generalOps.multi();
+				set(value + delta);
 				return value;
 			}
-		}
+		});
 	}
 
 	/**

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/ConnFactoryTracker.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2011 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.keyvalue.redis;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory;
+
+/**
+ * Basic utility to help with the destruction of {@link RedisConnectionFactory} inside JUnit 4 tests.
+ * Simply add the factory during setup and then call {@link #cleanUp()} through the <tt>@AfterClass</tt> method.
+ * 
+ * @author Costin Leau
+ */
+public abstract class ConnFactoryTracker {
+
+	private static Set<RedisConnectionFactory> connFactories = new LinkedHashSet<RedisConnectionFactory>();
+
+	public static void add(RedisConnectionFactory factory) {
+		connFactories.add(factory);
+	}
+
+	public static void cleanUp() {
+		if (connFactories != null) {
+			for (RedisConnectionFactory connectionFactory : connFactories) {
+				try {
+					((DisposableBean) connectionFactory).destroy();
+					System.out.println("Succesfully cleaned up factory " + connectionFactory);
+				} catch (Exception ex) {
+					System.err.println("Cannot clean factory " + connectionFactory + ex);
+				}
+			}
+		}
+	}
+}

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnectionIntegrationTests.java

@@ -25,6 +25,9 @@ import org.springframework.data.keyvalue.redis.connection.Message;
 import org.springframework.data.keyvalue.redis.connection.MessageListener;
 import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory;
 
+import redis.clients.jedis.BinaryJedis;
+import redis.clients.jedis.Transaction;
+
 public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrationTests {
 
 	JedisConnectionFactory factory;
@@ -121,6 +124,22 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati
 		connection.pSubscribe(listener, expectedPattern);
 	}
 
+	@Test
+	public void testMulti() throws Exception {
+		byte[] key = "key".getBytes();
+		byte[] value = "value".getBytes();
+
+		BinaryJedis jedis = (BinaryJedis) connection.getNativeConnection();
+		Transaction multi = jedis.multi();
+		//connection.set(key, value);
+		multi.set(value, key);
+		System.out.println(multi.exec());
+
+		connection.multi();
+		connection.set(value, key);
+		System.out.println(connection.exec());
+	}
+
 //	@Test
 //	public void setAdd() {
 //		connection.sadd("s1", "1");

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/core/SessionTest.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.core;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import org.junit.Test;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.keyvalue.redis.connection.RedisConnection;
+import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory;
+
+/**
+ * @author Costin Leau
+ */
+public class SessionTest {
+
+	@Test
+	public void testSession() throws Exception {
+		final RedisConnection conn = mock(RedisConnection.class);
+		RedisConnectionFactory factory = mock(RedisConnectionFactory.class);
+
+		when(factory.getConnection()).thenReturn(conn);
+		final StringRedisTemplate template = new StringRedisTemplate(factory);
+
+		template.execute(new SessionCallback() {
+			@Override
+			public Object execute(RedisOperations operations) {
+				checkConnection(template, conn);
+				template.discard();
+				assertSame(template, operations);
+				checkConnection(template, conn);
+				return null;
+			}
+		});
+	}
+
+	private void checkConnection(RedisTemplate template, final RedisConnection expectedConnection) {
+		template.execute(new RedisCallback<Object>() {
+
+			@Override
+			public Object doInRedis(RedisConnection connection) throws DataAccessException {
+				assertSame(expectedConnection, connection);
+				return null;
+			}
+		}, true);
+	}
+}

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/support/atomic/AtomicCountersParam.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2010-2011 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.keyvalue.redis.support.atomic;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.springframework.data.keyvalue.redis.SettingsUtils;
+import org.springframework.data.keyvalue.redis.connection.jedis.JedisConnectionFactory;
+
+/**
+ * @author Costin Leau
+ */
+public abstract class AtomicCountersParam {
+
+	public static Collection<Object[]> testParams() {
+		// create Jedis Factory
+		JedisConnectionFactory jedisConnFactory = new JedisConnectionFactory();
+		jedisConnFactory.setPort(SettingsUtils.getPort());
+		jedisConnFactory.setHostName(SettingsUtils.getHost());
+		jedisConnFactory.afterPropertiesSet();
+
+		// jredis factory
+		//		JredisConnectionFactory jredisConnFactory = new JredisConnectionFactory();
+		//		jredisConnFactory.setUsePool(true);
+		//		jredisConnFactory.setPort(SettingsUtils.getPort());
+		//		jredisConnFactory.setHostName(SettingsUtils.getHost());
+		//		jredisConnFactory.afterPropertiesSet();
+
+		return Arrays.asList(new Object[][] { { jedisConnFactory } });
+	}
+}

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/support/atomic/RedisAtomicIntegerTest.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2011 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.keyvalue.redis.support.atomic;
+
+import static org.junit.Assert.*;
+
+import java.util.Collection;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.springframework.data.keyvalue.redis.ConnFactoryTracker;
+import org.springframework.data.keyvalue.redis.connection.RedisConnection;
+import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory;
+
+/**
+ * @author Costin Leau
+ */
+@RunWith(Parameterized.class)
+public class RedisAtomicIntegerTest {
+
+	private RedisAtomicInteger counter;
+	private RedisConnectionFactory factory;
+
+
+	public RedisAtomicIntegerTest(RedisConnectionFactory factory) {
+		counter = new RedisAtomicInteger(getClass().getSimpleName(), factory);
+		this.factory = factory;
+	}
+
+	@After
+	public void stop() {
+		RedisConnection connection = factory.getConnection();
+		connection.flushDb();
+		connection.close();
+	}
+
+	@AfterClass
+	public static void cleanUp() {
+		ConnFactoryTracker.cleanUp();
+	}
+
+	@Parameters
+	public static Collection<Object[]> testParams() {
+		return AtomicCountersParam.testParams();
+	}
+
+	@Test
+	public void testCheckAndSet() throws Exception {
+		counter.set(0);
+		assertFalse(counter.compareAndSet(1, 10));
+		assertTrue(counter.compareAndSet(0, 10));
+		assertTrue(counter.compareAndSet(10, 0));
+	}
+}

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/support/collections/AbstractRedisCollectionTests.java

@@ -24,9 +24,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
-import java.util.LinkedHashSet;
 import java.util.List;
-import java.util.Set;
 
 import org.junit.After;
 import org.junit.AfterClass;
@@ -35,9 +33,8 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
-import org.springframework.beans.factory.DisposableBean;
+import org.springframework.data.keyvalue.redis.ConnFactoryTracker;
 import org.springframework.data.keyvalue.redis.connection.RedisConnection;
-import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory;
 import org.springframework.data.keyvalue.redis.core.RedisCallback;
 import org.springframework.data.keyvalue.redis.core.RedisTemplate;
 
@@ -54,8 +51,6 @@ public abstract class AbstractRedisCollectionTests<T> {
 	protected ObjectFactory<T> factory;
 	protected RedisTemplate template;
 
-	private static Set<RedisConnectionFactory> connFactories = new LinkedHashSet<RedisConnectionFactory>();
-
 	@Before
 	public void setUp() throws Exception {
 		collection = createCollection();
@@ -69,21 +64,12 @@ public abstract class AbstractRedisCollectionTests<T> {
 	public AbstractRedisCollectionTests(ObjectFactory<T> factory, RedisTemplate template) {
 		this.factory = factory;
 		this.template = template;
-		connFactories.add(template.getConnectionFactory());
+		ConnFactoryTracker.add(template.getConnectionFactory());
 	}
 
 	@AfterClass
 	public static void cleanUp() {
-		if (connFactories != null) {
-			for (RedisConnectionFactory connectionFactory : connFactories) {
-				try {
-					((DisposableBean) connectionFactory).destroy();
-					System.out.println("Succesfully cleaned up factory " + connectionFactory);
-				} catch (Exception ex) {
-					System.err.println("Cannot clean factory " + connectionFactory + ex);
-				}
-			}
-		}
+		ConnFactoryTracker.cleanUp();
 	}
 
 	@Parameters

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/support/collections/AbstractRedisMapTests.java

@@ -36,10 +36,9 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
-import org.springframework.beans.factory.DisposableBean;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.keyvalue.redis.ConnFactoryTracker;
 import org.springframework.data.keyvalue.redis.connection.RedisConnection;
-import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory;
 import org.springframework.data.keyvalue.redis.core.RedisCallback;
 import org.springframework.data.keyvalue.redis.core.RedisOperations;
 import org.springframework.data.keyvalue.redis.core.RedisTemplate;
@@ -57,8 +56,6 @@ public abstract class AbstractRedisMapTests<K, V> {
 	protected ObjectFactory<V> valueFactory;
 	protected RedisTemplate template;
 
-	private static Set<RedisConnectionFactory> connFactories = new LinkedHashSet<RedisConnectionFactory>();
-
 	abstract RedisMap<K, V> createMap();
 
 	@Before
@@ -70,21 +67,12 @@ public abstract class AbstractRedisMapTests<K, V> {
 		this.keyFactory = keyFactory;
 		this.valueFactory = valueFactory;
 		this.template = template;
-		connFactories.add(template.getConnectionFactory());
+		ConnFactoryTracker.add(template.getConnectionFactory());
 	}
 
 	@AfterClass
 	public static void cleanUp() {
-		if (connFactories != null) {
-			for (RedisConnectionFactory connectionFactory : connFactories) {
-				try {
-					((DisposableBean) connectionFactory).destroy();
-					System.out.println("Succesfully cleaned up factory " + connectionFactory);
-				} catch (Exception ex) {
-					System.err.println("Cannot clean factory " + connectionFactory + ex);
-				}
-			}
-		}
+		ConnFactoryTracker.cleanUp();
 	}
 
 	protected K getKey() {