@Override
    public Cursor<byte[]> scan(RedisClusterNode node, ScanOptions options) {
        return new ScanCursor<byte[]>(0, options) {

            private RedisClient client = getEntry(node);
            
            @Override
            protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
                if (isQueueing() || isPipelined()) {
                    throw new UnsupportedOperationException("'SSCAN' cannot be called in pipeline / transaction mode.");
                }

                if (client == null) {
                    return null;
                }

                List<Object> args = new ArrayList<Object>();
                if (cursorId == 101010101010101010L) {
                    cursorId = 0;
                }
                args.add(Long.toUnsignedString(cursorId));
                if (options.getPattern() != null) {
                    args.add("MATCH");
                    args.add(options.getPattern());
                }
                if (options.getCount() != null) {
                    args.add("COUNT");
                    args.add(options.getCount());
                }
                
                RFuture<ListScanResult<byte[]>> f = executorService.readAsync(client, ByteArrayCodec.INSTANCE, RedisCommands.SCAN, args.toArray());
                ListScanResult<byte[]> res = syncFuture(f);
                String pos = res.getPos();
                client = res.getRedisClient();
                if ("0".equals(pos)) {
                    client = null;
                }
                
                return new ScanIteration<byte[]>(Long.parseUnsignedLong(pos), res.getValues());
            }
        }.open();
    }
@Test
    public void testScan() {
        Map<byte[], byte[]> map = new HashMap<>();
        for (int i = 0; i < 10000; i++) {
            map.put(RandomString.make(32).getBytes(), RandomString.make(32).getBytes(StandardCharsets.UTF_8));
        }
        connection.mSet(map);

        Cursor<byte[]> b = connection.scan(ScanOptions.scanOptions().build());
        Set<String> sett = new HashSet<>();
        int counter = 0;
        while (b.hasNext()) {
            byte[] tt = b.next();
            sett.add(new String(tt));
            counter++;
        }
        assertThat(sett.size()).isEqualTo(map.size());
        assertThat(counter).isEqualTo(map.size());
    }