class ReactiveStreamsMongoLockProvider implements ExtensibleLockProvider {
    static final String LOCK_UNTIL = "lockUntil";
    static final String LOCKED_AT = "lockedAt";
    static final String LOCKED_BY = "lockedBy";
    static final String ID = "_id";
    static final String DEFAULT_SHEDLOCK_COLLECTION_NAME = "shedLock";

    private final String hostname;
    private final MongoCollection<Document> collection;

    /** Uses Mongo to coordinate locks */
    public ReactiveStreamsMongoLockProvider(MongoDatabase mongoDatabase) {
        this(mongoDatabase.getCollection(DEFAULT_SHEDLOCK_COLLECTION_NAME));
    }

    /**
     * Uses Mongo to coordinate locks
     *
     * @param collection
     *            Mongo collection to be used
     */
    public ReactiveStreamsMongoLockProvider(MongoCollection<Document> collection) {
        this.collection = collection;
        this.hostname = Utils.getHostname();
    }

    @Override
    public Optional<SimpleLock> lock(LockConfiguration lockConfiguration) {<FILL_FUNCTION_BODY>}

    private Optional<SimpleLock> extend(LockConfiguration lockConfiguration) {
        Instant now = now();
        Bson update = set(LOCK_UNTIL, lockConfiguration.getLockAtMostUntil());

        Document updatedDocument = execute(getCollection()
                .findOneAndUpdate(
                        and(eq(ID, lockConfiguration.getName()), gt(LOCK_UNTIL, now), eq(LOCKED_BY, hostname)),
                        update));

        if (updatedDocument != null) {
            return Optional.of(new ReactiveMongoLock(lockConfiguration, this));
        } else {
            return Optional.empty();
        }
    }

    private void unlock(LockConfiguration lockConfiguration) {
        // Set lockUtil to now or lockAtLeastUntil whichever is later
        execute(getCollection()
                .findOneAndUpdate(
                        eq(ID, lockConfiguration.getName()),
                        combine(set(LOCK_UNTIL, lockConfiguration.getUnlockTime()))));
    }

    @Nullable
    static <T> T execute(Publisher<T> command) {
        SingleLockableSubscriber<T> subscriber = new SingleLockableSubscriber<>();
        command.subscribe(subscriber);
        subscriber.await();
        Throwable error = subscriber.getError();
        if (error != null) {
            if (error instanceof RuntimeException) {
                throw (RuntimeException) error;
            } else {
                throw new LockException("Error when executing Mongo statement", error);
            }
        } else {
            return subscriber.getValue();
        }
    }

    private MongoCollection<Document> getCollection() {
        return collection;
    }

    private Instant now() {
        return ClockProvider.now();
    }

    private static final class ReactiveMongoLock extends AbstractSimpleLock {
        private final ReactiveStreamsMongoLockProvider mongoLockProvider;

        private ReactiveMongoLock(
                LockConfiguration lockConfiguration, ReactiveStreamsMongoLockProvider mongoLockProvider) {
            super(lockConfiguration);
            this.mongoLockProvider = mongoLockProvider;
        }

        @Override
        public void doUnlock() {
            mongoLockProvider.unlock(lockConfiguration);
        }

        @Override
        public Optional<SimpleLock> doExtend(LockConfiguration newLockConfiguration) {
            return mongoLockProvider.extend(newLockConfiguration);
        }
    }
}

        Instant now = now();
        Bson update = combine(
                set(LOCK_UNTIL, lockConfiguration.getLockAtMostUntil()), set(LOCKED_AT, now), set(LOCKED_BY, hostname));
        try {
            // There are three possible situations:
            // 1. The lock document does not exist yet - it is inserted - we have the lock
            // 2. The lock document exists and lockUtil <= now - it is updated - we have the
            // lock
            // 3. The lock document exists and lockUtil > now - Duplicate key exception is
            // thrown
            execute(getCollection()
                    .findOneAndUpdate(
                            and(eq(ID, lockConfiguration.getName()), lte(LOCK_UNTIL, now)),
                            update,
                            new FindOneAndUpdateOptions().upsert(true)));
            return Optional.of(new ReactiveMongoLock(lockConfiguration, this));
        } catch (MongoServerException e) {
            if (e.getCode() == 11000) { // duplicate key
                // Upsert attempts to insert when there were no filter matches.
                // This means there was a lock with matching ID with lockUntil > now.
                return Optional.empty();
            } else {
                throw e;
            }
        }
    
<no_super_class>