@Override
    public Flux<ReactiveRedisConnection.BooleanResponse<RenameCommand>> renameNX(Publisher<RenameCommand> commands) {
        return execute(commands, command -> {
            Assert.notNull(command.getKey(), "Key must not be null!");
            Assert.notNull(command.getNewName(), "New name must not be null!");

            byte[] keyBuf = toByteArray(command.getKey());
            byte[] newKeyBuf = toByteArray(command.getNewName());

            if (executorService.getConnectionManager().calcSlot(keyBuf) == executorService.getConnectionManager().calcSlot(newKeyBuf)) {
                return super.renameNX(commands);
            }

            return exists(command.getNewName())
                    .zipWith(read(keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.DUMP, keyBuf))
                    .filter(newKeyExistsAndDump -> !newKeyExistsAndDump.getT1() && Objects.nonNull(newKeyExistsAndDump.getT2()))
                    .map(Tuple2::getT2)
                    .zipWhen(value ->
                            pTtl(command.getKey())
                                    .filter(Objects::nonNull)
                                    .map(ttl -> Math.max(0, ttl))
                                    .switchIfEmpty(Mono.just(0L))

                    )
                    .flatMap(valueAndTtl -> write(newKeyBuf, StringCodec.INSTANCE, RedisCommands.RESTORE, newKeyBuf, valueAndTtl.getT2(), valueAndTtl.getT1())
                            .then(Mono.just(true)))
                    .switchIfEmpty(Mono.just(false))
                    .doOnSuccess(didRename -> {
                        if (didRename) {
                            del(command.getKey());
                        }
                    })
                    .map(didRename -> new BooleanResponse<>(command, didRename));
        });
    }
@Test
    public void testRenameNX() {
        connection.stringCommands().set(originalKey, value).block();
        if (hasTtl) {
            connection.keyCommands().expire(originalKey, Duration.ofSeconds(1000)).block();
        }

        Integer originalSlot = getSlotForKey(originalKey);
        newKey = getNewKeyForSlot(new String(originalKey.array()), getTargetSlot(originalSlot));

        Boolean result = connection.keyCommands().renameNX(originalKey, newKey).block();

        assertThat(result).isTrue();
        assertThat(connection.stringCommands().get(newKey).block()).isEqualTo(value);
        if (hasTtl) {
            assertThat(connection.keyCommands().ttl(newKey).block()).isGreaterThan(0);
        } else {
            assertThat(connection.keyCommands().ttl(newKey).block()).isEqualTo(-1);
        }

        connection.stringCommands().set(originalKey, value).block();

        result = connection.keyCommands().renameNX(originalKey, newKey).block();

        assertThat(result).isFalse();
    }