class RedissonBasedProxyManagerBuilder<K> extends AbstractRedisProxyManagerBuilder<RedissonBasedProxyManagerBuilder<K>> {

        private final CommandAsyncExecutor commandExecutor;
        private Mapper<K> keyMapper;

        private RedissonBasedProxyManagerBuilder(Mapper<K> keyMapper, CommandAsyncExecutor commandExecutor) {
            this.keyMapper = Objects.requireNonNull(keyMapper);
            this.commandExecutor = Objects.requireNonNull(commandExecutor);
        }

        public <Key> RedissonBasedProxyManagerBuilder<Key> withKeyMapper(Mapper<Key> keyMapper) {
            this.keyMapper = (Mapper) Objects.requireNonNull(keyMapper);
            return (RedissonBasedProxyManagerBuilder) this;
        }

        public RedissonBasedProxyManager<K> build() {
            return new RedissonBasedProxyManager<>(this);
        }

    }

    private RedissonBasedProxyManager(RedissonBasedProxyManagerBuilder<K> builder) {
        super(builder.getClientSideConfig());
        this.commandExecutor = builder.commandExecutor;
        this.expirationStrategy = builder.getNotNullExpirationStrategy();
        this.keyMapper = builder.keyMapper;
    }

    @Override
    public boolean isExpireAfterWriteSupported() {
        return true;
    }

    @Override
    protected CompareAndSwapOperation beginCompareAndSwapOperation(K key) {
        String stringKey = keyMapper.toString(key);
        List<Object> keys = Collections.singletonList(stringKey);
        return new CompareAndSwapOperation() {
            @Override
            public Optional<byte[]> getStateData(Optional<Long> timeoutNanos) {
                RFuture<byte[]> persistedState = commandExecutor.readAsync(stringKey, ByteArrayCodec.INSTANCE, RedisCommands.GET, stringKey);
                return Optional.ofNullable(getWithTimeout(persistedState, timeoutNanos));
            }

            @Override
            public boolean compareAndSwap(byte[] originalData, byte[] newData, RemoteBucketState newState, Optional<Long> timeoutNanos) {<FILL_FUNCTION_BODY>

                long ttlMillis = calculateTtlMillis(newState);
                if (ttlMillis > 0) {
                    if (originalData == null) {
                        // Redisson prohibits the usage null as values, so "replace" must not be used in such cases
                        RFuture<Boolean> redissonFuture = commandExecutor.writeAsync(stringKey, ByteArrayCodec.INSTANCE, SET, stringKey, encodeByteArray(newData), "PX", ttlMillis, "NX");
                        return getWithTimeout(redissonFuture, timeoutNanos);
                    } else {
                        Object[] params = new Object[] {originalData, newData, ttlMillis};
                        RFuture<Boolean> redissonFuture = commandExecutor.evalWriteAsync(stringKey, ByteArrayCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, LuaScripts.SCRIPT_COMPARE_AND_SWAP_PX, keys, params);
                        return getWithTimeout(redissonFuture, timeoutNanos);
                    }
                } else {
                    if (originalData == null) {
                        // Redisson prohibits the usage null as values, so "replace" must not be used in such cases
                        RFuture<Boolean> redissonFuture = commandExecutor.writeAsync(stringKey, ByteArrayCodec.INSTANCE, SET, stringKey, encodeByteArray(newData), "NX");
                        return getWithTimeout(redissonFuture, timeoutNanos);
                    } else {
                        Object[] params = new Object[] {originalData, newData};
                        RFuture<Boolean> redissonFuture = commandExecutor.evalWriteAsync(stringKey, ByteArrayCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, LuaScripts.SCRIPT_COMPARE_AND_SWAP, keys, params);
                        return getWithTimeout(redissonFuture, timeoutNanos);
                    }
                }
            
<methods>public CommandResult<T> execute(K, Request<T>) ,public CompletableFuture<CommandResult<T>> executeAsync(K, Request<T>) <variables>private static final CommandResult<?> UNSUCCESSFUL_CAS_RESULT