@Override
    public Cursor<Tuple> zScan(byte[] key, ScanOptions options) {
        return new KeyBoundCursor<Tuple>(key, 0, options) {

            private RedisClient client;

            @Override
            protected ScanIteration<Tuple> doScan(byte[] key, long cursorId, ScanOptions options) {
                if (isQueueing() || isPipelined()) {
                    throw new UnsupportedOperationException("'ZSCAN' cannot be called in pipeline / transaction mode.");
                }

                List<Object> args = new ArrayList<Object>();
                args.add(key);
                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<Tuple>> f = executorService.readAsync(client, key, ByteArrayCodec.INSTANCE, ZSCAN, args.toArray());
                ListScanResult<Tuple> res = syncFuture(f);
                client = res.getRedisClient();
                return new ScanIteration<Tuple>(Long.parseUnsignedLong(res.getPos()), res.getValues());
            }
        }.open();
    }
@Test
    public void testZScan() {
        connection.zAdd("key".getBytes(), 1, "value1".getBytes());
        connection.zAdd("key".getBytes(), 2, "value2".getBytes());

        Cursor<Tuple> t = connection.zScan("key".getBytes(), ScanOptions.scanOptions().build());
        assertThat(t.hasNext()).isTrue();
        assertThat(t.next().getValue()).isEqualTo("value1".getBytes());
        assertThat(t.hasNext()).isTrue();
        assertThat(t.next().getValue()).isEqualTo("value2".getBytes());
    }