RKeys.keysIterable added. #203

src/main/java/org/redisson/CommandExecutor.java

@@ -84,4 +84,6 @@ public interface CommandExecutor {
 
     <T, R> Future<R> readRandomAsync(RedisCommand<T> command, Codec codec, Object ... params);
 
+    <T, R> R read(Integer slot, Codec codec, RedisCommand<T> command, Object ... params);
+
 }

src/main/java/org/redisson/CommandExecutorService.java

@@ -177,6 +177,17 @@ public class CommandExecutorService implements CommandExecutor {
         return mainPromise;
     }
 
+    public <T, R> R read(Integer slot, Codec codec, RedisCommand<T> command, Object ... params) {
+        Future<R> res = readAsync(slot, codec, command, params);
+        return get(res);
+    }
+
+    public <T, R> Future<R> readAsync(Integer slot, Codec codec, RedisCommand<T> command, Object ... params) {
+        Promise<R> mainPromise = connectionManager.newPromise();
+        async(true, slot, null, codec, command, params, mainPromise, 0);
+        return mainPromise;
+    }
+
     public <T, R> Future<R> readAsync(String key, RedisCommand<T> command, Object ... params) {
         return readAsync(key, connectionManager.getCodec(), command, params);
     }

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

@@ -1,12 +1,18 @@
 package org.redisson;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.redisson.client.codec.StringCodec;
 import org.redisson.client.protocol.RedisCommands;
+import org.redisson.client.protocol.decoder.ListScanResult;
 import org.redisson.core.RKeys;
+import org.redisson.misc.CompositeIterable;
 
 import io.netty.util.concurrent.Future;
 
@@ -19,6 +25,74 @@ public class RedissonKeys implements RKeys {
         this.commandExecutor = commandExecutor;
     }
 
+
+    @Override
+    public Iterable<String> keysIterable() {
+        List<Iterable<String>> iterables = new ArrayList<Iterable<String>>();
+        for (final Integer slot : commandExecutor.getConnectionManager().getEntries().keySet()) {
+            Iterable<String> iterable = new Iterable<String>() {
+                @Override
+                public Iterator<String> iterator() {
+                    return createKeysIterator(slot);
+                }
+            };
+            iterables.add(iterable);
+        }
+        return new CompositeIterable<String>(iterables);
+    }
+
+    private ListScanResult<String> scanIterator(int slot, long startPos) {
+        return commandExecutor.read(slot, StringCodec.INSTANCE, RedisCommands.SCAN, startPos);
+    }
+
+    private Iterator<String> createKeysIterator(final int slot) {
+        return new Iterator<String>() {
+
+            private Iterator<String> iter;
+            private Long iterPos;
+
+            private boolean removeExecuted;
+            private String value;
+
+            @Override
+            public boolean hasNext() {
+                if (iter == null) {
+                    ListScanResult<String> res = scanIterator(slot, 0);
+                    iter = res.getValues().iterator();
+                    iterPos = res.getPos();
+                } else if (!iter.hasNext() && iterPos != 0) {
+                    ListScanResult<String> res = scanIterator(slot, iterPos);
+                    iter = res.getValues().iterator();
+                    iterPos = res.getPos();
+                }
+                return iter.hasNext();
+            }
+
+            @Override
+            public String next() {
+                if (!hasNext()) {
+                    throw new NoSuchElementException("No such element");
+                }
+
+                value = iter.next();
+                removeExecuted = false;
+                return value;
+            }
+
+            @Override
+            public void remove() {
+                if (removeExecuted) {
+                    throw new IllegalStateException("Element been already deleted");
+                }
+
+                iter.remove();
+                delete(value);
+                removeExecuted = true;
+            }
+
+        };
+    }
+
     @Override
     public String randomKey() {
         return commandExecutor.get(randomKeyAsync());

src/main/java/org/redisson/client/protocol/RedisCommands.java

@@ -43,6 +43,7 @@ import org.redisson.client.protocol.pubsub.PubSubStatusDecoder;
 
 public interface RedisCommands {
 
+    RedisCommand<ListScanResult<String>> SCAN = new RedisCommand<ListScanResult<String>>("SCAN", new NestedMultiDecoder(new ObjectListReplayDecoder<String>(), new ListScanResultReplayDecoder()), ValueType.MAP);
     RedisStrictCommand<String> RANDOM_KEY = new RedisStrictCommand<String>("RANDOMKEY");
     RedisStrictCommand<String> PING = new RedisStrictCommand<String>("PING");
 

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

@@ -4,6 +4,8 @@ import java.util.Collection;
 
 public interface RKeys extends RKeysAsync {
 
+    Iterable<String> keysIterable();
+
     String randomKey();
 
     /**

src/main/java/org/redisson/misc/CompositeIterable.java

@@ -0,0 +1,71 @@
+package org.redisson.misc;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+public class CompositeIterable<T> implements Iterable<T>, Iterator<T> {
+
+    private List<Iterable<T>> iterablesList;
+    private Iterable<T>[] iterables;
+
+    private Iterator<Iterator<T>> listIterator;
+    private Iterator<T> currentIterator;
+
+    public CompositeIterable(List<Iterable<T>> iterables) {
+        this.iterablesList = iterables;
+    }
+
+    public CompositeIterable(Iterable<T> ... iterables) {
+        this.iterables = iterables;
+    }
+
+    public CompositeIterable(CompositeIterable<T> iterable) {
+        this.iterables = iterable.iterables;
+        this.iterablesList = iterable.iterablesList;
+    }
+
+    @Override
+    public Iterator<T> iterator() {
+        List<Iterator<T>> iterators = new ArrayList<Iterator<T>>();
+        if (iterables != null) {
+            for (Iterable<T> iterable : iterables) {
+                iterators.add(iterable.iterator());
+            }
+        } else {
+            for (Iterable<T> iterable : iterablesList) {
+                iterators.add(iterable.iterator());
+            }
+        }
+        listIterator = iterators.iterator();
+        currentIterator = null;
+        return this;
+    }
+
+    @Override
+    public boolean hasNext() {
+        if (currentIterator == null || !currentIterator.hasNext()) {
+            while (listIterator.hasNext()) {
+                Iterator<T> iterator = listIterator.next();
+                if (iterator.hasNext()) {
+                    currentIterator = iterator;
+                    return true;
+                }
+            }
+            return false;
+        }
+        return currentIterator.hasNext();
+    }
+
+    @Override
+    public T next() {
+        hasNext();
+        return currentIterator.next();
+    }
+
+    @Override
+    public void remove() {
+        currentIterator.remove();
+    }
+
+}