From 64c527c088ac0ee3c6f690782f3d150a54ffef96 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Mon, 25 Jun 2012 19:06:12 +0300 Subject: [PATCH 1/5] handle exceptions of native execution --- .../connection/RedisPipelineException.java | 81 +++++++++++++++++++ .../connection/jedis/JedisConnection.java | 67 +++++++++------ .../connection/jredis/JredisConnection.java | 17 ++-- .../redis/connection/rjc/RjcConnection.java | 13 ++- .../redis/connection/srp/SrpConnection.java | 20 +++-- .../JRedisConnectionIntegrationTests.java | 4 + 6 files changed, 160 insertions(+), 42 deletions(-) create mode 100644 src/main/java/org/springframework/data/redis/connection/RedisPipelineException.java diff --git a/src/main/java/org/springframework/data/redis/connection/RedisPipelineException.java b/src/main/java/org/springframework/data/redis/connection/RedisPipelineException.java new file mode 100644 index 000000000..6f1a514f7 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/RedisPipelineException.java @@ -0,0 +1,81 @@ +/* + * Copyright 2011-2012 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.redis.connection; + +import java.util.Collections; +import java.util.List; + +import org.springframework.dao.InvalidDataAccessResourceUsageException; + +/** + * Exception thrown when executing/closing a pipeline that contains one or multiple invalid/incorrect statements. + * The exception contains the pipeline result, allowing for analysis and tracing. + *

+ * Typically, the first exception returned by the pipeline is used as the cause of this exception for easier + * debugging. + * + * @author Costin Leau + */ +public class RedisPipelineException extends InvalidDataAccessResourceUsageException { + + private final List results; + + /** + * Constructs a new RedisPipelineException instance. + * + * @param msg the message + * @param cause the cause + * @param pipelineResult the pipeline result + */ + public RedisPipelineException(String msg, Throwable cause, List pipelineResult) { + super(msg, cause); + results = Collections.unmodifiableList(pipelineResult); + } + + /** + * Constructs a new RedisPipelineException instance using a default message. + * + * @param cause the cause + * @param pipelineResult the pipeline result + */ + public RedisPipelineException(Exception cause, List pipelineResult) { + this("Pipeline contained one or more invalid commands", cause, pipelineResult); + } + + + /** + * Constructs a new RedisPipelineException instance. + * + * @param msg message + * @param pipelineResult pipeline partial results + */ + public RedisPipelineException(String msg, List pipelineResult) { + super(msg); + results = Collections.unmodifiableList(pipelineResult); + } + + /** + * Returns the result of the pipeline that caused the exception. + * Typically contains both the results of the successful statements but also + * the exceptions of the incorrect ones. + * + * @return result of the pipeline + */ + public List getPipelineResult() { + return results; + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java index 746f6178d..d43ae07de 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java @@ -30,6 +30,7 @@ import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.DataType; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisPipelineException; import org.springframework.data.redis.connection.RedisSubscribedConnectionException; import org.springframework.data.redis.connection.SortParameters; import org.springframework.data.redis.connection.Subscription; @@ -68,8 +69,8 @@ public class JedisConnection implements RedisConnection { static { CLIENT_FIELD = ReflectionUtils.findField(BinaryJedis.class, "client", Client.class); ReflectionUtils.makeAccessible(CLIENT_FIELD); - SEND_COMMAND = ReflectionUtils.findMethod(Connection.class, "sendCommand", - new Class[] { Command.class, byte[][].class }); + SEND_COMMAND = ReflectionUtils.findMethod(Connection.class, "sendCommand", new Class[] { Command.class, + byte[][].class }); ReflectionUtils.makeAccessible(SEND_COMMAND); GET_RESPONSE = ReflectionUtils.findMethod(Queable.class, "getResponse", Builder.class); ReflectionUtils.makeAccessible(GET_RESPONSE); @@ -135,27 +136,33 @@ public class JedisConnection implements RedisConnection { public Object execute(String command, byte[]... args) { Assert.hasText(command, "a valid command needs to be specified"); - List mArgs = new ArrayList(); - if (!ObjectUtils.isEmpty(args)) { - Collections.addAll(mArgs, args); + try { + List mArgs = new ArrayList(); + if (!ObjectUtils.isEmpty(args)) { + Collections.addAll(mArgs, args); + } + + Object result = ReflectionUtils.invokeMethod(SEND_COMMAND, client, + Command.valueOf(command.trim().toUpperCase()), mArgs.toArray(new byte[mArgs.size()][])); + if (isQueueing() || isPipelined()) { + Object target = (isPipelined() ? pipeline : transaction); + ReflectionUtils.invokeMethod(GET_RESPONSE, target, new Builder() { + public Object build(Object data) { + return data; + } + + public String toString() { + return "Object"; + } + }); + } + else { + client.getOne(); + } + return result; + } catch (Exception ex) { + throw convertJedisAccessException(ex); } - - Object result = ReflectionUtils.invokeMethod(SEND_COMMAND, client, - Command.valueOf(command.trim().toUpperCase()), mArgs.toArray(new byte[mArgs.size()][])); - if (isQueueing() || isPipelined()) { - Object target = (isPipelined() ? pipeline : transaction); - ReflectionUtils.invokeMethod(GET_RESPONSE, target, new Builder() { - public Object build(Object data) { - return data; - } - - public String toString() { - return "Object"; - } - }); - } - - return result; } public void close() throws DataAccessException { @@ -232,13 +239,25 @@ public class JedisConnection implements RedisConnection { if (pipeline != null) { List execute = pipeline.syncAndReturnAll(); if (execute != null && !execute.isEmpty()) { - return execute; + Exception cause = null; + for (int i = 0; i < execute.size(); i++) { + Object object = execute.get(i); + if (object instanceof Exception) { + DataAccessException dataAccessException = convertJedisAccessException((Exception) object); + if (cause == null) { + cause = dataAccessException; + } + execute.set(i, dataAccessException); + } + } + if (cause != null) { + throw new RedisPipelineException(cause, execute); + } } } return Collections.emptyList(); } - public List sort(byte[] key, SortParameters params) { SortingParams sortParams = JedisUtils.convertSortParams(params); diff --git a/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java b/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java index f0cdc25d1..d99e7dbe1 100644 --- a/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java @@ -89,14 +89,17 @@ public class JredisConnection implements RedisConnection { public Object execute(String command, byte[]... args) { Assert.hasText(command, "a valid command needs to be specified"); - List mArgs = new ArrayList(); - if (!ObjectUtils.isEmpty(args)) { - Collections.addAll(mArgs, args); + try { + List mArgs = new ArrayList(); + if (!ObjectUtils.isEmpty(args)) { + Collections.addAll(mArgs, args); + } + + return ReflectionUtils.invokeMethod(SERVICE_REQUEST, jredis, Command.valueOf(command.trim().toUpperCase()), + mArgs.toArray(new byte[mArgs.size()][])); + } catch (Exception ex) { + throw convertJredisAccessException(ex); } - - return ReflectionUtils.invokeMethod(SERVICE_REQUEST, jredis, Command.valueOf(command.trim().toUpperCase()), - mArgs.toArray(new byte[mArgs.size()][])); - } public void close() throws RedisSystemException { diff --git a/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java b/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java index 621dc9591..3d203e587 100644 --- a/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java @@ -83,9 +83,16 @@ public class RjcConnection implements RedisConnection { public Object execute(String command, byte[]... args) { Assert.hasText(command, "a valid command needs to be specified"); - connection.sendCommand(Command.valueOf(command.trim().toUpperCase()), - (ObjectUtils.isEmpty(args) ? new byte[0][] : args)); - return connection.getAll(); + try { + connection.sendCommand(Command.valueOf(command.trim().toUpperCase()), + (ObjectUtils.isEmpty(args) ? new byte[0][] : args)); + if (!isPipelined()) { + return connection.getAll(); + } + return null; + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } public void close() throws DataAccessException { diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java index b7bfa2617..b31b2ab13 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java @@ -80,14 +80,18 @@ public class SrpConnection implements RedisConnection { public Object execute(String command, byte[]... args) { Assert.hasText(command, "a valid command needs to be specified"); - String name = command.trim().toUpperCase(); - Command cmd = new Command(name.getBytes(Charsets.UTF_8), args); - if (isPipelined()) { - client.pipeline(name, cmd); - return null; - } - else { - return client.execute(name, cmd); + try { + String name = command.trim().toUpperCase(); + Command cmd = new Command(name.getBytes(Charsets.UTF_8), args); + if (isPipelined()) { + client.pipeline(name, cmd); + return null; + } + else { + return client.execute(name, cmd); + } + } catch (RedisException ex) { + throw convertSRAccessException(ex); } } diff --git a/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java index c38a6134c..8d92a6397 100644 --- a/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java @@ -91,4 +91,8 @@ public class JRedisConnectionIntegrationTests extends AbstractConnectionIntegrat public void testBitSet() throws Exception { } + @Ignore + public void exceptionExecuteNativeWithPipeline() throws Exception { + } + } \ No newline at end of file From 7a89656cf083fb027e0fe67aab71f4fce70d58ae Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Mon, 25 Jun 2012 19:49:20 +0300 Subject: [PATCH 2/5] add pipeline exception for Rjc --- .../redis/connection/RedisConnection.java | 3 ++- .../connection/RedisPipelineException.java | 13 +++++++-- .../redis/connection/rjc/RjcConnection.java | 8 +++--- .../AbstractConnectionIntegrationTests.java | 27 ++++++++++++++----- 4 files changed, 39 insertions(+), 12 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/RedisConnection.java b/src/main/java/org/springframework/data/redis/connection/RedisConnection.java index 5e23602ef..7d69efacf 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisConnection.java @@ -93,7 +93,8 @@ public interface RedisConnection extends RedisCommands { * Executes the commands in the pipeline and returns their result. * If the connection is not pipelined, an empty collection is returned. * + * @throws RedisPipelineException if the pipeline contains any incorrect/invalid statements * @return the result of the executed commands. */ - List closePipeline(); + List closePipeline() throws RedisPipelineException; } \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/RedisPipelineException.java b/src/main/java/org/springframework/data/redis/connection/RedisPipelineException.java index 6f1a514f7..76f10c4d3 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisPipelineException.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisPipelineException.java @@ -23,7 +23,7 @@ import org.springframework.dao.InvalidDataAccessResourceUsageException; /** * Exception thrown when executing/closing a pipeline that contains one or multiple invalid/incorrect statements. - * The exception contains the pipeline result, allowing for analysis and tracing. + * The exception might also contain the pipeline result (if the driver returns it), allowing for analysis and tracing. *

* Typically, the first exception returned by the pipeline is used as the cause of this exception for easier * debugging. @@ -56,6 +56,15 @@ public class RedisPipelineException extends InvalidDataAccessResourceUsageExcept this("Pipeline contained one or more invalid commands", cause, pipelineResult); } + /** + * Constructs a new RedisPipelineException instance using a default message + * and an empty pipeline result list. + * + * @param cause the cause + */ + public RedisPipelineException(Exception cause) { + this("Pipeline contained one or more invalid commands", cause, Collections.emptyList()); + } /** * Constructs a new RedisPipelineException instance. @@ -69,7 +78,7 @@ public class RedisPipelineException extends InvalidDataAccessResourceUsageExcept } /** - * Returns the result of the pipeline that caused the exception. + * Optionally returns the result of the pipeline that caused the exception. * Typically contains both the results of the successful statements but also * the exceptions of the incorrect ones. * diff --git a/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java b/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java index 3d203e587..1170e859a 100644 --- a/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java @@ -34,6 +34,7 @@ import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.DataType; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisPipelineException; import org.springframework.data.redis.connection.RedisSubscribedConnectionException; import org.springframework.data.redis.connection.SortParameters; import org.springframework.data.redis.connection.Subscription; @@ -142,9 +143,10 @@ public class RjcConnection implements RedisConnection { @SuppressWarnings("unchecked") public List closePipeline() { if (pipeline != null) { - List execute = client.getAll(); - if (execute != null && !execute.isEmpty()) { - return execute; + try { + List execute = client.getAll(); + } catch (Exception ex) { + throw new RedisPipelineException(convertRjcAccessException(ex)); } } return Collections.emptyList(); 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 94badf377..4b9bba276 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -209,7 +209,7 @@ public abstract class AbstractConnectionIntegrationTests { final BlockingDeque queue = new LinkedBlockingDeque(); final MessageListener ml = new MessageListener() { - + public void onMessage(Message message, byte[] pattern) { queue.add(message); System.out.println("received message"); @@ -225,7 +225,7 @@ public abstract class AbstractConnectionIntegrationTests { final AtomicBoolean flag = new AtomicBoolean(true); Runnable listener = new Runnable() { - + public void run() { subConn.subscribe(ml, channel); System.out.println("Subscribed"); @@ -264,7 +264,7 @@ public abstract class AbstractConnectionIntegrationTests { MessageListener listener = new MessageListener() { - + public void onMessage(Message message, byte[] pattern) { assertArrayEquals(expectedChannel, message.getChannel()); assertArrayEquals(expectedMessage, message.getBody()); @@ -272,7 +272,7 @@ public abstract class AbstractConnectionIntegrationTests { }; Thread th = new Thread(new Runnable() { - + public void run() { // sleep 1 second to let the registration happen try { @@ -301,7 +301,7 @@ public abstract class AbstractConnectionIntegrationTests { MessageListener listener = new MessageListener() { - + public void onMessage(Message message, byte[] pattern) { assertArrayEquals(expectedPattern, pattern); assertArrayEquals(expectedMessage, message.getBody()); @@ -310,7 +310,7 @@ public abstract class AbstractConnectionIntegrationTests { }; Thread th = new Thread(new Runnable() { - + public void run() { // sleep 1 second to let the registration happen try { @@ -340,4 +340,19 @@ public abstract class AbstractConnectionIntegrationTests { connection.execute("iNFo"); connection.execute("SET ", getClass() + "testSetNative", UUID.randomUUID().toString()); } + + @Test(expected = DataAccessException.class) + public void exceptionExecuteNative() throws Exception { + connection.execute("ZadD", getClass() + "#foo\t0.90\titem"); + } + + @Test(expected = DataAccessException.class) + public void exceptionExecuteNativeWithPipeline() throws Exception { + connection.openPipeline(); + connection.execute("iNFo"); + connection.execute("SET ", getClass() + "testSetNative", UUID.randomUUID().toString()); + connection.execute("ZadD", getClass() + "#foo\t0.90\titem"); + connection.closePipeline(); + } + } \ No newline at end of file From a169f71965bbaac8a8ae163d665c3568736a1a40 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Mon, 25 Jun 2012 20:33:17 +0300 Subject: [PATCH 3/5] add tracking of futures to Srp connection --- .../redis/connection/srp/SrpConnection.java | 266 ++++++++++-------- 1 file changed, 146 insertions(+), 120 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java index b31b2ab13..cd1066fbf 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java @@ -16,6 +16,7 @@ package org.springframework.data.redis.connection.srp; import java.io.IOException; +import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; @@ -42,6 +43,9 @@ import redis.client.RedisException; import redis.reply.Reply; import com.google.common.base.Charsets; +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; /** * {@code RedisConnection} implementation on top of spullara Redis Protocol library. @@ -56,8 +60,36 @@ public class SrpConnection implements RedisConnection { private boolean isClosed = false; private boolean isMulti = false; private Pipeline pipeline; + private PipelineTracker callback; private volatile SrpSubscription subscription; + private static class PipelineTracker implements FutureCallback { + + private List results = Collections.synchronizedList(new ArrayList()); + private List> futures = new ArrayList>(); + + public void onSuccess(Object result) { + results.add(result); + } + + public void onFailure(Throwable t) { + results.add(t); + } + + public List complete() { + try { + return Futures.successfulAsList(futures).get(); + } catch (Exception ex) { + return results; + } + } + + public void addCommand(ListenableFuture future) { + futures.add(future); + Futures.addCallback(future, this); + } + } + public SrpConnection(String host, int port, BlockingQueue queue) { try { this.client = new RedisClient(host, port); @@ -126,24 +158,13 @@ public class SrpConnection implements RedisConnection { public void openPipeline() { if (pipeline == null) { + callback = new PipelineTracker(); pipeline = client.pipeline(); } } public List closePipeline() { - // if (pipeline != null) { - // //ListenableFuture reply = pipeline.exec(); - // pipeline = null; - // if (reply != null) { - // try { - // return SrpUtils.toList(reply.get().data()); - // } catch (Exception ex) { - // throw convertSRAccessException(ex); - // } - // } - // } - throw new UnsupportedOperationException(); - //return Collections.emptyList(); + return callback.complete(); } @@ -153,7 +174,7 @@ public class SrpConnection implements RedisConnection { try { if (isPipelined()) { - pipeline.sort(key, sort, null, (Object[]) null); + pipeline(pipeline.sort(key, sort, null, (Object[]) null)); return null; } return SrpUtils.toBytesList((Reply[]) client.sort(key, sort, null, (Object[]) null).data()); @@ -168,7 +189,7 @@ public class SrpConnection implements RedisConnection { try { if (isPipelined()) { - pipeline.sort(key, sort, null, (Object[]) null); + pipeline(pipeline.sort(key, sort, null, (Object[]) null)); return null; } return ((Long) client.sort(key, sort, null, (Object[]) null).data()); @@ -180,7 +201,7 @@ public class SrpConnection implements RedisConnection { public Long dbSize() { try { if (isPipelined()) { - pipeline.dbsize(); + pipeline(pipeline.dbsize()); return null; } return client.dbsize().data(); @@ -194,7 +215,7 @@ public class SrpConnection implements RedisConnection { public void flushDb() { try { if (isPipelined()) { - pipeline.flushdb(); + pipeline(pipeline.flushdb()); return; } client.flushdb(); @@ -207,7 +228,7 @@ public class SrpConnection implements RedisConnection { public void flushAll() { try { if (isPipelined()) { - pipeline.flushall(); + pipeline(pipeline.flushall()); return; } client.flushall(); @@ -220,7 +241,7 @@ public class SrpConnection implements RedisConnection { public void bgSave() { try { if (isPipelined()) { - pipeline.bgsave(); + pipeline(pipeline.bgsave()); return; } client.bgsave(); @@ -233,7 +254,7 @@ public class SrpConnection implements RedisConnection { public void bgWriteAof() { try { if (isPipelined()) { - pipeline.bgrewriteaof(); + pipeline(pipeline.bgrewriteaof()); return; } client.bgrewriteaof(); @@ -246,7 +267,7 @@ public class SrpConnection implements RedisConnection { public void save() { try { if (isPipelined()) { - pipeline.save(); + pipeline(pipeline.save()); return; } client.save(); @@ -259,7 +280,7 @@ public class SrpConnection implements RedisConnection { public List getConfig(String param) { try { if (isPipelined()) { - pipeline.config_get(param); + pipeline(pipeline.config_get(param)); return null; } return Collections.singletonList(client.config_get(param).toString()); @@ -272,7 +293,7 @@ public class SrpConnection implements RedisConnection { public Properties info() { try { if (isPipelined()) { - pipeline.info(); + pipeline(pipeline.info()); return null; } return SrpUtils.info(client.info()); @@ -285,7 +306,7 @@ public class SrpConnection implements RedisConnection { public Long lastSave() { try { if (isPipelined()) { - pipeline.lastsave(); + pipeline(pipeline.lastsave()); return null; } return client.lastsave().data(); @@ -298,7 +319,7 @@ public class SrpConnection implements RedisConnection { public void setConfig(String param, String value) { try { if (isPipelined()) { - pipeline.config_set(param, value); + pipeline(pipeline.config_set(param, value)); return; } client.config_set(param, value); @@ -312,7 +333,7 @@ public class SrpConnection implements RedisConnection { public void resetConfigStats() { try { if (isPipelined()) { - pipeline.config_resetstat(); + pipeline(pipeline.config_resetstat()); return; } client.config_resetstat(); @@ -326,7 +347,7 @@ public class SrpConnection implements RedisConnection { byte[] save = "SAVE".getBytes(Charsets.UTF_8); try { if (isPipelined()) { - pipeline.shutdown(save, null); + pipeline(pipeline.shutdown(save, null)); return; } client.shutdown(save, null); @@ -339,7 +360,7 @@ public class SrpConnection implements RedisConnection { public byte[] echo(byte[] message) { try { if (isPipelined()) { - pipeline.echo(message); + pipeline(pipeline.echo(message)); return null; } return client.echo(message).data(); @@ -352,7 +373,7 @@ public class SrpConnection implements RedisConnection { public String ping() { try { if (isPipelined()) { - pipeline.ping(); + pipeline(pipeline.ping()); } return client.ping().data(); } catch (Exception ex) { @@ -364,7 +385,7 @@ public class SrpConnection implements RedisConnection { public Long del(byte[]... keys) { try { if (isPipelined()) { - pipeline.del((Object[]) keys); + pipeline(pipeline.del((Object[]) keys)); return null; } return client.del((Object[]) keys).data(); @@ -405,7 +426,7 @@ public class SrpConnection implements RedisConnection { public Boolean exists(byte[] key) { try { if (isPipelined()) { - pipeline.exists(key); + pipeline(pipeline.exists(key)); return null; } return client.exists(key).data() == 1; @@ -418,7 +439,7 @@ public class SrpConnection implements RedisConnection { public Boolean expire(byte[] key, long seconds) { try { if (isPipelined()) { - pipeline.expire(key, seconds); + pipeline(pipeline.expire(key, seconds)); return null; } return client.expire(key, seconds).data() == 1; @@ -431,7 +452,7 @@ public class SrpConnection implements RedisConnection { public Boolean expireAt(byte[] key, long unixTime) { try { if (isPipelined()) { - pipeline.expireat(key, unixTime); + pipeline(pipeline.expireat(key, unixTime)); return null; } return client.expireat(key, unixTime).data() == 1; @@ -444,7 +465,7 @@ public class SrpConnection implements RedisConnection { public Set keys(byte[] pattern) { try { if (isPipelined()) { - pipeline.keys(pattern); + pipeline(pipeline.keys(pattern)); return null; } return SrpUtils.toSet(client.keys(pattern).data()); @@ -475,7 +496,7 @@ public class SrpConnection implements RedisConnection { public Boolean persist(byte[] key) { try { if (isPipelined()) { - pipeline.persist(key); + pipeline(pipeline.persist(key)); return null; } return client.persist(key).data() == 1; @@ -488,7 +509,7 @@ public class SrpConnection implements RedisConnection { public Boolean move(byte[] key, int dbIndex) { try { if (isPipelined()) { - pipeline.move(key, dbIndex); + pipeline(pipeline.move(key, dbIndex)); return null; } return client.move(key, dbIndex).data() == 1; @@ -501,7 +522,7 @@ public class SrpConnection implements RedisConnection { public byte[] randomKey() { try { if (isPipelined()) { - pipeline.randomkey(); + pipeline(pipeline.randomkey()); return null; } return client.randomkey().data(); @@ -514,7 +535,7 @@ public class SrpConnection implements RedisConnection { public void rename(byte[] oldName, byte[] newName) { try { if (isPipelined()) { - pipeline.rename(oldName, newName); + pipeline(pipeline.rename(oldName, newName)); return; } client.rename(oldName, newName); @@ -527,7 +548,7 @@ public class SrpConnection implements RedisConnection { public Boolean renameNX(byte[] oldName, byte[] newName) { try { if (isPipelined()) { - pipeline.renamenx(oldName, newName); + pipeline(pipeline.renamenx(oldName, newName)); return null; } return (client.renamenx(oldName, newName).data() == 1); @@ -552,7 +573,7 @@ public class SrpConnection implements RedisConnection { public Long ttl(byte[] key) { try { if (isPipelined()) { - pipeline.ttl(key); + pipeline(pipeline.ttl(key)); return null; } return client.ttl(key).data(); @@ -565,7 +586,7 @@ public class SrpConnection implements RedisConnection { public DataType type(byte[] key) { try { if (isPipelined()) { - pipeline.type(key); + pipeline(pipeline.type(key)); return null; } return DataType.fromCode(client.type(key).data()); @@ -590,7 +611,7 @@ public class SrpConnection implements RedisConnection { } try { if (isPipelined()) { - pipeline.watch((Object[]) keys); + pipeline(pipeline.watch((Object[]) keys)); } else { client.watch((Object[]) keys); @@ -608,7 +629,7 @@ public class SrpConnection implements RedisConnection { public byte[] get(byte[] key) { try { if (isPipelined()) { - pipeline.get(key); + pipeline(pipeline.get(key)); return null; } @@ -622,7 +643,7 @@ public class SrpConnection implements RedisConnection { public void set(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.set(key, value); + pipeline(pipeline.set(key, value)); return; } client.set(key, value); @@ -636,7 +657,7 @@ public class SrpConnection implements RedisConnection { public byte[] getSet(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.getset(key, value); + pipeline(pipeline.getset(key, value)); return null; } return client.getset(key, value).data(); @@ -649,7 +670,7 @@ public class SrpConnection implements RedisConnection { public Long append(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.append(key, value); + pipeline(pipeline.append(key, value)); return null; } return client.append(key, value).data(); @@ -662,7 +683,7 @@ public class SrpConnection implements RedisConnection { public List mGet(byte[]... keys) { try { if (isPipelined()) { - pipeline.mget((Object[]) keys); + pipeline(pipeline.mget((Object[]) keys)); return null; } return SrpUtils.toBytesList(client.mget((Object[]) keys).data()); @@ -675,7 +696,7 @@ public class SrpConnection implements RedisConnection { public void mSet(Map tuples) { try { if (isPipelined()) { - pipeline.mset((Object[]) SrpUtils.convert(tuples)); + pipeline(pipeline.mset((Object[]) SrpUtils.convert(tuples))); return; } client.mset((Object[]) SrpUtils.convert(tuples)); @@ -688,7 +709,7 @@ public class SrpConnection implements RedisConnection { public void mSetNX(Map tuples) { try { if (isPipelined()) { - pipeline.msetnx((Object[]) SrpUtils.convert(tuples)); + pipeline(pipeline.msetnx((Object[]) SrpUtils.convert(tuples))); return; } client.msetnx((Object[]) SrpUtils.convert(tuples)); @@ -701,7 +722,7 @@ public class SrpConnection implements RedisConnection { public void setEx(byte[] key, long time, byte[] value) { try { if (isPipelined()) { - pipeline.setex(key, time, value); + pipeline(pipeline.setex(key, time, value)); return; } client.setex(key, time, value); @@ -714,7 +735,7 @@ public class SrpConnection implements RedisConnection { public Boolean setNX(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.setnx(key, value); + pipeline(pipeline.setnx(key, value)); return null; } return client.setnx(key, value).data() == 1; @@ -727,7 +748,7 @@ public class SrpConnection implements RedisConnection { public byte[] getRange(byte[] key, long start, long end) { try { if (isPipelined()) { - pipeline.getrange(key, start, end); + pipeline(pipeline.getrange(key, start, end)); return null; } return client.getrange(key, start, end).data(); @@ -740,7 +761,7 @@ public class SrpConnection implements RedisConnection { public Long decr(byte[] key) { try { if (isPipelined()) { - pipeline.decr(key); + pipeline(pipeline.decr(key)); return null; } return client.decr(key).data(); @@ -753,7 +774,7 @@ public class SrpConnection implements RedisConnection { public Long decrBy(byte[] key, long value) { try { if (isPipelined()) { - pipeline.decrby(key, value); + pipeline(pipeline.decrby(key, value)); return null; } return client.decrby(key, value).data(); @@ -766,7 +787,7 @@ public class SrpConnection implements RedisConnection { public Long incr(byte[] key) { try { if (isPipelined()) { - pipeline.incr(key); + pipeline(pipeline.incr(key)); return null; } return client.incr(key).data(); @@ -779,7 +800,7 @@ public class SrpConnection implements RedisConnection { public Long incrBy(byte[] key, long value) { try { if (isPipelined()) { - pipeline.incrby(key, value); + pipeline(pipeline.incrby(key, value)); return null; } return client.incrby(key, value).data(); @@ -837,7 +858,7 @@ public class SrpConnection implements RedisConnection { public Long strLen(byte[] key) { try { if (isPipelined()) { - pipeline.strlen(key); + pipeline(pipeline.strlen(key)); return null; } return client.strlen(key).data(); @@ -854,7 +875,7 @@ public class SrpConnection implements RedisConnection { public Long lPush(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.lpush(key, new Object[] { value }); + pipeline(pipeline.lpush(key, new Object[] { value })); return null; } return client.lpush(key, new Object[] { value }).data(); @@ -867,7 +888,7 @@ public class SrpConnection implements RedisConnection { public Long rPush(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.rpush(key, new Object[] { value }); + pipeline(pipeline.rpush(key, new Object[] { value })); return null; } return client.rpush(key, new Object[] { value }).data(); @@ -880,7 +901,7 @@ public class SrpConnection implements RedisConnection { public List bLPop(int timeout, byte[]... keys) { try { if (isPipelined()) { - // pipeline.blpop(timeout, keys); + // pipeline(pipeline.blpop(timeout, keys)); return null; } // return SrpUtils.toBytesList(client.blpop(timeout, keys).data()); @@ -894,7 +915,7 @@ public class SrpConnection implements RedisConnection { public List bRPop(int timeout, byte[]... keys) { try { if (isPipelined()) { - // pipeline.brpop(timeout, keys); + // pipeline(pipeline.brpop(timeout, keys)); return null; } // return SrpUtils.toBytesList(client.brpop(timeout, keys).data()); @@ -908,7 +929,7 @@ public class SrpConnection implements RedisConnection { public byte[] lIndex(byte[] key, long index) { try { if (isPipelined()) { - pipeline.lindex(key, index); + pipeline(pipeline.lindex(key, index)); return null; } return client.lindex(key, index).data(); @@ -921,7 +942,7 @@ public class SrpConnection implements RedisConnection { public Long lInsert(byte[] key, Position where, byte[] pivot, byte[] value) { try { if (isPipelined()) { - pipeline.linsert(key, SrpUtils.convertPosition(where), pivot, value); + pipeline(pipeline.linsert(key, SrpUtils.convertPosition(where), pivot, value)); return null; } return client.linsert(key, SrpUtils.convertPosition(where), pivot, value).data(); @@ -934,7 +955,7 @@ public class SrpConnection implements RedisConnection { public Long lLen(byte[] key) { try { if (isPipelined()) { - pipeline.llen(key); + pipeline(pipeline.llen(key)); return null; } return client.llen(key).data(); @@ -947,7 +968,7 @@ public class SrpConnection implements RedisConnection { public byte[] lPop(byte[] key) { try { if (isPipelined()) { - pipeline.lpop(key); + pipeline(pipeline.lpop(key)); return null; } return client.lpop(key).data(); @@ -960,7 +981,7 @@ public class SrpConnection implements RedisConnection { public List lRange(byte[] key, long start, long end) { try { if (isPipelined()) { - pipeline.lrange(key, start, end); + pipeline(pipeline.lrange(key, start, end)); return null; } return SrpUtils.toBytesList(client.lrange(key, start, end).data()); @@ -973,7 +994,7 @@ public class SrpConnection implements RedisConnection { public Long lRem(byte[] key, long count, byte[] value) { try { if (isPipelined()) { - pipeline.lrem(key, count, value); + pipeline(pipeline.lrem(key, count, value)); return null; } return client.lrem(key, count, value).data(); @@ -986,7 +1007,7 @@ public class SrpConnection implements RedisConnection { public void lSet(byte[] key, long index, byte[] value) { try { if (isPipelined()) { - pipeline.lset(key, index, value); + pipeline(pipeline.lset(key, index, value)); return; } client.lset(key, index, value); @@ -999,7 +1020,7 @@ public class SrpConnection implements RedisConnection { public void lTrim(byte[] key, long start, long end) { try { if (isPipelined()) { - pipeline.ltrim(key, start, end); + pipeline(pipeline.ltrim(key, start, end)); return; } client.ltrim(key, start, end); @@ -1012,7 +1033,7 @@ public class SrpConnection implements RedisConnection { public byte[] rPop(byte[] key) { try { if (isPipelined()) { - pipeline.rpop(key); + pipeline(pipeline.rpop(key)); return null; } return client.rpop(key).data(); @@ -1025,7 +1046,7 @@ public class SrpConnection implements RedisConnection { public byte[] rPopLPush(byte[] srcKey, byte[] dstKey) { try { if (isPipelined()) { - pipeline.rpoplpush(srcKey, dstKey); + pipeline(pipeline.rpoplpush(srcKey, dstKey)); return null; } return client.rpoplpush(srcKey, dstKey).data(); @@ -1038,7 +1059,7 @@ public class SrpConnection implements RedisConnection { public byte[] bRPopLPush(int timeout, byte[] srcKey, byte[] dstKey) { try { if (isPipelined()) { - pipeline.brpoplpush(srcKey, dstKey, timeout); + pipeline(pipeline.brpoplpush(srcKey, dstKey, timeout)); return null; } return client.brpoplpush(srcKey, dstKey, timeout).data(); @@ -1051,7 +1072,7 @@ public class SrpConnection implements RedisConnection { public Long lPushX(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.lpushx(key, value); + pipeline(pipeline.lpushx(key, value)); return null; } return client.lpushx(key, value).data(); @@ -1064,7 +1085,7 @@ public class SrpConnection implements RedisConnection { public Long rPushX(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.rpushx(key, value); + pipeline(pipeline.rpushx(key, value)); return null; } return client.rpushx(key, value).data(); @@ -1082,7 +1103,7 @@ public class SrpConnection implements RedisConnection { public Boolean sAdd(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.sadd(key, new Object[] { value }); + pipeline(pipeline.sadd(key, new Object[] { value })); return null; } return (client.sadd(key, new Object[] { value }).data() == 1); @@ -1095,7 +1116,7 @@ public class SrpConnection implements RedisConnection { public Long sCard(byte[] key) { try { if (isPipelined()) { - pipeline.scard(key); + pipeline(pipeline.scard(key)); return null; } return client.scard(key).data(); @@ -1108,7 +1129,7 @@ public class SrpConnection implements RedisConnection { public Set sDiff(byte[]... keys) { try { if (isPipelined()) { - pipeline.sdiff((Object[]) keys); + pipeline(pipeline.sdiff((Object[]) keys)); return null; } return SrpUtils.toSet(client.sdiff((Object[]) keys).data()); @@ -1121,7 +1142,7 @@ public class SrpConnection implements RedisConnection { public Long sDiffStore(byte[] destKey, byte[]... keys) { try { if (isPipelined()) { - pipeline.sdiffstore(destKey, (Object[]) keys); + pipeline(pipeline.sdiffstore(destKey, (Object[]) keys)); return null; } return client.sdiffstore(destKey, (Object[]) keys).data(); @@ -1134,7 +1155,7 @@ public class SrpConnection implements RedisConnection { public Set sInter(byte[]... keys) { try { if (isPipelined()) { - pipeline.sinter((Object[]) keys); + pipeline(pipeline.sinter((Object[]) keys)); return null; } return SrpUtils.toSet(client.sinter((Object[]) keys).data()); @@ -1147,7 +1168,7 @@ public class SrpConnection implements RedisConnection { public Long sInterStore(byte[] destKey, byte[]... keys) { try { if (isPipelined()) { - pipeline.sinterstore(destKey, (Object[]) keys); + pipeline(pipeline.sinterstore(destKey, (Object[]) keys)); return null; } return client.sinterstore(destKey, (Object[]) keys).data(); @@ -1160,7 +1181,7 @@ public class SrpConnection implements RedisConnection { public Boolean sIsMember(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.sismember(key, value); + pipeline(pipeline.sismember(key, value)); return null; } return client.sismember(key, value).data() == 1; @@ -1173,7 +1194,7 @@ public class SrpConnection implements RedisConnection { public Set sMembers(byte[] key) { try { if (isPipelined()) { - pipeline.smembers(key); + pipeline(pipeline.smembers(key)); return null; } return SrpUtils.toSet(client.smembers(key).data()); @@ -1186,7 +1207,7 @@ public class SrpConnection implements RedisConnection { public Boolean sMove(byte[] srcKey, byte[] destKey, byte[] value) { try { if (isPipelined()) { - pipeline.smove(srcKey, destKey, value); + pipeline(pipeline.smove(srcKey, destKey, value)); return null; } return client.smove(srcKey, destKey, value).data() == 1; @@ -1199,7 +1220,7 @@ public class SrpConnection implements RedisConnection { public byte[] sPop(byte[] key) { try { if (isPipelined()) { - pipeline.spop(key); + pipeline(pipeline.spop(key)); return null; } return client.spop(key).data(); @@ -1212,7 +1233,7 @@ public class SrpConnection implements RedisConnection { public byte[] sRandMember(byte[] key) { try { if (isPipelined()) { - pipeline.srandmember(key); + pipeline(pipeline.srandmember(key)); return null; } return client.srandmember(key).data(); @@ -1225,7 +1246,7 @@ public class SrpConnection implements RedisConnection { public Boolean sRem(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.srem(key, new Object[] { value }); + pipeline(pipeline.srem(key, new Object[] { value })); return null; } return client.srem(key, new Object[] { value }).data() == 1; @@ -1238,7 +1259,7 @@ public class SrpConnection implements RedisConnection { public Set sUnion(byte[]... keys) { try { if (isPipelined()) { - pipeline.sunion((Object[]) keys); + pipeline(pipeline.sunion((Object[]) keys)); return null; } return SrpUtils.toSet(client.sunion((Object[]) keys).data()); @@ -1251,7 +1272,7 @@ public class SrpConnection implements RedisConnection { public Long sUnionStore(byte[] destKey, byte[]... keys) { try { if (isPipelined()) { - pipeline.sunionstore(destKey, (Object[]) keys); + pipeline(pipeline.sunionstore(destKey, (Object[]) keys)); return null; } return client.sunionstore(destKey, (Object[]) keys).data(); @@ -1268,7 +1289,7 @@ public class SrpConnection implements RedisConnection { public Boolean zAdd(byte[] key, double score, byte[] value) { try { if (isPipelined()) { - pipeline.zadd(new Object[] { key, score, value }); + pipeline(pipeline.zadd(new Object[] { key, score, value })); return null; } return client.zadd(new Object[] { key, score, value }).data() == 1; @@ -1281,7 +1302,7 @@ public class SrpConnection implements RedisConnection { public Long zCard(byte[] key) { try { if (isPipelined()) { - pipeline.zcard(key); + pipeline(pipeline.zcard(key)); return null; } return client.zcard(key).data(); @@ -1294,7 +1315,7 @@ public class SrpConnection implements RedisConnection { public Long zCount(byte[] key, double min, double max) { try { if (isQueueing()) { - pipeline.zcount(key, min, max); + pipeline(pipeline.zcount(key, min, max)); return null; } return client.zcount(key, min, max).data(); @@ -1307,7 +1328,7 @@ public class SrpConnection implements RedisConnection { public Double zIncrBy(byte[] key, double increment, byte[] value) { try { if (isPipelined()) { - pipeline.zincrby(key, increment, value); + pipeline(pipeline.zincrby(key, increment, value)); return null; } return SrpUtils.toDouble(client.zincrby(key, increment, value).data()); @@ -1335,7 +1356,7 @@ public class SrpConnection implements RedisConnection { try { if (isQueueing()) { - pipeline.zinterstore(args); + pipeline(pipeline.zinterstore(args)); return null; } return client.zinterstore(args).data(); @@ -1347,7 +1368,7 @@ public class SrpConnection implements RedisConnection { public Set zRange(byte[] key, long start, long end) { try { if (isPipelined()) { - pipeline.zrange(key, start, end, null); + pipeline(pipeline.zrange(key, start, end, null)); return null; } return SrpUtils.toSet(client.zrange(key, start, end, null).data()); @@ -1360,7 +1381,7 @@ public class SrpConnection implements RedisConnection { public Set zRangeWithScores(byte[] key, long start, long end) { try { if (isPipelined()) { - pipeline.zrange(key, start, end, SrpUtils.WITHSCORES); + pipeline(pipeline.zrange(key, start, end, SrpUtils.WITHSCORES)); return null; } return SrpUtils.convertTuple(client.zrange(key, start, end, SrpUtils.WITHSCORES)); @@ -1373,7 +1394,7 @@ public class SrpConnection implements RedisConnection { public Set zRangeByScore(byte[] key, double min, double max) { try { if (isPipelined()) { - pipeline.zrangebyscore(key, min, max, null, null); + pipeline(pipeline.zrangebyscore(key, min, max, null, null)); return null; } return SrpUtils.toSet(client.zrangebyscore(key, min, max, null, null).data()); @@ -1386,7 +1407,7 @@ public class SrpConnection implements RedisConnection { public Set zRangeByScoreWithScores(byte[] key, double min, double max) { try { if (isPipelined()) { - pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, null); + pipeline(pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, null)); return null; } return SrpUtils.convertTuple(client.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, null)); @@ -1399,7 +1420,7 @@ public class SrpConnection implements RedisConnection { public Set zRevRangeWithScores(byte[] key, long start, long end) { try { if (isPipelined()) { - pipeline.zrevrange(key, start, end, SrpUtils.WITHSCORES); + pipeline(pipeline.zrevrange(key, start, end, SrpUtils.WITHSCORES)); return null; } return SrpUtils.convertTuple(client.zrevrange(key, start, end, SrpUtils.WITHSCORES)); @@ -1413,7 +1434,7 @@ public class SrpConnection implements RedisConnection { try { byte[] limit = SrpUtils.limit(offset, count); if (isPipelined()) { - pipeline.zrangebyscore(key, min, max, null, limit); + pipeline(pipeline.zrangebyscore(key, min, max, null, limit)); return null; } return SrpUtils.toSet(client.zrangebyscore(key, min, max, null, limit).data()); @@ -1427,7 +1448,7 @@ public class SrpConnection implements RedisConnection { try { byte[] limit = SrpUtils.limit(offset, count); if (isPipelined()) { - pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit); + pipeline(pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit)); return null; } return SrpUtils.convertTuple(client.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit)); @@ -1504,7 +1525,7 @@ public class SrpConnection implements RedisConnection { public Boolean zRem(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.zrem(key, new Object[] { value }); + pipeline(pipeline.zrem(key, new Object[] { value })); return null; } return client.zrem(key, new Object[] { value }).data() == 1; @@ -1517,7 +1538,7 @@ public class SrpConnection implements RedisConnection { public Long zRemRange(byte[] key, long start, long end) { try { if (isPipelined()) { - pipeline.zremrangebyrank(key, start, end); + pipeline(pipeline.zremrangebyrank(key, start, end)); return null; } return client.zremrangebyrank(key, start, end).data(); @@ -1530,7 +1551,7 @@ public class SrpConnection implements RedisConnection { public Long zRemRangeByScore(byte[] key, double min, double max) { try { if (isPipelined()) { - pipeline.zremrangebyscore(key, min, max); + pipeline(pipeline.zremrangebyscore(key, min, max)); return null; } return client.zremrangebyscore(key, min, max).data(); @@ -1543,7 +1564,7 @@ public class SrpConnection implements RedisConnection { public Set zRevRange(byte[] key, long start, long end) { try { if (isPipelined()) { - pipeline.zrevrange(key, start, end, null); + pipeline(pipeline.zrevrange(key, start, end, null)); return null; } return SrpUtils.toSet(client.zrevrange(key, start, end, null).data()); @@ -1556,7 +1577,7 @@ public class SrpConnection implements RedisConnection { public Long zRevRank(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.zrevrank(key, value); + pipeline(pipeline.zrevrank(key, value)); return null; } return (Long) client.zrevrank(key, value).data(); @@ -1569,7 +1590,7 @@ public class SrpConnection implements RedisConnection { public Double zScore(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.zscore(key, value); + pipeline(pipeline.zscore(key, value)); return null; } return SrpUtils.toDouble(client.zscore(key, value).data()); @@ -1587,7 +1608,7 @@ public class SrpConnection implements RedisConnection { public Long zUnionStore(byte[] destKey, byte[]... sets) { try { if (isPipelined()) { - pipeline.zunionstore(destKey, sets.length, (Object[]) sets); + pipeline(pipeline.zunionstore(destKey, sets.length, (Object[]) sets)); return null; } return client.zunionstore(destKey, sets.length, (Object[]) sets).data(); @@ -1604,7 +1625,7 @@ public class SrpConnection implements RedisConnection { public Boolean hSet(byte[] key, byte[] field, byte[] value) { try { if (isPipelined()) { - pipeline.hset(key, field, value); + pipeline(pipeline.hset(key, field, value)); return null; } return client.hset(key, field, value).data() == 1; @@ -1617,7 +1638,7 @@ public class SrpConnection implements RedisConnection { public Boolean hSetNX(byte[] key, byte[] field, byte[] value) { try { if (isPipelined()) { - pipeline.hsetnx(key, field, value); + pipeline(pipeline.hsetnx(key, field, value)); return null; } return client.hsetnx(key, field, value).data() == 1; @@ -1630,7 +1651,7 @@ public class SrpConnection implements RedisConnection { public Boolean hDel(byte[] key, byte[] field) { try { if (isPipelined()) { - pipeline.hdel(key, new Object[] { field }); + pipeline(pipeline.hdel(key, new Object[] { field })); return null; } return client.hdel(key, new Object[] { field }).data() == 1; @@ -1643,7 +1664,7 @@ public class SrpConnection implements RedisConnection { public Boolean hExists(byte[] key, byte[] field) { try { if (isPipelined()) { - pipeline.hexists(key, field); + pipeline(pipeline.hexists(key, field)); return null; } return client.hexists(key, field).data() == 1; @@ -1656,7 +1677,7 @@ public class SrpConnection implements RedisConnection { public byte[] hGet(byte[] key, byte[] field) { try { if (isPipelined()) { - pipeline.hget(key, field); + pipeline(pipeline.hget(key, field)); return null; } return client.hget(key, field).data(); @@ -1669,7 +1690,7 @@ public class SrpConnection implements RedisConnection { public Map hGetAll(byte[] key) { try { if (isPipelined()) { - pipeline.hgetall(key); + pipeline(pipeline.hgetall(key)); return null; } return SrpUtils.toMap(client.hgetall(key).data()); @@ -1682,7 +1703,7 @@ public class SrpConnection implements RedisConnection { public Long hIncrBy(byte[] key, byte[] field, long delta) { try { if (isPipelined()) { - pipeline.hincrby(key, field, delta); + pipeline(pipeline.hincrby(key, field, delta)); return null; } return client.hincrby(key, field, delta).data(); @@ -1695,7 +1716,7 @@ public class SrpConnection implements RedisConnection { public Set hKeys(byte[] key) { try { if (isPipelined()) { - pipeline.hkeys(key); + pipeline(pipeline.hkeys(key)); return null; } return SrpUtils.toSet(client.hkeys(key).data()); @@ -1708,7 +1729,7 @@ public class SrpConnection implements RedisConnection { public Long hLen(byte[] key) { try { if (isPipelined()) { - pipeline.hlen(key); + pipeline(pipeline.hlen(key)); return null; } return client.hlen(key).data(); @@ -1721,7 +1742,7 @@ public class SrpConnection implements RedisConnection { public List hMGet(byte[] key, byte[]... fields) { try { if (isPipelined()) { - pipeline.hmget(key, (Object[]) fields); + pipeline(pipeline.hmget(key, (Object[]) fields)); return null; } return SrpUtils.toBytesList(client.hmget(key, (Object[]) fields).data()); @@ -1734,7 +1755,7 @@ public class SrpConnection implements RedisConnection { public void hMSet(byte[] key, Map tuple) { try { if (isPipelined()) { - pipeline.hmset(key, SrpUtils.convert(tuple)); + pipeline(pipeline.hmset(key, SrpUtils.convert(tuple))); return; } client.hmset(key, SrpUtils.convert(tuple)); @@ -1747,7 +1768,7 @@ public class SrpConnection implements RedisConnection { public List hVals(byte[] key) { try { if (isPipelined()) { - pipeline.hvals(key); + pipeline(pipeline.hvals(key)); return null; } return SrpUtils.toBytesList(client.hvals(key).data()); @@ -1767,7 +1788,7 @@ public class SrpConnection implements RedisConnection { throw new UnsupportedOperationException(); } if (isPipelined()) { - pipeline.publish(channel, message); + pipeline(pipeline.publish(channel, message)); return null; } return client.publish(channel, message).data(); @@ -1828,4 +1849,9 @@ public class SrpConnection implements RedisConnection { "Connection already subscribed; use the connection Subscription to cancel or add new channels"); } } + + // processing method that adds a listener to the future in order to track down the results and close the pipeline + private void pipeline(ListenableFuture future) { + callback.addCommand(future); + } } \ No newline at end of file From d46b23fe1b97e791cd9b9916cfaf9119bab43dda Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Mon, 25 Jun 2012 20:46:06 +0300 Subject: [PATCH 4/5] finish closePipeline for Srp --- .../redis/connection/srp/SrpConnection.java | 295 ++++++++++-------- .../AbstractConnectionIntegrationTests.java | 2 +- 2 files changed, 164 insertions(+), 133 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java index cd1066fbf..9eeaa2ee1 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java @@ -31,6 +31,7 @@ import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.DataType; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisPipelineException; import org.springframework.data.redis.connection.RedisSubscribedConnectionException; import org.springframework.data.redis.connection.SortParameters; import org.springframework.data.redis.connection.Subscription; @@ -65,8 +66,8 @@ public class SrpConnection implements RedisConnection { private static class PipelineTracker implements FutureCallback { - private List results = Collections.synchronizedList(new ArrayList()); - private List> futures = new ArrayList>(); + private final List results = Collections.synchronizedList(new ArrayList()); + private final List> futures = new ArrayList>(); public void onSuccess(Object result) { results.add(result); @@ -78,16 +79,23 @@ public class SrpConnection implements RedisConnection { public List complete() { try { - return Futures.successfulAsList(futures).get(); + Futures.successfulAsList(futures).get(); } catch (Exception ex) { - return results; + // ignore } + + return results; } public void addCommand(ListenableFuture future) { futures.add(future); Futures.addCallback(future, this); } + + public void close() { + results.clear(); + futures.clear(); + } } public SrpConnection(String host, int port, BlockingQueue queue) { @@ -99,7 +107,7 @@ public class SrpConnection implements RedisConnection { } } - protected DataAccessException convertSRAccessException(Exception ex) { + protected DataAccessException convertSrpAccessException(Exception ex) { if (ex instanceof RedisException) { return SrpUtils.convertSRedisAccessException((RedisException) ex); } @@ -107,7 +115,7 @@ public class SrpConnection implements RedisConnection { return new RedisConnectionFailureException("Redis connection failed", (IOException) ex); } - return new RedisSystemException("Unknown SRedis exception", ex); + return new RedisSystemException("Unknown SRP exception", ex); } public Object execute(String command, byte[]... args) { @@ -116,14 +124,14 @@ public class SrpConnection implements RedisConnection { String name = command.trim().toUpperCase(); Command cmd = new Command(name.getBytes(Charsets.UTF_8), args); if (isPipelined()) { - client.pipeline(name, cmd); + pipeline(client.pipeline(name, cmd)); return null; } else { return client.execute(name, cmd); } } catch (RedisException ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -134,7 +142,7 @@ public class SrpConnection implements RedisConnection { try { client.close(); } catch (IOException ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -164,7 +172,30 @@ public class SrpConnection implements RedisConnection { } public List closePipeline() { - return callback.complete(); + if (pipeline != null) { + List execute = new ArrayList(callback.complete()); + callback.close(); + callback = null; + if (execute != null && !execute.isEmpty()) { + Exception cause = null; + for (int i = 0; i < execute.size(); i++) { + Object object = execute.get(i); + if (object instanceof Exception) { + DataAccessException dataAccessException = convertSrpAccessException((Exception) object); + if (cause == null) { + cause = dataAccessException; + } + execute.set(i, dataAccessException); + } + } + if (cause != null) { + throw new RedisPipelineException(cause, execute); + } + return execute; + } + } + + return Collections.emptyList(); } @@ -179,7 +210,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toBytesList((Reply[]) client.sort(key, sort, null, (Object[]) null).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -194,7 +225,7 @@ public class SrpConnection implements RedisConnection { } return ((Long) client.sort(key, sort, null, (Object[]) null).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -206,7 +237,7 @@ public class SrpConnection implements RedisConnection { } return client.dbsize().data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -220,7 +251,7 @@ public class SrpConnection implements RedisConnection { } client.flushdb(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -233,7 +264,7 @@ public class SrpConnection implements RedisConnection { } client.flushall(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -246,7 +277,7 @@ public class SrpConnection implements RedisConnection { } client.bgsave(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -259,7 +290,7 @@ public class SrpConnection implements RedisConnection { } client.bgrewriteaof(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -272,7 +303,7 @@ public class SrpConnection implements RedisConnection { } client.save(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -285,7 +316,7 @@ public class SrpConnection implements RedisConnection { } return Collections.singletonList(client.config_get(param).toString()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -298,7 +329,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.info(client.info()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -311,7 +342,7 @@ public class SrpConnection implements RedisConnection { } return client.lastsave().data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -324,7 +355,7 @@ public class SrpConnection implements RedisConnection { } client.config_set(param, value); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -338,7 +369,7 @@ public class SrpConnection implements RedisConnection { } client.config_resetstat(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -352,7 +383,7 @@ public class SrpConnection implements RedisConnection { } client.shutdown(save, null); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -365,7 +396,7 @@ public class SrpConnection implements RedisConnection { } return client.echo(message).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -377,7 +408,7 @@ public class SrpConnection implements RedisConnection { } return client.ping().data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -390,7 +421,7 @@ public class SrpConnection implements RedisConnection { } return client.del((Object[]) keys).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -404,7 +435,7 @@ public class SrpConnection implements RedisConnection { client.discard(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -418,7 +449,7 @@ public class SrpConnection implements RedisConnection { } return Collections.singletonList((Object) exec); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -431,7 +462,7 @@ public class SrpConnection implements RedisConnection { } return client.exists(key).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -444,7 +475,7 @@ public class SrpConnection implements RedisConnection { } return client.expire(key, seconds).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -457,7 +488,7 @@ public class SrpConnection implements RedisConnection { } return client.expireat(key, unixTime).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -470,7 +501,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.keys(pattern).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -488,7 +519,7 @@ public class SrpConnection implements RedisConnection { } client.multi(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -501,7 +532,7 @@ public class SrpConnection implements RedisConnection { } return client.persist(key).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -514,7 +545,7 @@ public class SrpConnection implements RedisConnection { } return client.move(key, dbIndex).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -527,7 +558,7 @@ public class SrpConnection implements RedisConnection { } return client.randomkey().data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -540,7 +571,7 @@ public class SrpConnection implements RedisConnection { } client.rename(oldName, newName); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -553,7 +584,7 @@ public class SrpConnection implements RedisConnection { } return (client.renamenx(oldName, newName).data() == 1); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -565,7 +596,7 @@ public class SrpConnection implements RedisConnection { } client.select(dbIndex); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -578,7 +609,7 @@ public class SrpConnection implements RedisConnection { } return client.ttl(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -591,7 +622,7 @@ public class SrpConnection implements RedisConnection { } return DataType.fromCode(client.type(key).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -600,7 +631,7 @@ public class SrpConnection implements RedisConnection { try { client.unwatch(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -617,7 +648,7 @@ public class SrpConnection implements RedisConnection { client.watch((Object[]) keys); } } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -635,7 +666,7 @@ public class SrpConnection implements RedisConnection { return client.get(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -648,7 +679,7 @@ public class SrpConnection implements RedisConnection { } client.set(key, value); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -662,7 +693,7 @@ public class SrpConnection implements RedisConnection { } return client.getset(key, value).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -675,7 +706,7 @@ public class SrpConnection implements RedisConnection { } return client.append(key, value).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -688,7 +719,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toBytesList(client.mget((Object[]) keys).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -701,7 +732,7 @@ public class SrpConnection implements RedisConnection { } client.mset((Object[]) SrpUtils.convert(tuples)); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -714,7 +745,7 @@ public class SrpConnection implements RedisConnection { } client.msetnx((Object[]) SrpUtils.convert(tuples)); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -727,7 +758,7 @@ public class SrpConnection implements RedisConnection { } client.setex(key, time, value); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -740,7 +771,7 @@ public class SrpConnection implements RedisConnection { } return client.setnx(key, value).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -753,7 +784,7 @@ public class SrpConnection implements RedisConnection { } return client.getrange(key, start, end).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -766,7 +797,7 @@ public class SrpConnection implements RedisConnection { } return client.decr(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -779,7 +810,7 @@ public class SrpConnection implements RedisConnection { } return client.decrby(key, value).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -792,7 +823,7 @@ public class SrpConnection implements RedisConnection { } return client.incr(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -805,7 +836,7 @@ public class SrpConnection implements RedisConnection { } return client.incrby(key, value).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -820,7 +851,7 @@ public class SrpConnection implements RedisConnection { } return (client.getbit(key, offset).data() == 1); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -835,7 +866,7 @@ public class SrpConnection implements RedisConnection { } client.setbit(key, offset, SrpUtils.asBit(value)); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -850,7 +881,7 @@ public class SrpConnection implements RedisConnection { } client.setrange(key, start, value); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -863,7 +894,7 @@ public class SrpConnection implements RedisConnection { } return client.strlen(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -880,7 +911,7 @@ public class SrpConnection implements RedisConnection { } return client.lpush(key, new Object[] { value }).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -893,7 +924,7 @@ public class SrpConnection implements RedisConnection { } return client.rpush(key, new Object[] { value }).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -907,7 +938,7 @@ public class SrpConnection implements RedisConnection { // return SrpUtils.toBytesList(client.blpop(timeout, keys).data()); throw new UnsupportedOperationException(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -921,7 +952,7 @@ public class SrpConnection implements RedisConnection { // return SrpUtils.toBytesList(client.brpop(timeout, keys).data()); throw new UnsupportedOperationException(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -934,7 +965,7 @@ public class SrpConnection implements RedisConnection { } return client.lindex(key, index).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -947,7 +978,7 @@ public class SrpConnection implements RedisConnection { } return client.linsert(key, SrpUtils.convertPosition(where), pivot, value).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -960,7 +991,7 @@ public class SrpConnection implements RedisConnection { } return client.llen(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -973,7 +1004,7 @@ public class SrpConnection implements RedisConnection { } return client.lpop(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -986,7 +1017,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toBytesList(client.lrange(key, start, end).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -999,7 +1030,7 @@ public class SrpConnection implements RedisConnection { } return client.lrem(key, count, value).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1012,7 +1043,7 @@ public class SrpConnection implements RedisConnection { } client.lset(key, index, value); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1025,7 +1056,7 @@ public class SrpConnection implements RedisConnection { } client.ltrim(key, start, end); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1038,7 +1069,7 @@ public class SrpConnection implements RedisConnection { } return client.rpop(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1051,7 +1082,7 @@ public class SrpConnection implements RedisConnection { } return client.rpoplpush(srcKey, dstKey).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1064,7 +1095,7 @@ public class SrpConnection implements RedisConnection { } return client.brpoplpush(srcKey, dstKey, timeout).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1077,7 +1108,7 @@ public class SrpConnection implements RedisConnection { } return client.lpushx(key, value).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1090,7 +1121,7 @@ public class SrpConnection implements RedisConnection { } return client.rpushx(key, value).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1108,7 +1139,7 @@ public class SrpConnection implements RedisConnection { } return (client.sadd(key, new Object[] { value }).data() == 1); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1121,7 +1152,7 @@ public class SrpConnection implements RedisConnection { } return client.scard(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1134,7 +1165,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.sdiff((Object[]) keys).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1147,7 +1178,7 @@ public class SrpConnection implements RedisConnection { } return client.sdiffstore(destKey, (Object[]) keys).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1160,7 +1191,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.sinter((Object[]) keys).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1173,7 +1204,7 @@ public class SrpConnection implements RedisConnection { } return client.sinterstore(destKey, (Object[]) keys).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1186,7 +1217,7 @@ public class SrpConnection implements RedisConnection { } return client.sismember(key, value).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1199,7 +1230,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.smembers(key).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1212,7 +1243,7 @@ public class SrpConnection implements RedisConnection { } return client.smove(srcKey, destKey, value).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1225,7 +1256,7 @@ public class SrpConnection implements RedisConnection { } return client.spop(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1238,7 +1269,7 @@ public class SrpConnection implements RedisConnection { } return client.srandmember(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1251,7 +1282,7 @@ public class SrpConnection implements RedisConnection { } return client.srem(key, new Object[] { value }).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1264,7 +1295,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.sunion((Object[]) keys).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1277,7 +1308,7 @@ public class SrpConnection implements RedisConnection { } return client.sunionstore(destKey, (Object[]) keys).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1294,7 +1325,7 @@ public class SrpConnection implements RedisConnection { } return client.zadd(new Object[] { key, score, value }).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1307,7 +1338,7 @@ public class SrpConnection implements RedisConnection { } return client.zcard(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1320,7 +1351,7 @@ public class SrpConnection implements RedisConnection { } return client.zcount(key, min, max).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1333,7 +1364,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toDouble(client.zincrby(key, increment, value).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1361,7 +1392,7 @@ public class SrpConnection implements RedisConnection { } return client.zinterstore(args).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1373,7 +1404,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.zrange(key, start, end, null).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1386,7 +1417,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.convertTuple(client.zrange(key, start, end, SrpUtils.WITHSCORES)); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1399,7 +1430,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.zrangebyscore(key, min, max, null, null).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1412,7 +1443,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.convertTuple(client.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, null)); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1425,7 +1456,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.convertTuple(client.zrevrange(key, start, end, SrpUtils.WITHSCORES)); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1439,7 +1470,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.zrangebyscore(key, min, max, null, limit).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1453,7 +1484,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.convertTuple(client.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit)); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1466,7 +1497,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.zrevrangebyscore(key, min, max, null, limit).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1478,7 +1509,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.zrevrangebyscore(key, min, max, null, null).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1491,7 +1522,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.convertTuple(client.zrevrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit)); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1503,7 +1534,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.convertTuple(client.zrevrangebyscore(key, min, max, SrpUtils.WITHSCORES, null)); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1517,7 +1548,7 @@ public class SrpConnection implements RedisConnection { } return (Long) client.zrank(key, value).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1530,7 +1561,7 @@ public class SrpConnection implements RedisConnection { } return client.zrem(key, new Object[] { value }).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1543,7 +1574,7 @@ public class SrpConnection implements RedisConnection { } return client.zremrangebyrank(key, start, end).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1556,7 +1587,7 @@ public class SrpConnection implements RedisConnection { } return client.zremrangebyscore(key, min, max).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1569,7 +1600,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.zrevrange(key, start, end, null).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1582,7 +1613,7 @@ public class SrpConnection implements RedisConnection { } return (Long) client.zrevrank(key, value).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1595,7 +1626,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toDouble(client.zscore(key, value).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1613,7 +1644,7 @@ public class SrpConnection implements RedisConnection { } return client.zunionstore(destKey, sets.length, (Object[]) sets).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1630,7 +1661,7 @@ public class SrpConnection implements RedisConnection { } return client.hset(key, field, value).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1643,7 +1674,7 @@ public class SrpConnection implements RedisConnection { } return client.hsetnx(key, field, value).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1656,7 +1687,7 @@ public class SrpConnection implements RedisConnection { } return client.hdel(key, new Object[] { field }).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1669,7 +1700,7 @@ public class SrpConnection implements RedisConnection { } return client.hexists(key, field).data() == 1; } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1682,7 +1713,7 @@ public class SrpConnection implements RedisConnection { } return client.hget(key, field).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1695,7 +1726,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toMap(client.hgetall(key).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1708,7 +1739,7 @@ public class SrpConnection implements RedisConnection { } return client.hincrby(key, field, delta).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1721,7 +1752,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toSet(client.hkeys(key).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1734,7 +1765,7 @@ public class SrpConnection implements RedisConnection { } return client.hlen(key).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1747,7 +1778,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toBytesList(client.hmget(key, (Object[]) fields).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1760,7 +1791,7 @@ public class SrpConnection implements RedisConnection { } client.hmset(key, SrpUtils.convert(tuple)); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1773,7 +1804,7 @@ public class SrpConnection implements RedisConnection { } return SrpUtils.toBytesList(client.hvals(key).data()); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1793,7 +1824,7 @@ public class SrpConnection implements RedisConnection { } return client.publish(channel, message).data(); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1822,7 +1853,7 @@ public class SrpConnection implements RedisConnection { subscription = new SrpSubscription(listener, client); subscription.pSubscribe(patterns); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(ex); } } @@ -1839,7 +1870,7 @@ public class SrpConnection implements RedisConnection { subscription.subscribe(channels); } catch (Exception ex) { - throw convertSRAccessException(ex); + throw convertSrpAccessException(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 4b9bba276..e26e259c7 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -346,7 +346,7 @@ public abstract class AbstractConnectionIntegrationTests { connection.execute("ZadD", getClass() + "#foo\t0.90\titem"); } - @Test(expected = DataAccessException.class) + @Test(expected = RedisPipelineException.class) public void exceptionExecuteNativeWithPipeline() throws Exception { connection.openPipeline(); connection.execute("iNFo"); From ebe05a04a04f83a524b8d231e11f29c0d68fea46 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Mon, 25 Jun 2012 21:20:58 +0300 Subject: [PATCH 5/5] fix bug that return an empty list for #closePipeline --- .../connection/jedis/JedisConnection.java | 5 ++++- .../redis/connection/srp/SrpConnection.java | 13 +++++++------ .../AbstractConnectionIntegrationTests.java | 19 ++++++++++++++++++- 3 files changed, 29 insertions(+), 8 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java index d43ae07de..5bbb6d820 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java @@ -237,7 +237,7 @@ public class JedisConnection implements RedisConnection { @SuppressWarnings("unchecked") public List closePipeline() { if (pipeline != null) { - List execute = pipeline.syncAndReturnAll(); + List execute = pipeline.syncAndReturnAll(); if (execute != null && !execute.isEmpty()) { Exception cause = null; for (int i = 0; i < execute.size(); i++) { @@ -250,9 +250,12 @@ public class JedisConnection implements RedisConnection { execute.set(i, dataAccessException); } } + if (cause != null) { throw new RedisPipelineException(cause, execute); } + + return execute; } } return Collections.emptyList(); diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java index 9eeaa2ee1..5fda0f75d 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java @@ -64,13 +64,13 @@ public class SrpConnection implements RedisConnection { private PipelineTracker callback; private volatile SrpSubscription subscription; - private static class PipelineTracker implements FutureCallback { + private static class PipelineTracker implements FutureCallback { private final List results = Collections.synchronizedList(new ArrayList()); - private final List> futures = new ArrayList>(); + private final List> futures = new ArrayList>(); - public void onSuccess(Object result) { - results.add(result); + public void onSuccess(Reply result) { + results.add(result.data()); } public void onFailure(Throwable t) { @@ -87,7 +87,7 @@ public class SrpConnection implements RedisConnection { return results; } - public void addCommand(ListenableFuture future) { + public void addCommand(ListenableFuture future) { futures.add(future); Futures.addCallback(future, this); } @@ -191,6 +191,7 @@ public class SrpConnection implements RedisConnection { if (cause != null) { throw new RedisPipelineException(cause, execute); } + return execute; } } @@ -1882,7 +1883,7 @@ public class SrpConnection implements RedisConnection { } // processing method that adds a listener to the future in order to track down the results and close the pipeline - private void pipeline(ListenableFuture future) { + private void pipeline(ListenableFuture future) { callback.addCommand(future); } } \ No newline at end of file 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 e26e259c7..11dfa2185 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -347,7 +347,7 @@ public abstract class AbstractConnectionIntegrationTests { } @Test(expected = RedisPipelineException.class) - public void exceptionExecuteNativeWithPipeline() throws Exception { + public void testExceptionExecuteNativeWithPipeline() throws Exception { connection.openPipeline(); connection.execute("iNFo"); connection.execute("SET ", getClass() + "testSetNative", UUID.randomUUID().toString()); @@ -355,4 +355,21 @@ public abstract class AbstractConnectionIntegrationTests { connection.closePipeline(); } + @Test + public void testExecuteNativeWithPipeline() throws Exception { + String key1 = getClass() + "#ExecuteNativeWithPipeline#1"; + String value1 = UUID.randomUUID().toString(); + String key2 = getClass() + "#ExecuteNativeWithPipeline#2"; + String value2 = UUID.randomUUID().toString(); + + connection.openPipeline(); + connection.execute("SET", key1, value1); + connection.execute("SET", key2, value2); + connection.execute("GET", key1); + connection.execute("GET", key2); + List result = connection.closePipeline(); + assertEquals(4, result.size()); + assertArrayEquals(value1.getBytes(), (byte[]) result.get(2)); + assertArrayEquals(value2.getBytes(), (byte[]) result.get(3)); + } } \ No newline at end of file