RKeys.getKeysByPattern with count added

src/main/java/org/redisson/RedissonKeys.java

@@ -72,13 +72,17 @@ public class RedissonKeys implements RKeys {
     }
 
     @Override
-    public Iterable<String> getKeysByPattern(final String pattern) {
+    public Iterable<String> getKeysByPattern(String pattern) {
+        return getKeysByPattern(pattern, 10);
+    }
+    
+    public Iterable<String> getKeysByPattern(final String pattern, final int count) {
         List<Iterable<String>> iterables = new ArrayList<Iterable<String>>();
         for (final ClusterSlotRange slot : commandExecutor.getConnectionManager().getEntries().keySet()) {
             Iterable<String> iterable = new Iterable<String>() {
                 @Override
                 public Iterator<String> iterator() {
-                    return createKeysIterator(slot.getStartSlot(), pattern);
+                    return createKeysIterator(slot.getStartSlot(), pattern, count);
                 }
             };
             iterables.add(iterable);
@@ -86,26 +90,27 @@ public class RedissonKeys implements RKeys {
         return new CompositeIterable<String>(iterables);
     }
 
+
     @Override
     public Iterable<String> getKeys() {
         return getKeysByPattern(null);
     }
 
-    private ListScanResult<String> scanIterator(int slot, long startPos, String pattern) {
+    private ListScanResult<String> scanIterator(InetSocketAddress client, int slot, long startPos, String pattern, int count) {
         if (pattern == null) {
-            Future<ListScanResult<String>> f = commandExecutor.writeAsync(slot, StringCodec.INSTANCE, RedisCommands.SCAN, startPos);
+            Future<ListScanResult<String>> f = commandExecutor.readAsync(client, slot, StringCodec.INSTANCE, RedisCommands.SCAN, startPos, "COUNT", count);
             return commandExecutor.get(f);
         }
-        Future<ListScanResult<String>> f = commandExecutor.writeAsync(slot, StringCodec.INSTANCE, RedisCommands.SCAN, startPos, "MATCH", pattern);
+        Future<ListScanResult<String>> f = commandExecutor.readAsync(client, slot, StringCodec.INSTANCE, RedisCommands.SCAN, startPos, "MATCH", pattern, "COUNT", count);
         return commandExecutor.get(f);
     }
 
-    private Iterator<String> createKeysIterator(final int slot, final String pattern) {
+    private Iterator<String> createKeysIterator(final int slot, final String pattern, final int count) {
         return new RedissonBaseIterator<String>() {
 
             @Override
             ListScanResult<String> iterator(InetSocketAddress client, long nextIterPos) {
-                return RedissonKeys.this.scanIterator(slot, nextIterPos, pattern);
+                return RedissonKeys.this.scanIterator(client, slot, nextIterPos, pattern, count);
             }
 
             @Override

src/main/java/org/redisson/command/CommandAsyncExecutor.java

@@ -45,6 +45,8 @@ public interface CommandAsyncExecutor {
 
     <T, R> Future<R> writeAsync(Integer slot, Codec codec, RedisCommand<T> command, Object ... params);
 
+    <T, R> Future<R> readAsync(InetSocketAddress client, int slot, Codec codec, RedisCommand<T> command, Object ... params);
+    
     <T, R> Future<R> readAsync(InetSocketAddress client, String key, Codec codec, RedisCommand<T> command, Object ... params);
 
     <T, R> Future<R> evalWriteAllAsync(RedisCommand<T> command, SlotCallback<T, R> callback, String script, List<Object> keys, Object ... params);

src/main/java/org/redisson/command/CommandAsyncService.java

@@ -26,7 +26,6 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
 
 import org.redisson.RedisClientResult;
 import org.redisson.RedissonShutdownException;
@@ -116,6 +115,13 @@ public class CommandAsyncService implements CommandAsyncExecutor {
         return l.await(timeout, timeoutUnit);
     }
     
+    @Override
+    public <T, R> Future<R> readAsync(InetSocketAddress client, int slot, Codec codec, RedisCommand<T> command, Object ... params) {
+        Promise<R> mainPromise = connectionManager.newPromise();
+        async(true, new NodeSource(slot, client), codec, command, params, mainPromise, 0);
+        return mainPromise;
+    }
+    
     @Override
     public <T, R> Future<R> readAsync(InetSocketAddress client, String key, Codec codec, RedisCommand<T> command, Object ... params) {
         Promise<R> mainPromise = connectionManager.newPromise();

src/main/java/org/redisson/core/RKeys.java

@@ -37,17 +37,42 @@ public interface RKeys extends RKeysAsync {
     int getSlot(String key);
 
     /**
-     * Get all keys by pattern using iterator. Keys traversing with SCAN operation
-     *
+     * Get all keys by pattern using iterator. 
+     * Keys traversed with SCAN operation. Each SCAN operation loads 
+     * up to <b>10</b> keys per request. 
+     * <p/>
      *  Supported glob-style patterns:
+     *  <p/>
      *    h?llo subscribes to hello, hallo and hxllo
+     *    <p/>
      *    h*llo subscribes to hllo and heeeello
+     *    <p/>
      *    h[ae]llo subscribes to hello and hallo, but not hillo
-     *
+     * 
+     * @param pattern - match pattern
      * @return
      */
     Iterable<String> getKeysByPattern(String pattern);
 
+    /**
+     * Get all keys by pattern using iterator. 
+     * Keys traversed with SCAN operation. Each SCAN operation loads 
+     * up to <code>count</code> keys per request. 
+     * <p/>
+     *  Supported glob-style patterns:
+     *  <p/>
+     *    h?llo subscribes to hello, hallo and hxllo
+     *    <p/>
+     *    h*llo subscribes to hllo and heeeello
+     *    <p/>
+     *    h[ae]llo subscribes to hello and hallo, but not hillo
+     *
+     * @param pattern - match pattern
+     * @param count - keys loaded per request to Redis
+     * @return
+     */
+    Iterable<String> getKeysByPattern(String pattern, int count);
+    
     /**
      * Get all keys using iterator. Keys traversing with SCAN operation
      *