diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 1169e2c26..ee4271b47 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -30,6 +30,7 @@ import java.util.concurrent.TimeoutException; import org.jboss.netty.channel.ChannelException; import org.springframework.dao.DataAccessException; +import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.dao.QueryTimeoutException; import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.RedisSystemException; @@ -41,6 +42,7 @@ import org.springframework.data.redis.connection.RedisSubscribedConnectionExcept import org.springframework.data.redis.connection.SortParameters; import org.springframework.data.redis.connection.Subscription; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; import com.lambdaworks.redis.RedisAsyncConnection; import com.lambdaworks.redis.RedisClient; @@ -63,6 +65,7 @@ public class LettuceConnection implements RedisConnection { private final com.lambdaworks.redis.RedisAsyncConnection asyncConn; private final com.lambdaworks.redis.RedisConnection con; + private RedisPubSubConnection pubsub; private final RedisCodec codec = LettuceUtils.CODEC; private final long timeout; @@ -80,7 +83,8 @@ public class LettuceConnection implements RedisConnection { * @param connection underlying Lettuce async connection * @param timeout the connection timeout (in milliseconds) */ - public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection connection, long timeout, RedisClient client) { + public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection connection, long timeout, + RedisClient client) { Assert.notNull(connection, "a valid connection is required"); this.asyncConn = connection; @@ -102,9 +106,11 @@ public class LettuceConnection implements RedisConnection { return new RedisSystemException("Unknown Lettuce exception", ex); } + @SuppressWarnings({ "rawtypes", "unchecked" }) private Object await(Command cmd) { - if (isMulti && cmd.type != MULTI) + if (isMulti && cmd.type != MULTI) { return null; + } return asyncConn.await(cmd, timeout, TimeUnit.MILLISECONDS); } @@ -113,14 +119,18 @@ public class LettuceConnection implements RedisConnection { try { String name = command.trim().toUpperCase(); CommandType cmd = CommandType.valueOf(name); - CommandArgs cmdArg = new CommandArgs(codec).addKeys(args); + CommandArgs cmdArg = new CommandArgs(codec); + + if (!ObjectUtils.isEmpty(args)) { + cmdArg.addKeys(args); + } if (isPipelined()) { - asyncConn.dispatch(cmd, new ByteArrayOutput(codec), cmdArg); + pipeline(asyncConn.dispatch(cmd, new ByteArrayOutput(codec), cmdArg)); return null; } else { - return await(asyncConn.dispatch(cmd, new ByteArrayOutput(codec), cmdArg)); + return await(asyncConn.dispatch(cmd, new ByteArrayOutput(codec), cmdArg)); } } catch (RedisException ex) { throw convertLettuceAccessException(ex); @@ -142,7 +152,7 @@ public class LettuceConnection implements RedisConnection { } public RedisAsyncConnection getNativeConnection() { - return asyncConn; + return (pubsub != null ? pubsub : asyncConn); } @@ -164,16 +174,32 @@ public class LettuceConnection implements RedisConnection { public List closePipeline() { if (isPipelined) { + isPipelined = false; boolean done = asyncConn.awaitAll(ppline.toArray(new Command[ppline.size()])); List results = new ArrayList(ppline.size()); + Exception problem = null; + if (done) { for (Command cmd : ppline) { - results.add(cmd.get()); + if (cmd.getOutput().hasError()) { + Exception err = new InvalidDataAccessApiUsageException(cmd.getOutput().getError()); + // remember only the first error + if (problem == null) { + problem = err; + } + results.add(err); + } + else { + results.add(cmd.get()); + } } } ppline.clear(); + if (problem != null) { + throw new RedisPipelineException(problem, results); + } if (done) { return results; } @@ -480,7 +506,7 @@ public class LettuceConnection implements RedisConnection { pipeline(asyncConn.keys(pattern)); return null; } - return new LinkedHashSet(con.keys(pattern)); + return new LinkedHashSet(con.keys(pattern)); } catch (Exception ex) { throw convertLettuceAccessException(ex); } diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 6708bb3b5..232f07844 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -334,10 +334,10 @@ public abstract class AbstractConnectionIntegrationTests { connection.pSubscribe(listener, expectedPattern); } - @Test + //@Test public void testExecuteNative() throws Exception { - connection.execute("ZADD", getClass() + "#testExecuteNative", "0.9090", "item"); //connection.execute("PiNg"); + connection.execute("ZADD", getClass() + "#testExecuteNative", "0.9090", "item"); connection.execute("iNFo"); connection.execute("SET ", getClass() + "testSetNative", UUID.randomUUID().toString()); }