Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add RedisTemplate executePipelined methods

Browse Source 
DATAREDIS-119

Add methods to execute RedisCallbacks and
SessionCallbacks in a pipeline and return
deserialized results
This commit is contained in:
Jennifer Hickey
2013-07-25 15:16:12 -07:00
parent f90e373877
commit a5fc567d4a
3 changed files with 200 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
src/main/java/org/springframework/data/redis/core/RedisOperations.java
View File

@@ -64,16 +64,45 @@ public interface RedisOperations<K, V> {
*/
<T> T execute(SessionCallback<T> session);
// /**
// * Executes the given action object on a pipelined connection, returning the results. Note that the callback <b>cannot</b>
// * return a non-null value as it gets overwritten by the pipeline.
// *
// * @param <T> list element return type
// * @param action callback object to execute
// * @return list of objects returned by the pipeline
// */
// List<V> executePipelined(RedisCallback<?> action);
/**
* Executes the given action object on a pipelined connection, returning the results. Note that the callback <b>cannot</b>
* return a non-null value as it gets overwritten by the pipeline.
*
* This method will use the default serializers to deserialize results
*
* @param action callback object to execute
* @return list of objects returned by the pipeline
*/
List<Object> executePipelined(RedisCallback<?> action);
/**
* Executes the given action object on a pipelined connection, returning the results using a dedicated serializer.
* Note that the callback <b>cannot</b> return a non-null value as it gets overwritten by the pipeline.
*
* @param action callback object to execute
* @param resultSerializer The Serializer to use for individual values or Collections of values. If any
* returned values are hashes, this serializer will be used to deserialize both the key and value
* @return list of objects returned by the pipeline
*/
List<Object> executePipelined(final RedisCallback<?> action, final RedisSerializer<?> resultSerializer);
/**
* Executes the given Redis session on a pipelined connection. Allows transactions to be pipelined.
* Note that the callback <b>cannot</b> return a non-null value as it gets overwritten by the pipeline.
* @param session Session callback
* @return list of objects returned by the pipeline
*/
List<Object> executePipelined(final SessionCallback<?> session);
/**
* Executes the given Redis session on a pipelined connection, returning the results using a dedicated serializer.
* Allows transactions to be pipelined.
* Note that the callback <b>cannot</b> return a non-null value as it gets overwritten by the pipeline.
* @param session Session callback
* @param resultSerializer
* @return list of objects returned by the pipeline
*/
List<Object> executePipelined(final SessionCallback<?> session, final RedisSerializer<?> resultSerializer);
Boolean hasKey(K key);

106
src/main/java/org/springframework/data/redis/core/RedisTemplate.java
View File

@@ -27,6 +27,7 @@ import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.RedisConnection;
@@ -201,42 +202,75 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
}
}
// @SuppressWarnings("unchecked")
// public List<V> executePipelined(final RedisCallback<?> action) {
// return executePipelined(action, valueSerializer);
// }
//
// /**
// * Executes the given action object on a pipelined connection, returning the results using a dedicated serializer.
// * Note that the callback <b>cannot</b> return a non-null value as it gets overwritten by the pipeline.
// *
// * @param action callback object to execute
// * @param resultSerializer
// * @return list of objects returned by the pipeline
// */
// public <T> List<T> executePipelined(final RedisCallback<?> action, final RedisSerializer<T> resultSerializer) {
// return execute(new RedisCallback<List<T>>() {
// public List<T> doInRedis(RedisConnection connection) throws DataAccessException {
// connection.openPipeline();
// boolean pipelinedClosed = false;
// try {
// Object result = action.doInRedis(connection);
// if (result != null) {
// throw new InvalidDataAccessApiUsageException(
// "Callback cannot returned a non-null value as it gets overwritten by the pipeline");
// }
// List<Object> closePipeline = connection.closePipeline();
// pipelinedClosed = true;
// //return SerializationUtils.deserialize(pipeline, resultSerializer);
//
// } finally {
// if (!pipelinedClosed) {
// connection.closePipeline();
// }
// }
// }
// });
// }
public List<Object> executePipelined(final SessionCallback<?> session) {
return executePipelined(session, valueSerializer);
}
public List<Object> executePipelined(final SessionCallback<?> session, final RedisSerializer<?> resultSerializer) {
Assert.isTrue(initialized, "template not initialized; call afterPropertiesSet() before using it");
Assert.notNull(session, "Callback object must not be null");
RedisConnectionFactory factory = getConnectionFactory();
// bind connection
RedisConnectionUtils.bindConnection(factory);
try {
return execute(new RedisCallback<List<Object>>() {
public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
connection.openPipeline();
boolean pipelinedClosed = false;
try {
Object result = executeSession(session);
if (result != null) {
throw new InvalidDataAccessApiUsageException(
"Callback cannot return a non-null value as it gets overwritten by the pipeline");
}
List<Object> closePipeline = connection.closePipeline();
pipelinedClosed = true;
return deserializeMixedResults(closePipeline, resultSerializer,
hashKeySerializer, hashValueSerializer);
} finally {
if (!pipelinedClosed) {
connection.closePipeline();
}
}
}
});
} finally {
RedisConnectionUtils.unbindConnection(factory);
}
}
public List<Object> executePipelined(final RedisCallback<?> action) {
return executePipelined(action, valueSerializer);
}
public List<Object> executePipelined(final RedisCallback<?> action, final RedisSerializer<?> resultSerializer) {
return execute(new RedisCallback<List<Object>>() {
public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
connection.openPipeline();
boolean pipelinedClosed = false;
try {
Object result = action.doInRedis(connection);
if (result != null) {
throw new InvalidDataAccessApiUsageException(
"Callback cannot return a non-null value as it gets overwritten by the pipeline");
}
List<Object> closePipeline = connection.closePipeline();
pipelinedClosed = true;
return deserializeMixedResults(closePipeline, resultSerializer,
resultSerializer, resultSerializer);
} finally {
if (!pipelinedClosed) {
connection.closePipeline();
}
}
}
});
}
private Object executeSession(SessionCallback<?> session) {
return session.execute(this);
}
protected RedisConnection createRedisConnectionProxy(RedisConnection pm) {
Class<?>[] ifcs = ClassUtils.getAllInterfacesForClass(pm.getClass(), getClass().getClassLoader());

92
src/test/java/org/springframework/data/redis/core/StringRedisTemplateTests.java
View File

@@ -34,6 +34,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.redis.RedisTestProfileValueSource;
import org.springframework.data.redis.SettingsUtils;
import org.springframework.data.redis.TestCondition;
@@ -215,4 +216,95 @@ public class StringRedisTemplateTests {
// first value is "OK" from set call, results should still be in byte[]
assertEquals("bar", new String((byte[])results.get(1)));
}
@SuppressWarnings("rawtypes")
@Test
public void testExecutePipelined() {
List<Object> results = redisTemplate.executePipelined(new RedisCallback() {
public Object doInRedis(RedisConnection connection) throws DataAccessException {
StringRedisConnection stringRedisConn = (StringRedisConnection) connection;
stringRedisConn.set("foo", "bar");
stringRedisConn.get("foo");
stringRedisConn.rPush("foolist", "a");
stringRedisConn.rPush("foolist", "b");
stringRedisConn.lRange("foolist", 0, -1);
return null;
}
});
assertEquals(Arrays.asList(new Object[] {"bar", 1l, 2l, Arrays.asList(new String[] {"a", "b"})}), results);
}
@SuppressWarnings("rawtypes")
@Test
public void testExecutePipelinedCustomSerializer() {
List<Object> results = redisTemplate.executePipelined(new RedisCallback() {
public Object doInRedis(RedisConnection connection) throws DataAccessException {
StringRedisConnection stringRedisConn = (StringRedisConnection) connection;
stringRedisConn.set("foo", "5");
stringRedisConn.get("foo");
stringRedisConn.rPush("foolist", "10");
stringRedisConn.rPush("foolist", "11");
stringRedisConn.lRange("foolist", 0, -1);
return null;
}
}, new GenericToStringSerializer<Long>(Long.class));
assertEquals(Arrays.asList(new Object[] {5l, 1l, 2l, Arrays.asList(new Long[] {10l, 11l})}), results);
}
@Test(expected=InvalidDataAccessApiUsageException.class)
public void testExecutePipelinedNonNullRedisCallback() {
redisTemplate.executePipelined(new RedisCallback<String>() {
public String doInRedis(RedisConnection connection) throws DataAccessException {
return "Hey There";
}
});
}
@SuppressWarnings({"rawtypes", "unchecked"})
@Test
public void testExecutePipelinedTx() {
List<Object> pipelinedResults = redisTemplate.executePipelined(new SessionCallback() {
public Object execute(RedisOperations operations) throws DataAccessException {
operations.multi();
operations.opsForList().leftPush("foo", "bar");
operations.opsForList().rightPop("foo");
operations.opsForList().size("foo");
operations.exec();
operations.opsForValue().set("foo", "bar");
operations.opsForValue().get("foo");
return null;
}
});
// Should contain the List of deserialized exec results and the result of the last call to get()
assertEquals(Arrays.asList(new Object[] {Arrays.asList(new Object[] {1l, "bar", 0l}), "bar"}), pipelinedResults);
}
@SuppressWarnings({"rawtypes", "unchecked"})
@Test
public void testExecutePipelinedTxCustomSerializer() {
List<Object> pipelinedResults = redisTemplate.executePipelined(new SessionCallback() {
public Object execute(RedisOperations operations) throws DataAccessException {
operations.multi();
operations.opsForList().leftPush("fooList", "5");
operations.opsForList().rightPop("fooList");
operations.opsForList().size("fooList");
operations.exec();
operations.opsForValue().set("foo", "2");
operations.opsForValue().get("foo");
return null;
}
},new GenericToStringSerializer<Long>(Long.class));
// Should contain the List of deserialized exec results and the result of the last call to get()
assertEquals(Arrays.asList(new Object[] {Arrays.asList(new Object[] {1l, 5l, 0l}), 2l}), pipelinedResults);
}
@Test(expected=InvalidDataAccessApiUsageException.class)
public void testExecutePipelinedNonNullSessionCallback() {
redisTemplate.executePipelined(new SessionCallback<String>() {
@SuppressWarnings("rawtypes")
public String execute(RedisOperations operations) throws DataAccessException {
return "Whatup";
}
});
}
}