class AbstractR2dbcStorageAccessor extends AbstractStorageAccessor {
    private final String tableName;

    public AbstractR2dbcStorageAccessor(@NonNull String tableName) {
        this.tableName = requireNonNull(tableName, "tableName can not be null");
    }

    @Override
    public boolean insertRecord(@NonNull LockConfiguration lockConfiguration) {
        return Boolean.TRUE.equals(
                Mono.from(insertRecordReactive(lockConfiguration)).block());
    }

    @Override
    public boolean updateRecord(@NonNull LockConfiguration lockConfiguration) {
        return Boolean.TRUE.equals(
                Mono.from(updateRecordReactive(lockConfiguration)).block());
    }

    @Override
    public boolean extend(@NonNull LockConfiguration lockConfiguration) {
        return Boolean.TRUE.equals(Mono.from(extendReactive(lockConfiguration)).block());
    }

    @Override
    public void unlock(@NonNull LockConfiguration lockConfiguration) {
        Mono.from(unlockReactive(lockConfiguration)).block();
    }

    public Publisher<Boolean> insertRecordReactive(@NonNull LockConfiguration lockConfiguration) {<FILL_FUNCTION_BODY>}

    public Publisher<Boolean> updateRecordReactive(@NonNull LockConfiguration lockConfiguration) {
        String sql = "UPDATE " + tableName + " SET lock_until = " + toParameter(1, "lock_until") + ", locked_at = "
                + toParameter(2, "locked_at") + ", locked_by = " + toParameter(3, "locked_by") + " WHERE name = "
                + toParameter(4, "name") + " AND lock_until <= " + toParameter(5, "now");
        return executeCommand(
                sql,
                statement -> {
                    Instant now = ClockProvider.now();
                    bind(statement, 0, "lock_until", lockConfiguration.getLockAtMostUntil());
                    bind(statement, 1, "locked_at", now);
                    bind(statement, 2, "locked_by", getHostname());
                    bind(statement, 3, "name", lockConfiguration.getName());
                    bind(statement, 4, "now", now);
                    return Mono.from(statement.execute())
                            .flatMap(it -> Mono.from(it.getRowsUpdated()))
                            .map(it -> it > 0);
                },
                this::handleUpdateException);
    }

    public Publisher<Boolean> extendReactive(@NonNull LockConfiguration lockConfiguration) {
        String sql = "UPDATE " + tableName + " SET lock_until = " + toParameter(1, "lock_until") + " WHERE name = "
                + toParameter(2, "name") + " AND locked_by = " + toParameter(3, "locked_by") + " AND lock_until > "
                + toParameter(4, "now");

        logger.debug("Extending lock={} until={}", lockConfiguration.getName(), lockConfiguration.getLockAtMostUntil());

        return executeCommand(
                sql,
                statement -> {
                    bind(statement, 0, "lock_until", lockConfiguration.getLockAtMostUntil());
                    bind(statement, 1, "name", lockConfiguration.getName());
                    bind(statement, 2, "locked_by", getHostname());
                    bind(statement, 3, "now", ClockProvider.now());
                    return Mono.from(statement.execute())
                            .flatMap(it -> Mono.from(it.getRowsUpdated()))
                            .map(it -> it > 0);
                },
                this::handleUnlockException);
    }

    public Publisher<Void> unlockReactive(@NonNull LockConfiguration lockConfiguration) {
        String sql = "UPDATE " + tableName + " SET lock_until = " + toParameter(1, "lock_until") + " WHERE name = "
                + toParameter(2, "name");
        return executeCommand(
                sql,
                statement -> {
                    bind(statement, 0, "lock_until", lockConfiguration.getUnlockTime());
                    bind(statement, 1, "name", lockConfiguration.getName());
                    return Mono.from(statement.execute())
                            .flatMap(it -> Mono.from(it.getRowsUpdated()))
                            .then();
                },
                (s, t) -> handleUnlockException(s, t).then());
    }

    protected abstract <T> Mono<T> executeCommand(
            String sql, Function<Statement, Mono<T>> body, BiFunction<String, Throwable, Mono<T>> exceptionHandler);

    protected abstract String toParameter(int index, String name);

    protected abstract void bind(Statement statement, int index, String name, Object value);

    Mono<Boolean> handleInsertionException(String sql, Throwable e) {
        if (e instanceof R2dbcDataIntegrityViolationException) {
            // lock record already exists
        } else {
            // can not throw exception here, some drivers (Postgres) do not throw
            // SQLIntegrityConstraintViolationException on duplicate key
            // we will try update in the next step, su if there is another problem, an
            // exception will be
            // thrown there
            logger.debug("Exception thrown when inserting record", e);
        }
        return Mono.just(false);
    }

    Mono<Boolean> handleUpdateException(String sql, Throwable e) {
        return Mono.error(new LockException("Unexpected exception when locking", e));
    }

    Mono<Boolean> handleUnlockException(String sql, Throwable e) {
        return Mono.error(new LockException("Unexpected exception when unlocking", e));
    }
}

        // Try to insert if the record does not exist (not optimal, but the simplest
        // platform agnostic
        // way)
        String sql = "INSERT INTO " + tableName + "(name, lock_until, locked_at, locked_by) VALUES("
                + toParameter(1, "name") + ", " + toParameter(2, "lock_until") + ", " + toParameter(3, "locked_at")
                + ", " + toParameter(4, "locked_by") + ")";
        return executeCommand(
                sql,
                statement -> {
                    bind(statement, 0, "name", lockConfiguration.getName());
                    bind(statement, 1, "lock_until", lockConfiguration.getLockAtMostUntil());
                    bind(statement, 2, "locked_at", ClockProvider.now());
                    bind(statement, 3, "locked_by", getHostname());
                    return Mono.from(statement.execute())
                            .flatMap(it -> Mono.from(it.getRowsUpdated()))
                            .map(it -> it > 0);
                },
                this::handleInsertionException);
    
<methods>public non-sealed void <init>() <variables>protected final Logger logger