Refactor Pipeline / Transaction to consume responses based on their requests, not rely on pipelinedCommands

* remove pipelinedCommands field at Connection class
** it was a risky state value
*** it was under 0 or over 0(though all commands are executed) while some situation
* remove Connection.getAll(), Connection.getAll(int except)

src/main/java/redis/clients/jedis/Connection.java

@@ -21,7 +21,6 @@ public class Connection {
     private Socket socket;
     private RedisOutputStream outputStream;
     private RedisInputStream inputStream;
-    private int pipelinedCommands = 0;
     private int timeout = Protocol.DEFAULT_TIMEOUT;
 
     public Socket getSocket() {
@@ -81,14 +80,12 @@ public class Connection {
     protected Connection sendCommand(final Command cmd, final byte[]... args) {
         connect();
         Protocol.sendCommand(outputStream, cmd, args);
-        pipelinedCommands++;
         return this;
     }
     
     protected Connection sendCommand(final Command cmd) {
         connect();
         Protocol.sendCommand(outputStream, cmd, new byte[0][]);
-        pipelinedCommands++;
         return this;
     }
 
@@ -161,7 +158,6 @@ public class Connection {
 
     protected String getStatusCodeReply() {
         flush();
-        pipelinedCommands--;
         final byte[] resp = (byte[]) Protocol.read(inputStream);
         if (null == resp) {
             return null;
@@ -181,13 +177,11 @@ public class Connection {
 
     public byte[] getBinaryBulkReply() {
         flush();
-        pipelinedCommands--;
         return (byte[]) Protocol.read(inputStream);
     }
 
     public Long getIntegerReply() {
         flush();
-        pipelinedCommands--;
         return (Long) Protocol.read(inputStream);
     }
 
@@ -198,45 +192,23 @@ public class Connection {
     @SuppressWarnings("unchecked")
     public List<byte[]> getBinaryMultiBulkReply() {
         flush();
-        pipelinedCommands--;
         return (List<byte[]>) Protocol.read(inputStream);
     }
 
     @SuppressWarnings("unchecked")
     public List<Object> getObjectMultiBulkReply() {
         flush();
-        pipelinedCommands--;
         return (List<Object>) Protocol.read(inputStream);
     }
     
     @SuppressWarnings("unchecked")
     public List<Long> getIntegerMultiBulkReply() {
         flush();
-        pipelinedCommands--;
         return (List<Long>) Protocol.read(inputStream);
     }
 
-    public List<Object> getAll() {
-        return getAll(0);
-    }
-
-    public List<Object> getAll(int except) {
-        List<Object> all = new ArrayList<Object>();
-        flush();
-        while (pipelinedCommands > except) {
-        	try{
-                all.add(Protocol.read(inputStream));
-        	}catch(JedisDataException e){
-        		all.add(e);
-        	}
-            pipelinedCommands--;
-        }
-        return all;
-    }
-
     public Object getOne() {
         flush();
-        pipelinedCommands--;
         return Protocol.read(inputStream);
     }
 }

src/main/java/redis/clients/jedis/Pipeline.java

@@ -62,17 +62,24 @@ public class Pipeline extends MultiKeyPipelineBase {
     protected Client getClient(String key) {
         return client;
     }
-
+    
+    public Object getOneWithJedisDataException() {
+    	try {
+    		return client.getOne();
+    	} catch (JedisDataException e) {
+    		return e;
+    	}
+    }
+    
     /**
      * Syncronize pipeline by reading all responses. This operation close the
      * pipeline. In order to get return values from pipelined commands, capture
      * the different Response<?> of the commands you execute.
      */
     public void sync() {
-        List<Object> unformatted = client.getAll();
-        for (Object o : unformatted) {
-            generateResponse(o);
-        }
+    	while (hasPipelinedResponse()) {
+    		generateResponse(getOneWithJedisDataException());
+    	}
     }
 
     /**
@@ -84,12 +91,11 @@ public class Pipeline extends MultiKeyPipelineBase {
      * @return A list of all the responses in the order you executed them.
      */
     public List<Object> syncAndReturnAll() {
-        List<Object> unformatted = client.getAll();
         List<Object> formatted = new ArrayList<Object>();
         
-        for (Object o : unformatted) {
-            try {
-            	formatted.add(generateResponse(o).get());
+        while (hasPipelinedResponse()) {
+        	try {
+            	formatted.add(generateResponse(getOneWithJedisDataException()).get());
             } catch (JedisDataException e) {
                 formatted.add(e);
             }

src/main/java/redis/clients/jedis/Queable.java

@@ -24,4 +24,11 @@ public class Queable {
         return lr;
     }
 
+    protected boolean hasPipelinedResponse() {
+    	return pipelinedResponses.size() > 0;
+    }
+    
+    protected int getPipelinedResponseLength() {
+    	return pipelinedResponses.size();
+    }
 }

src/main/java/redis/clients/jedis/Transaction.java

@@ -30,9 +30,26 @@ public class Transaction extends MultiKeyPipelineBase {
         return client;
     }
 
+    public Object getOneWithJedisDataException() {
+    	try {
+    		return client.getOne();
+    	} catch (JedisDataException e) {
+    		return e;
+    	}
+    }
+    
+    private void consumeResponse(int count) {
+    	for (int i = 0 ; i < count ; i++)
+    		getOneWithJedisDataException();
+    }
+    
     public List<Object> exec() {
+    	// Discard multi
+    	consumeResponse(1);
+    	// Discard QUEUED or ERROR
+    	consumeResponse(getPipelinedResponseLength());
+    	
         client.exec();
-        client.getAll(1); // Discard all but the last reply
 
         List<Object> unformatted = client.getObjectMultiBulkReply();
         if (unformatted == null) {
@@ -50,8 +67,12 @@ public class Transaction extends MultiKeyPipelineBase {
     }
 
     public List<Response<?>> execGetResponse() {
+    	// Discard multi
+    	consumeResponse(1);
+    	// Discard QUEUED or ERROR
+    	consumeResponse(getPipelinedResponseLength());
+    	
         client.exec();
-        client.getAll(1); // Discard all but the last reply
 
         List<Object> unformatted = client.getObjectMultiBulkReply();
         if (unformatted == null) {
@@ -65,8 +86,8 @@ public class Transaction extends MultiKeyPipelineBase {
     }
 
     public String discard() {
+    	consumeResponse(getPipelinedResponseLength());
         client.discard();
-        client.getAll(1); // Discard all but the last reply
         inTransaction = false;
         clean();
         return client.getStatusCodeReply();