Merge pull request #145 from TioBorracho/Cleaning

When there is a connection error, disconnect to leave everything clean.

.gitignore

@@ -4,3 +4,6 @@
 .gradle/
 target/
 build/
+/appendonly.aof
+/dump.rdb
+/nohup.out

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

@@ -61,6 +61,24 @@ public class Connection {
         }
     }
 
+    protected Object read() {
+        try {
+           return protocol.read(inputStream);
+        } catch (JedisConnectionException e) {
+            disconnect();
+            throw new JedisConnectionException(e);
+        }
+    }
+
+    protected void sendProtocolCommand(final Command cmd, final byte[]... args) {
+       try {
+            protocol.sendCommand(outputStream, cmd, args);
+        } catch (JedisConnectionException e) {
+            disconnect();
+            throw new JedisConnectionException(e);
+        }
+    }
+
     protected Connection sendCommand(final Command cmd, final String... args) {
         final byte[][] bargs = new byte[args.length][];
         for (int i = 0; i < args.length; i++) {
@@ -71,14 +89,14 @@ public class Connection {
 
     protected Connection sendCommand(final Command cmd, final byte[]... args) {
         connect();
-        protocol.sendCommand(outputStream, cmd, args);
+        sendProtocolCommand(cmd, args);
         pipelinedCommands++;
         return this;
     }
 
     protected Connection sendCommand(final Command cmd) {
         connect();
-        protocol.sendCommand(outputStream, cmd, new byte[0][]);
+        sendProtocolCommand(cmd, new byte[0][]);
         pipelinedCommands++;
         return this;
     }
@@ -145,7 +163,7 @@ public class Connection {
     protected String getStatusCodeReply() {
         flush();
         pipelinedCommands--;
-        final byte[] resp = (byte[]) protocol.read(inputStream);
+        final byte[] resp = (byte[]) read(); 
         if (null == resp) {
             return null;
         } else {
@@ -165,13 +183,13 @@ public class Connection {
     public byte[] getBinaryBulkReply() {
         flush();
         pipelinedCommands--;
-        return (byte[]) protocol.read(inputStream);
+        return (byte[]) read();
     }
 
     public Long getIntegerReply() {
         flush();
         pipelinedCommands--;
-        return (Long) protocol.read(inputStream);
+        return (Long) read();
     }
 
     public List<String> getMultiBulkReply() {
@@ -182,14 +200,14 @@ public class Connection {
     public List<byte[]> getBinaryMultiBulkReply() {
         flush();
         pipelinedCommands--;
-        return (List<byte[]>) protocol.read(inputStream);
+        return (List<byte[]>) read();
     }
 
     @SuppressWarnings("unchecked")
     public List<Object> getObjectMultiBulkReply() {
         flush();
         pipelinedCommands--;
-        return (List<Object>) protocol.read(inputStream);
+        return (List<Object>) read();
     }
 
     public List<Object> getAll() {
@@ -200,7 +218,7 @@ public class Connection {
         List<Object> all = new ArrayList<Object>();
         flush();
         while (pipelinedCommands > except) {
-            all.add(protocol.read(inputStream));
+            all.add(read());
             pipelinedCommands--;
         }
         return all;
@@ -209,6 +227,6 @@ public class Connection {
     public Object getOne() {
         flush();
         pipelinedCommands--;
-        return protocol.read(inputStream);
+        return read();
     }
-}
+}