Simple implementation for blpop and brpop

.gitignore

@@ -7,3 +7,7 @@ target
 cache
 *.releaseBackup
 release.properties
+.project
+.classpath
+.settings
+

netty-client/pom.xml

@@ -32,7 +32,7 @@
     <dependency>
       <groupId>org.webbitserver</groupId>
       <artifactId>webbit</artifactId>
-      <version>0.4.14</version>
+      <version>0.4.15</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
@@ -42,7 +42,7 @@
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-compiler-plugin</artifactId>
-        <version>2.5.1</version>
+        <version>3.1</version>
         <configuration>
           <source>1.7</source>
           <target>1.7</target>

netty/pom.xml

@@ -14,7 +14,7 @@
     <dependency>
       <groupId>io.netty</groupId>
       <artifactId>netty</artifactId>
-      <version>3.6.6.Final</version>
+      <version>3.9.4.Final</version>
     </dependency>
     <dependency>
       <groupId>com.github.spullara.redis</groupId>

netty4-client/pom.xml

@@ -36,7 +36,7 @@
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-compiler-plugin</artifactId>
-        <version>2.5.1</version>
+        <version>3.1</version>
         <configuration>
           <source>1.7</source>
           <target>1.7</target>

netty4-server/src/main/java/redis/server/netty/RedisServer.java

@@ -465,18 +465,20 @@ public interface RedisServer {
    * List
    *
    * @param key0
+   * @param timeout1
    * @return MultiBulkReply
    */
-  public MultiBulkReply blpop(byte[][] key0) throws RedisException;
+  public MultiBulkReply blpop(byte[][] key0, byte[] timeout1) throws RedisException;
 
   /**
    * Remove and get the last element in a list, or block until one is available
    * List
    *
    * @param key0
+   * @param timeout1
    * @return MultiBulkReply
    */
-  public MultiBulkReply brpop(byte[][] key0) throws RedisException;
+  public MultiBulkReply brpop(byte[][] key0, byte[] timeout1) throws RedisException;
 
   /**
    * Pop a value from a list, push it to another list and return it; or block until one is available

netty4-server/src/main/java/redis/server/netty/SimpleRedisServer.java

@@ -1,13 +1,5 @@
 package redis.server.netty;
 
-import io.netty.buffer.ByteBuf;
-import redis.netty4.*;
-import redis.util.*;
-
-import java.lang.reflect.Field;
-import java.security.SecureRandom;
-import java.util.*;
-
 import static java.lang.Double.parseDouble;
 import static java.lang.Integer.MAX_VALUE;
 import static redis.netty4.BulkReply.NIL_REPLY;
@@ -16,6 +8,31 @@
 import static redis.netty4.StatusReply.QUIT;
 import static redis.util.Encoding.bytesToNum;
 import static redis.util.Encoding.numToBytes;
+import io.netty.buffer.ByteBuf;
+
+import java.security.SecureRandom;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.SortedSet;
+
+import redis.netty4.BulkReply;
+import redis.netty4.IntegerReply;
+import redis.netty4.MultiBulkReply;
+import redis.netty4.Reply;
+import redis.netty4.StatusReply;
+import redis.util.BytesKey;
+import redis.util.BytesKeyObjectMap;
+import redis.util.BytesKeySet;
+import redis.util.BytesValue;
+import redis.util.ZSet;
+import redis.util.ZSetEntry;
 
 public class SimpleRedisServer implements RedisServer {
 
@@ -259,24 +276,6 @@ private static int _torange(byte[] offset1, int length) throws RedisException {
   }
 
   private static Random r = new SecureRandom();
-  private static Field tableField;
-  private static Field nextField;
-  private static Field mapField;
-
-  static {
-    try {
-      tableField = HashMap.class.getDeclaredField("table");
-      tableField.setAccessible(true);
-      nextField = Class.forName("java.util.HashMap$Entry").getDeclaredField("next");
-      nextField.setAccessible(true);
-      mapField = HashSet.class.getDeclaredField("map");
-      mapField.setAccessible(true);
-    } catch (Exception e) {
-      e.printStackTrace();
-      tableField = null;
-      nextField = null;
-    }
-  }
 
   private static RedisException noSuchKey() {
     return new RedisException("no such key");
@@ -1104,25 +1103,67 @@ public MultiBulkReply time() throws RedisException {
    * List
    *
    * @param key0
+   * @param timeout1
    * @return MultiBulkReply
    */
   @Override
-  public MultiBulkReply blpop(byte[][] key0) throws RedisException {
-    // TODO: Blocking
-    return null;
+  public MultiBulkReply blpop(byte[][] key0, byte[] timeout1) throws RedisException {
+    // TODO No async processing possible at this level, so just short waits are possible...
+    long endTime = System.currentTimeMillis() + 100; // timeoutMillis(timeout1);
+    do {
+      // TODO Why is the timeout doubled as the last entry in key0?
+      for (int i = 0; i < key0.length - 1; i++) {
+        byte[] key = key0[i];
+        BulkReply reply = lpop(key);
+        if (reply != NIL_REPLY) {
+          return new MultiBulkReply(new Reply[]{ new BulkReply(key) , reply });
+        }
+      }
+
+      Thread.yield();
+    } while (System.currentTimeMillis() < endTime);
+
+    return MultiBulkReply.EMPTY;
   }
 
   /**
    * Remove and get the last element in a list, or block until one is available
    * List
    *
    * @param key0
+   * @param timeout1
    * @return MultiBulkReply
    */
   @Override
-  public MultiBulkReply brpop(byte[][] key0) throws RedisException {
-    // TODO: Blocking
-    return null;
+  public MultiBulkReply brpop(byte[][] key0, byte[] timeout1) throws RedisException {
+    // TODO No async processing possible at this level, so just short waits are possible...
+    long endTime = System.currentTimeMillis() + 100; // timeoutMillis(timeout1);
+    do {
+      // TODO Why is the timeout doubled as the last entry in key0?
+      for (int i = 0; i < key0.length - 1; i++) {
+        byte[] key = key0[i];
+        BulkReply reply = rpop(key);
+        if (reply != NIL_REPLY) {
+          return new MultiBulkReply(new Reply[]{ new BulkReply(key) , reply });
+        }
+      }
+
+      Thread.yield();
+    } while (System.currentTimeMillis() < endTime);
+
+    return MultiBulkReply.EMPTY;
+  }
+
+  /**
+   * Decode timeout.
+   *
+   * @param timeout
+   *          Timeout string as bytes.
+   * @return Timeout in milliseconds.
+   */
+  private int timeoutMillis(byte[] timeout) {
+    // TODO Set encoding always to UTF-8?
+    return Integer.parseInt(new String(timeout)) * 1000;
   }
 
   /**
@@ -1545,8 +1586,8 @@ public MultiBulkReply keys(byte[] pattern0) throws RedisException {
       throw new RedisException("wrong number of arguments for KEYS");
     }
     List<Reply<ByteBuf>> replies = new ArrayList<Reply<ByteBuf>>();
-    Iterator<Object> it = data.keySet().iterator();
-    while(it.hasNext()) {        
+    Iterator<BytesKey> it = data.keySet().iterator();
+    while(it.hasNext()) {
       BytesKey key = (BytesKey) it.next();
       byte[] bytes = key.getBytes();
       boolean expired = false;
@@ -1696,39 +1737,21 @@ public BulkReply randomkey() throws RedisException {
     // This implementation mirrors that of Redis. I'm not
     // sure I believe that this is a great algorithm but
     // it beats the alternatives that are very inefficient.
-    if (tableField != null) {
-      int size = data.size();
-      if (size == 0) {
-        return NIL_REPLY;
-      }
-      try {
-        BytesKey key = getRandomKey(data);
-        return new BulkReply(key.getBytes());
-      } catch (Exception e) {
-        throw new RedisException(e);
-      }
+    int size = data.size();
+    if (size == 0) {
+      return NIL_REPLY;
+    }
+    try {
+      BytesKey key = getRandomKey(data.keySet());
+      return new BulkReply(key.getBytes());
+    } catch (Exception e) {
+      throw new RedisException(e);
     }
-    return null;
   }
 
-  private BytesKey getRandomKey(Map data1) throws IllegalAccessException {
-    Map.Entry[] table = (Map.Entry[]) tableField.get(data1);
-    int length = table.length;
-    Map.Entry entry;
-    do {
-      entry = table[r.nextInt(length)];
-    } while (entry == null);
-
-    int entries = 0;
-    Map.Entry current = entry;
-    do {
-      entries++;
-      current = (Map.Entry) nextField.get(current);
-    } while (current != null);
-    int choose = r.nextInt(entries);
-    current = entry;
-    while (choose-- != 0) current = (Map.Entry) nextField.get(current);
-    return (BytesKey) current.getKey();
+  private BytesKey getRandomKey(Set<BytesKey> data1) throws IllegalAccessException {
+    // TODO somewhat inefficient, but works in jdk8 too
+    return data1.toArray(new BytesKey[data1.size()])[r.nextInt(data1.size())];
   }
 
   /**
@@ -2021,7 +2044,7 @@ public MultiBulkReply hgetall(byte[] key0) throws RedisException {
     int size = hash.size();
     Reply[] replies = new Reply[size * 2];
     int i = 0;
-    for (Map.Entry<Object, byte[]> entry : hash.entrySet()) {
+    for (Map.Entry<BytesKey, byte[]> entry : hash.entrySet()) {
       replies[i++] = new BulkReply(((BytesKey) entry.getKey()).getBytes());
       replies[i++] = new BulkReply(entry.getValue());
     }
@@ -2427,13 +2450,10 @@ public IntegerReply smove(byte[] source0, byte[] destination1, byte[] member2) t
    */
   @Override
   public BulkReply spop(byte[] key0) throws RedisException {
-    if (mapField == null || tableField == null) {
-      throw new RedisException("Not supported");
-    }
     BytesKeySet set = _getset(key0, false);
     if (set.size() == 0) return NIL_REPLY;
     try {
-      BytesKey key = getRandomKey((Map) mapField.get(set));
+      BytesKey key = getRandomKey(set);
       set.remove(key);
       return new BulkReply(key.getBytes());
     } catch (IllegalAccessException e) {
@@ -2450,15 +2470,12 @@ public BulkReply spop(byte[] key0) throws RedisException {
    */
   @Override
   public Reply srandmember(byte[] key0, byte[] count1) throws RedisException {
-    if (mapField == null || tableField == null) {
-      throw new RedisException("Not supported");
-    }
     BytesKeySet set = _getset(key0, false);
     int size = set.size();
     try {
       if (count1 == null) {
         if (size == 0) return NIL_REPLY;
-        BytesKey key = getRandomKey((Map) mapField.get(set));
+        BytesKey key = getRandomKey(set);
         return new BulkReply(key.getBytes());
       } else {
         int count = _toint(count1);
@@ -2475,7 +2492,7 @@ public Reply srandmember(byte[] key0, byte[] count1) throws RedisException {
         for (int i = 0; i < count; i++) {
           BytesKey key;
           do {
-            key = getRandomKey((Map) mapField.get(set));
+            key = getRandomKey(set);
           } while (found != null && !found.add(key));
           replies[i] = new BulkReply(key.getBytes());
         }

netty4/pom.xml

@@ -14,12 +14,12 @@
     <dependency>
       <groupId>io.netty</groupId>
       <artifactId>netty-all</artifactId>
-      <version>4.0.10.Final</version>
+      <version>4.0.23.Final</version>
     </dependency>
     <dependency>
       <groupId>org.javassist</groupId>
       <artifactId>javassist</artifactId>
-      <version>3.18.0-GA</version>
+      <version>3.18.2-GA</version>
     </dependency>
     <dependency>
       <groupId>com.github.spullara.redis</groupId>

pom.xml

@@ -22,6 +22,10 @@
   <url>http://github.com/spullara/redis-protocol</url>
   <description>A very fast Java redis redis.client.</description>
 
+  <prerequisites>
+    <maven>3.2</maven>
+  </prerequisites>
+
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
   </properties>
@@ -72,21 +76,21 @@
     <dependency>
       <groupId>com.google.guava</groupId>
       <artifactId>guava</artifactId>
-      <version>13.0.1</version>
+      <version>14.0.1</version>
     </dependency>
 
     <!-- CLI -->
     <dependency>
       <groupId>com.github.spullara.cli-parser</groupId>
       <artifactId>cli-parser</artifactId>
-      <version>1.1</version>
+      <version>1.1.2</version>
     </dependency>
 
     <!-- Testing -->
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
-      <version>4.8.2</version>
+      <version>4.11</version>
       <scope>test</scope>
     </dependency>
   </dependencies>
@@ -96,10 +100,10 @@
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-compiler-plugin</artifactId>
-        <version>2.5.1</version>
+        <version>3.1</version>
         <configuration>
-          <source>1.6</source>
-          <target>1.6</target>
+          <source>1.7</source>
+          <target>1.7</target>
           <encoding>UTF-8</encoding>
         </configuration>
       </plugin>
@@ -120,7 +124,7 @@
           <plugin>
             <groupId>org.apache.maven.plugins</groupId>
             <artifactId>maven-gpg-plugin</artifactId>
-            <version>1.4</version>
+            <version>1.5</version>
             <executions>
               <execution>
                 <id>sign-artifacts</id>