diff --git a/src/main/java/org/springframework/data/redis/core/RedisOperations.java b/src/main/java/org/springframework/data/redis/core/RedisOperations.java index 901b621c7..635f5f34e 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisOperations.java +++ b/src/main/java/org/springframework/data/redis/core/RedisOperations.java @@ -64,16 +64,45 @@ public interface RedisOperations { */ T execute(SessionCallback session); - // /** - // * Executes the given action object on a pipelined connection, returning the results. Note that the callback cannot - // * return a non-null value as it gets overwritten by the pipeline. - // * - // * @param list element return type - // * @param action callback object to execute - // * @return list of objects returned by the pipeline - // */ - // List executePipelined(RedisCallback action); + /** + * Executes the given action object on a pipelined connection, returning the results. Note that the callback cannot + * 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 executePipelined(RedisCallback action); + /** + * Executes the given action object on a pipelined connection, returning the results using a dedicated serializer. + * Note that the callback cannot 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 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 cannot 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 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 cannot 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 executePipelined(final SessionCallback session, final RedisSerializer resultSerializer); Boolean hasKey(K key); diff --git a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java index 43d7d05b0..8c0cf779a 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java @@ -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 extends RedisAccessor implements RedisOperation } } - // @SuppressWarnings("unchecked") - // public List 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 cannot 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 List executePipelined(final RedisCallback action, final RedisSerializer resultSerializer) { - // return execute(new RedisCallback>() { - // public List 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 closePipeline = connection.closePipeline(); - // pipelinedClosed = true; - // //return SerializationUtils.deserialize(pipeline, resultSerializer); - // - // } finally { - // if (!pipelinedClosed) { - // connection.closePipeline(); - // } - // } - // } - // }); - // } + public List executePipelined(final SessionCallback session) { + return executePipelined(session, valueSerializer); + } + + public List 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>() { + public List 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 closePipeline = connection.closePipeline(); + pipelinedClosed = true; + return deserializeMixedResults(closePipeline, resultSerializer, + hashKeySerializer, hashValueSerializer); + } finally { + if (!pipelinedClosed) { + connection.closePipeline(); + } + } + } + }); + } finally { + RedisConnectionUtils.unbindConnection(factory); + } + } + + public List executePipelined(final RedisCallback action) { + return executePipelined(action, valueSerializer); + } + + public List executePipelined(final RedisCallback action, final RedisSerializer resultSerializer) { + return execute(new RedisCallback>() { + public List 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 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()); diff --git a/src/test/java/org/springframework/data/redis/core/StringRedisTemplateTests.java b/src/test/java/org/springframework/data/redis/core/StringRedisTemplateTests.java index 5cdb7c6b2..2e014f833 100644 --- a/src/test/java/org/springframework/data/redis/core/StringRedisTemplateTests.java +++ b/src/test/java/org/springframework/data/redis/core/StringRedisTemplateTests.java @@ -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 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 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.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() { + public String doInRedis(RedisConnection connection) throws DataAccessException { + return "Hey There"; + } + }); + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + @Test + public void testExecutePipelinedTx() { + List 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 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.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() { + @SuppressWarnings("rawtypes") + public String execute(RedisOperations operations) throws DataAccessException { + return "Whatup"; + } + }); + } }