@Override
public <T> T evalSha(String scriptSha, ReturnType returnType, int numKeys, byte[]... keysAndArgs) {
    if (isQueueing()) {
        throw new UnsupportedOperationException();
    }
    if (isPipelined()) {
        throw new UnsupportedOperationException();
    }

    RedisCommand<?> c = toCommand(returnType, "EVALSHA");
    List<Object> params = new ArrayList<Object>();
    params.add(scriptSha);
    params.add(numKeys);
    params.addAll(Arrays.asList(keysAndArgs));
    return write(null, ByteArrayCodec.INSTANCE, c, params.toArray());
}
 

@Override
public Flux<CommandResponse<KeyCommand, Flux<ByteBuffer>>> sScan(Publisher<KeyScanCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");
        Assert.notNull(command.getOptions(), "ScanOptions must not be null!");
        
        byte[] keyBuf = toByteArray(command.getKey());
        Flux<byte[]> flux = Flux.create(new SetReactiveIterator<byte[]>() {
            @Override
            protected RFuture<ListScanResult<Object>> scanIterator(RedisClient client, long nextIterPos) {
                if (command.getOptions().getPattern() == null) {
                    return executorService.readAsync(client, keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.SSCAN, 
                            keyBuf, nextIterPos, "COUNT", Optional.ofNullable(command.getOptions().getCount()).orElse(10L));
                }

                return executorService.readAsync(client, keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.SSCAN, 
                            keyBuf, nextIterPos, "MATCH", command.getOptions().getPattern(), 
                                                "COUNT", Optional.ofNullable(command.getOptions().getCount()).orElse(10L));
            }
        });
        return Mono.just(new CommandResponse<>(command, flux.map(v -> ByteBuffer.wrap(v))));
    });
}
 

@Override
protected void doSubscribe(byte[]... channels) {
    List<RFuture<?>> list = new ArrayList<RFuture<?>>();
    for (byte[] channel : channels) {
        RFuture<PubSubConnectionEntry> f = subscribeService.subscribe(ByteArrayCodec.INSTANCE, new ChannelName(channel), new BaseRedisPubSubListener() {
            @Override
            public void onMessage(CharSequence ch, Object message) {
                if (!Arrays.equals(((ChannelName) ch).getName(), channel)) {
                    return;
                }

                DefaultMessage msg = new DefaultMessage(((ChannelName) ch).getName(), (byte[])message);
                getListener().onMessage(msg, null);
            }
        });
        list.add(f);
    }
    for (RFuture<?> future : list) {
        connectionManager.getCommandExecutor().syncSubscription(future);
    }
}
 

@Override
public Flux<MultiValueResponse<List<ByteBuffer>, ByteBuffer>> mGet(Publisher<List<ByteBuffer>> keysets) {
    return execute(keysets, coll -> {

        Assert.notNull(coll, "List must not be null!");
        
        Object[] params = coll.stream().map(buf -> toByteArray(buf)).toArray(Object[]::new);

        Mono<List<byte[]>> m = read(null, ByteArrayCodec.INSTANCE, RedisCommands.MGET, params);
        return m.map(v -> {
            List<ByteBuffer> values = v.stream().map(array -> {
                if (array == null) {
                    return ByteBuffer.allocate(0);
                }
                return ByteBuffer.wrap(array);
            }).collect(Collectors.toList());
            return new MultiValueResponse<>(coll, values);
        });
    });
}
 

@Override
public Mono<Void> subscribe(ByteBuffer... channels) {
    monosListener.acquire();
    return Mono.defer(() -> {
        RedissonPromise<Void> result = new RedissonPromise<>();
        result.onComplete((r, ex) -> {
            monosListener.release();
        });
        CountableListener<Void> listener = new CountableListener<>(result, null, channels.length);
        for (ByteBuffer channel : channels) {
            ChannelName cn = toChannelName(channel);
            RFuture<PubSubConnectionEntry> f = subscribeService.subscribe(ByteArrayCodec.INSTANCE, cn);
            f.onComplete((res, e) -> RedissonReactiveSubscription.this.channels.put(cn, res));
            f.onComplete(listener);
        }

        return Mono.fromFuture(result);
    });
}
 

@Override
public List<ByteRecord> xClaim(byte[] key, String group, String newOwner, XClaimOptions options) {
    Assert.notNull(key, "Key must not be null!");
    Assert.notNull(group, "Group name must not be null!");
    Assert.notNull(newOwner, "NewOwner must not be null!");
    Assert.notEmpty(options.getIds(), "Ids collection must not be empty!");

    List<Object> params = new ArrayList<>();
    params.add(key);
    params.add(group);
    params.add(newOwner);
    params.add(Objects.requireNonNull(options.getIdleTime()).toMillis());
    params.addAll(Arrays.asList(options.getIdsAsStringArray()));

    return connection.write(key, ByteArrayCodec.INSTANCE, new RedisCommand<List<ByteRecord>>("XCLAIM",
                                    new ListMultiDecoder2(
                                            new ByteRecordReplayDecoder(key),
                                            new ObjectDecoder(new StreamIdDecoder()),
                                            new StreamObjectMapReplayDecoder()), RedisCommand.ValueType.MAP), params.toArray());
}
 

@Override
public Flux<CommandResponse<KeyCommand, Flux<Entry<ByteBuffer, ByteBuffer>>>> hScan(
        Publisher<KeyScanCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");
        Assert.notNull(command.getOptions(), "ScanOptions must not be null!");
        
        byte[] keyBuf = toByteArray(command.getKey());
        Flux<Entry<Object, Object>> flux = Flux.create(new MapReactiveIterator<Object, Object, Entry<Object, Object>>(null, null, 0) {
            @Override
            public RFuture<MapScanResult<Object, Object>> scanIterator(RedisClient client, long nextIterPos) {
                if (command.getOptions().getPattern() == null) {
                    return executorService.readAsync(client, keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.HSCAN, 
                            keyBuf, nextIterPos, "COUNT", Optional.ofNullable(command.getOptions().getCount()).orElse(10L));
                }

                return executorService.readAsync(client, keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.HSCAN, 
                            keyBuf, nextIterPos, "MATCH", command.getOptions().getPattern(), 
                                                "COUNT", Optional.ofNullable(command.getOptions().getCount()).orElse(10L));
            }
        });
        Flux<Entry<ByteBuffer, ByteBuffer>> f = flux.map(v -> Collections.singletonMap(ByteBuffer.wrap((byte[])v.getKey()), ByteBuffer.wrap((byte[])v.getValue())).entrySet().iterator().next());
        return Mono.just(new CommandResponse<>(command, f));
    });
}
 

@Override
public Mono<Void> subscribe(ByteBuffer... channels) {
    monosListener.acquire();
    return Mono.defer(() -> {
        RedissonPromise<Void> result = new RedissonPromise<>();
        result.onComplete((r, ex) -> {
            monosListener.release();
        });
        CountableListener<Void> listener = new CountableListener<>(result, null, channels.length);
        for (ByteBuffer channel : channels) {
            ChannelName cn = toChannelName(channel);
            RFuture<PubSubConnectionEntry> f = subscribeService.subscribe(ByteArrayCodec.INSTANCE, cn);
            f.onComplete((res, e) -> RedissonReactiveSubscription.this.channels.put(cn, res));
            f.onComplete(listener);
        }

        return Mono.fromFuture(result);
    });
}
 

@Override
public Flux<ByteBufferResponse<BRPopLPushCommand>> bRPopLPush(Publisher<BRPopLPushCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");
        Assert.notNull(command.getDestination(), "Destination key must not be null!");
        Assert.notNull(command.getTimeout(), "Timeout must not be null!");
        
        byte[] keyBuf = toByteArray(command.getKey());
        byte[] destinationBuf = toByteArray(command.getDestination());
        
        Mono<byte[]> m = write(keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.BRPOPLPUSH, 
                                keyBuf, destinationBuf, command.getTimeout().getSeconds());
        return m.map(v -> new ByteBufferResponse<>(command, ByteBuffer.wrap(v)));
    });
}
 

@Override
public Flux<ByteBufferResponse<BRPopLPushCommand>> bRPopLPush(Publisher<BRPopLPushCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");
        Assert.notNull(command.getDestination(), "Destination key must not be null!");
        Assert.notNull(command.getTimeout(), "Timeout must not be null!");
        
        byte[] keyBuf = toByteArray(command.getKey());
        byte[] destinationBuf = toByteArray(command.getDestination());
        
        Mono<byte[]> m = write(keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.BRPOPLPUSH, 
                                keyBuf, destinationBuf, command.getTimeout().getSeconds());
        return m.map(v -> new ByteBufferResponse<>(command, ByteBuffer.wrap(v)));
    });
}
 

@Override
public Flux<CommandResponse<KeyCommand, Flux<Tuple>>> zScan(Publisher<KeyScanCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");
        Assert.notNull(command.getOptions(), "ScanOptions must not be null!");
        
        byte[] keyBuf = toByteArray(command.getKey());
        Flux<Tuple> flux = Flux.create(new SetReactiveIterator<Tuple>() {
            @Override
            protected RFuture<ListScanResult<Object>> scanIterator(RedisClient client, long nextIterPos) {
                if (command.getOptions().getPattern() == null) {
                    return executorService.readAsync(client, keyBuf, ByteArrayCodec.INSTANCE, ZSCAN, 
                            keyBuf, nextIterPos, "COUNT", Optional.ofNullable(command.getOptions().getCount()).orElse(10L));
                }

                return executorService.readAsync(client, keyBuf, ByteArrayCodec.INSTANCE, ZSCAN, 
                            keyBuf, nextIterPos, "MATCH", command.getOptions().getPattern(), 
                                                "COUNT", Optional.ofNullable(command.getOptions().getCount()).orElse(10L));
            }
        });
        return Mono.just(new CommandResponse<>(command, flux));
    });
}
 

@Override
public Set<byte[]> zRangeByLex(byte[] key, Range range, Limit limit) {
    String min = value(range.getMin().getValue(), range.getMin().isIncluding(), "-");
    String max = value(range.getMax().getValue(), range.getMax().isIncluding(), "+");
    
    List<Object> args = new ArrayList<Object>();
    args.add(key);
    args.add(min);
    args.add(max);
    
    if (limit != null) {
        args.add("LIMIT");
        args.add(limit.getOffset());
        args.add(limit.getCount());
    }
    
    return read(key, ByteArrayCodec.INSTANCE, ZRANGEBYLEX, args.toArray());
}
 

@Override
protected void doSubscribe(byte[]... channels) {
    List<RFuture<?>> list = new ArrayList<RFuture<?>>();
    for (byte[] channel : channels) {
        RFuture<PubSubConnectionEntry> f = subscribeService.subscribe(ByteArrayCodec.INSTANCE, new ChannelName(channel), new BaseRedisPubSubListener() {
            @Override
            public void onMessage(CharSequence ch, Object message) {
                if (!Arrays.equals(((ChannelName) ch).getName(), channel)) {
                    return;
                }

                DefaultMessage msg = new DefaultMessage(((ChannelName) ch).getName(), (byte[])message);
                getListener().onMessage(msg, null);
            }
        });
        list.add(f);
    }
    for (RFuture<?> future : list) {
        connectionManager.getCommandExecutor().syncSubscription(future);
    }
}
 

@Override
public Flux<CommandResponse<KeyCommand, Flux<Tuple>>> zScan(Publisher<KeyScanCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");
        Assert.notNull(command.getOptions(), "ScanOptions must not be null!");
        
        byte[] keyBuf = toByteArray(command.getKey());
        Flux<Tuple> flux = Flux.create(new SetReactiveIterator<Tuple>() {
            @Override
            protected RFuture<ListScanResult<Object>> scanIterator(RedisClient client, long nextIterPos) {
                if (command.getOptions().getPattern() == null) {
                    return executorService.readAsync(client, keyBuf, ByteArrayCodec.INSTANCE, ZSCAN, 
                            keyBuf, nextIterPos, "COUNT", Optional.ofNullable(command.getOptions().getCount()).orElse(10L));
                }

                return executorService.readAsync(client, keyBuf, ByteArrayCodec.INSTANCE, ZSCAN, 
                            keyBuf, nextIterPos, "MATCH", command.getOptions().getPattern(), 
                                                "COUNT", Optional.ofNullable(command.getOptions().getCount()).orElse(10L));
            }
        });
        return Mono.just(new CommandResponse<>(command, flux));
    });
}
 

@Override
public Set<Tuple> zRangeByScoreWithScores(byte[] key, Range range, Limit limit) {
    String min = value(range.getMin().getValue(), range.getMin().isIncluding(), "-inf");
    String max = value(range.getMax().getValue(), range.getMax().isIncluding(), "+inf");
    
    List<Object> args = new ArrayList<Object>();
    args.add(key);
    args.add(min);
    args.add(max);
    args.add("WITHSCORES");
    
    if (limit != null) {
        args.add("LIMIT");
        args.add(limit.getOffset());
        args.add(limit.getCount());
    }
    
    return read(key, ByteArrayCodec.INSTANCE, ZRANGEBYSCORE, args.toArray());
}
 

@Override
public Set<Tuple> zRevRangeByScoreWithScores(byte[] key, Range range, Limit limit) {
    String min = value(range.getMin().getValue(), range.getMin().isIncluding(), "-inf");
    String max = value(range.getMax().getValue(), range.getMax().isIncluding(), "+inf");
    
    List<Object> args = new ArrayList<Object>();
    args.add(key);
    args.add(max);
    args.add(min);
    args.add("WITHSCORES");
    
    if (limit != null) {
        args.add("LIMIT");
        args.add(limit.getOffset());
        args.add(limit.getCount());
    }
    
    return read(key, ByteArrayCodec.INSTANCE, ZREVRANGEBYSCOREWITHSCORES, args.toArray());
}
 

@Override
public Flux<ByteBufferResponse<PopCommand>> pop(Publisher<PopCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");
        Assert.notNull(command.getDirection(), "Direction must not be null!");
        
        byte[] keyBuf = toByteArray(command.getKey());
        RedisCommand<Object> redisCommand = RedisCommands.LPOP;
        if (command.getDirection() == Direction.RIGHT) {
            redisCommand = RedisCommands.RPOP;
        }
        
        Mono<byte[]> m = write(keyBuf, ByteArrayCodec.INSTANCE, redisCommand, keyBuf);
        return m.map(v -> new ByteBufferResponse<>(command, ByteBuffer.wrap(v)));
    });
}
 

@Override
public Set<byte[]> zRangeByScore(byte[] key, Range range, Limit limit) {
    String min = value(range.getMin().getValue(), range.getMin().isIncluding(), "-inf");
    String max = value(range.getMax().getValue(), range.getMax().isIncluding(), "+inf");
    
    List<Object> args = new ArrayList<Object>();
    args.add(key);
    args.add(min);
    args.add(max);
    
    if (limit != null) {
        args.add("LIMIT");
        args.add(limit.getOffset());
        args.add(limit.getCount());
    }
    
    return read(key, ByteArrayCodec.INSTANCE, RedisCommands.ZRANGEBYSCORE, args.toArray());
}
 

@Override
public Flux<PopResponse> bPop(Publisher<BPopCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKeys(), "Keys must not be null!");
        Assert.notNull(command.getDirection(), "Direction must not be null!");
        Assert.notNull(command.getTimeout(), "Timeout must not be null!");
        
        RedisCommand<List<Object>> redisCommand = RedisCommands.BLPOP;
        if (command.getDirection() == Direction.RIGHT) {
            redisCommand = RedisCommands.BRPOP;
        }
        
        List<Object> params = new ArrayList<Object>(command.getKeys().size() + 1);
        params.addAll(command.getKeys().stream().map(v -> toByteArray(v)).collect(Collectors.toList()));
        params.add(command.getTimeout().getSeconds());
        
        Mono<List<byte[]>> m = write((byte[])params.get(0), ByteArrayCodec.INSTANCE, redisCommand, params.toArray());
        return m.map(v -> new PopResponse(command, 
                new PopResult(v.stream().map(e -> ByteBuffer.wrap(e)).collect(Collectors.toList()))));
    });
}
 

@Override
public Set<Tuple> zRevRangeByScoreWithScores(byte[] key, Range range, Limit limit) {
    String min = value(range.getMin().getValue(), range.getMin().isIncluding(), "-inf");
    String max = value(range.getMax().getValue(), range.getMax().isIncluding(), "+inf");
    
    List<Object> args = new ArrayList<Object>();
    args.add(key);
    args.add(max);
    args.add(min);
    args.add("WITHSCORES");
    
    if (limit != null) {
        args.add("LIMIT");
        args.add(limit.getOffset());
        args.add(limit.getCount());
    }
    
    return read(key, ByteArrayCodec.INSTANCE, ZREVRANGEBYSCOREWITHSCORES, args.toArray());
}
 

@Override
public Set<byte[]> zRangeByLex(byte[] key, Range range, Limit limit) {
    String min = value(range.getMin().getValue(), range.getMin().isIncluding(), "-");
    String max = value(range.getMax().getValue(), range.getMax().isIncluding(), "+");
    
    List<Object> args = new ArrayList<Object>();
    args.add(key);
    args.add(min);
    args.add(max);
    
    if (limit != null) {
        args.add("LIMIT");
        args.add(limit.getOffset());
        args.add(limit.getCount());
    }
    
    return read(key, ByteArrayCodec.INSTANCE, ZRANGEBYLEX, args.toArray());
}
 

@Override
public Flux<ByteBufferResponse<SetCommand>> getSet(Publisher<SetCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");
        Assert.notNull(command.getValue(), "Value must not be null!");

        if (command.getExpiration().isPresent() || command.getOption().isPresent()) {
            throw new IllegalArgumentException("Command must not define expiration nor option for GETSET.");
        }

        byte[] keyBuf = toByteArray(command.getKey());
        byte[] valueBuf = toByteArray(command.getValue());
        
        Mono<byte[]> m = write(keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.GETSET, keyBuf, valueBuf);
        return m.map(v -> new ByteBufferResponse<>(command, ByteBuffer.wrap(v)));
    });
}
 

@Override
public Set<Tuple> zRangeByScoreWithScores(byte[] key, Range range, Limit limit) {
    String min = value(range.getMin().getValue(), range.getMin().isIncluding(), "-inf");
    String max = value(range.getMax().getValue(), range.getMax().isIncluding(), "+inf");
    
    List<Object> args = new ArrayList<Object>();
    args.add(key);
    args.add(min);
    args.add(max);
    args.add("WITHSCORES");
    
    if (limit != null) {
        args.add("LIMIT");
        args.add(limit.getOffset());
        args.add(limit.getCount());
    }
    
    return read(key, ByteArrayCodec.INSTANCE, ZRANGEBYSCORE, args.toArray());
}
 

@Test
public void testGet() {
    // Set a key/value
    RBucket<byte[]> bucket = client.getBucket("test", ByteArrayCodec.INSTANCE);
    bucket.set("abcd".getBytes());

    // Check that the backend can retrieve the key/value
    assertArrayEquals("abcd".getBytes(), cache.get("test"));
}
 

@Override
public Mono<ByteBuffer> randomKey(RedisClusterNode node) {
    MasterSlaveEntry entry = getEntry(node);
    Mono<byte[]> m = executorService.reactive(() -> {
        return executorService.readRandomAsync(entry, ByteArrayCodec.INSTANCE, RedisCommands.RANDOM_KEY);
    });
    return m.map(v -> ByteBuffer.wrap(v));
}
 

@Override
public Flux<ByteBufferResponse<KeyCommand>> get(Publisher<KeyCommand> keys) {
    return execute(keys, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");

        byte[] keyBuf = toByteArray(command.getKey());
        Mono<byte[]> m = read(keyBuf, ByteArrayCodec.INSTANCE, GET, keyBuf);
        return m.map(v -> new ByteBufferResponse<>(command, ByteBuffer.wrap(v)))
                .defaultIfEmpty(new AbsentByteBufferResponse<>(command));
    });
}
 

@Override
public Flux<CommandResponse<SUnionCommand, Flux<ByteBuffer>>> sUnion(Publisher<SUnionCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKeys(), "Key must not be null!");

        List<byte[]> list = command.getKeys().stream().map(v -> toByteArray(v)).collect(Collectors.toList());
        Mono<Set<byte[]>> m = write((byte[])list.get(0), ByteArrayCodec.INSTANCE, RedisCommands.SUNION, list.toArray());
        return m.map(v -> new CommandResponse<>(command, 
                                Flux.fromIterable(v).map(e -> ByteBuffer.wrap(e))));
    });
}
 

@Override
public Flux<CommandResponse<KeyCommand, Flux<ByteBuffer>>> hVals(Publisher<KeyCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");

        byte[] keyBuf = toByteArray(command.getKey());
        
        Mono<List<byte[]>> m = read(keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.HVALS, keyBuf);
        return m.map(v -> new CommandResponse<>(command, Flux.fromIterable(v).map(e -> ByteBuffer.wrap(e))));
    });
}
 

@Override
public GeoResults<GeoLocation<byte[]>> geoRadius(byte[] key, Circle within, GeoRadiusCommandArgs args) {
    List<Object> params = new ArrayList<Object>();
    params.add(key);
    params.add(convert(within.getCenter().getX()));
    params.add(convert(within.getCenter().getY()));
    params.add(within.getRadius().getValue());
    params.add(within.getRadius().getMetric().getAbbreviation());
    
    RedisCommand<GeoResults<GeoLocation<byte[]>>> command;
    if (args.getFlags().contains(GeoRadiusCommandArgs.Flag.WITHCOORD)) {
        command = new RedisCommand<GeoResults<GeoLocation<byte[]>>>("GEORADIUS", postitionDecoder);
        params.add("WITHCOORD");
    } else {
        MultiDecoder<GeoResults<GeoLocation<byte[]>>> distanceDecoder = new ListMultiDecoder2(new GeoResultsDecoder(within.getRadius().getMetric()), new GeoDistanceDecoder());
        command = new RedisCommand<GeoResults<GeoLocation<byte[]>>>("GEORADIUS", distanceDecoder);
        params.add("WITHDIST");
    }
    
    if (args.getLimit() != null) {
        params.add("COUNT");
        params.add(args.getLimit());
    }
    if (args.getSortDirection() != null) {
        params.add(args.getSortDirection().name());
    }
    
    return read(key, ByteArrayCodec.INSTANCE, command, params.toArray());
}
 

@Override
public Flux<ByteBufferResponse<LIndexCommand>> lIndex(Publisher<LIndexCommand> commands) {
    return execute(commands, command -> {

        Assert.notNull(command.getKey(), "Key must not be null!");
        Assert.notNull(command.getIndex(), "Index value must not be null!");

        byte[] keyBuf = toByteArray(command.getKey());
        Mono<byte[]> m = read(keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.LINDEX, 
                                        keyBuf, command.getIndex());
        return m.map(v -> new ByteBufferResponse<>(command, ByteBuffer.wrap(v)));
    });
}