Intoduce dedicated excpetion for Redis error "busy"

* Introduce JedisRedisBusyException

* Add test for JedisRedisBusyException in case of redis error "busy"

If redis returns a message starting with "-BUSY" a JedisRedisBusyException containing that message is thrown.

* Harmonize name of constant with those of other reponses

* Rename exception class to comply with conventions

Rename JedisRedisBusyException to JedisBusyException.

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

@@ -6,6 +6,7 @@
 
 import redis.clients.jedis.commands.ProtocolCommand;
 import redis.clients.jedis.exceptions.JedisAskDataException;
+import redis.clients.jedis.exceptions.JedisBusyException;
 import redis.clients.jedis.exceptions.JedisClusterException;
 import redis.clients.jedis.exceptions.JedisConnectionException;
 import redis.clients.jedis.exceptions.JedisDataException;
@@ -19,6 +20,8 @@ public final class Protocol {
   private static final String ASK_RESPONSE = "ASK";
   private static final String MOVED_RESPONSE = "MOVED";
   private static final String CLUSTERDOWN_RESPONSE = "CLUSTERDOWN";
+  private static final String BUSY_RESPONSE = "BUSY";
+
   public static final String DEFAULT_HOST = "localhost";
   public static final int DEFAULT_PORT = 6379;
   public static final int DEFAULT_SENTINEL_PORT = 26379;
@@ -114,6 +117,8 @@ private static void processError(final RedisInputStream is) {
           Integer.valueOf(askInfo[2])), Integer.valueOf(askInfo[0]));
     } else if (message.startsWith(CLUSTERDOWN_RESPONSE)) {
       throw new JedisClusterException(message);
+    } else if (message.startsWith(BUSY_RESPONSE)) {
+      throw new JedisBusyException(message);
     }
     throw new JedisDataException(message);
   }

src/main/java/redis/clients/jedis/exceptions/JedisBusyException.java

@@ -0,0 +1,19 @@
+package redis.clients.jedis.exceptions;
+
+public class JedisBusyException extends JedisDataException {
+
+    private static final long serialVersionUID = 3992655220229243478L;
+
+    public JedisBusyException(final String message) {
+        super(message);
+    }
+
+    public JedisBusyException(final Throwable cause) {
+        super(cause);
+    }
+
+    public JedisBusyException(final String message, final Throwable cause) {
+        super(message, cause);
+    }
+
+}

src/test/java/redis/clients/jedis/tests/ProtocolTest.java

@@ -13,6 +13,7 @@
 import org.junit.Test;
 
 import redis.clients.jedis.Protocol;
+import redis.clients.jedis.exceptions.JedisBusyException;
 import redis.clients.util.RedisInputStream;
 import redis.clients.util.RedisOutputStream;
 import redis.clients.util.SafeEncoder;
@@ -119,4 +120,17 @@ public void nullMultiBulkReply() {
     List<String> response = (List<String>) Protocol.read(new RedisInputStream(is));
     assertNull(response);
   }
-}
+
+  @Test
+  public void busyReply() {
+    final String busyMessage = "BUSY Redis is busy running a script.";
+    final InputStream is = new ByteArrayInputStream(('-' + busyMessage + "\r\n").getBytes());
+    try {
+      Protocol.read(new RedisInputStream(is));
+    } catch (final JedisBusyException e) {
+      assertEquals(busyMessage, e.getMessage());
+      return;
+    }
+    fail("Expected a JedisBusyException to be thrown.");
+  }
+}