@Test
void findAndUpdate() {
    ReactiveMongoDatabase database = client.getDatabase(DATABASE);
    ReactiveMongoCollection<Document> collection = database.getCollection("test");

    CompletableFuture.allOf(
            collection
                    .insertOne(new Document("id", 1).append("name", "superman").append("type", "heroes")
                            .append("stars", 5))
                    .subscribeAsCompletionStage(),
            collection.insertOne(
                    new Document("id", 2).append("name", "batman").append("type", "heroes").append("stars", 4))
                    .subscribeAsCompletionStage(),
            collection
                    .insertOne(new Document("id", 3).append("name", "frogman").append("type", "villain")
                            .append("stars", 1))
                    .subscribeAsCompletionStage(),
            collection.insertOne(
                    new Document("id", 4).append("name", "joker").append("type", "villain").append("stars", 5))
                    .subscribeAsCompletionStage())
            .join();

    Document frogman = collection.findOneAndUpdate(new Document("id", 3), inc("stars", 3),
            new FindOneAndUpdateOptions().returnDocument(ReturnDocument.AFTER)).await().indefinitely();
    Document batman = collection.findOneAndUpdate(new Document("id", 2), inc("stars", -1)).await().indefinitely();

    assertThat(frogman).contains(entry("stars", 4), entry("name", "frogman")); // Returned after update
    assertThat(batman).contains(entry("stars", 4), entry("name", "batman")); // Returned the before update

}
 

private void initializeClusterInstanceVersion() {
	String databaseName = CommonUtils.getApplication(this.endpoint).replaceAll("\\W", "_");
	MongoDatabase database = this.mongoClient.getDatabase(databaseName);
	MongoCollection<Document> instances = database.getCollection(CONSTANTS_TB_INSTS);

	Bson condition = Filters.eq("_id", this.endpoint);

	Document increases = new Document();
	increases.append("version", 1L);

	Document document = new Document();
	document.append("$inc", increases);

	Document target = instances.findOneAndUpdate(condition, document, new FindOneAndUpdateOptions().upsert(true));
	this.instanceVersion = (target == null) ? 1 : (target.getLong("version") + 1);
}
 

protected final FluentFuture<Optional<T>> doModify(
        final Constraints.ConstraintHost criteria,
        final Constraints.Constraint update,
        final FindOneAndUpdateOptions options) {

  checkNotNull(criteria, "criteria");
  checkNotNull(update, "update");

  return submit(new Callable<Optional<T>>() {
    @Override
    public Optional<T> call() throws Exception {
      @Nullable T result = collection().findOneAndUpdate(
              convertToBson(criteria),
              convertToBson(update),
              options);

      return Optional.fromNullable(result);
    }
  });
}
 

protected final FluentFuture<Integer> doUpdateFirst(
        final Constraints.ConstraintHost criteria,
        final Constraints.Constraint update,
        final FindOneAndUpdateOptions options
) {
  checkNotNull(criteria, "criteria");
  checkNotNull(update, "update");
  checkNotNull(options, "options");

  return submit(new Callable<Integer>() {
    @Override
    public Integer call() {
      T result = collection().findOneAndUpdate(
              convertToBson(criteria),
              convertToBson(update),
              options);

      return result == null ? 0 : 1;
    }
  });
}
 

public long nextSequence(String name) {
    MongoDatabase mdb = mongoClient.getDatabase("dbus");
    MongoCollection mongoCollection = mdb.getCollection("dbus_sequence");
    Document filter = new Document("_id", name);
    Document update = new Document("$inc", new Document("value", 1L));
    FindOneAndUpdateOptions options = new FindOneAndUpdateOptions();
    options.upsert(true);
    options.returnDocument(ReturnDocument.AFTER);
    Document doc = (Document) mongoCollection.findOneAndUpdate(filter, update, options);
    return doc.getLong("value");
}
 

private boolean relockTransactionInMongoDB(TransactionXid transactionXid, String identifier) {
	byte[] global = transactionXid.getGlobalTransactionId();
	String instanceId = ByteUtils.byteArrayToString(global);

	try {
		String application = CommonUtils.getApplication(this.endpoint);
		String databaseName = application.replaceAll("\\W", "_");
		MongoDatabase mdb = this.mongoClient.getDatabase(databaseName);
		MongoCollection<Document> collection = mdb.getCollection(CONSTANTS_TB_LOCKS);

		Bson condition = Filters.eq(CONSTANTS_FD_GLOBAL, instanceId);

		Document increases = new Document();
		increases.append("times", 1);

		Document document = new Document();
		document.append("$inc", increases);

		collection.findOneAndUpdate(condition, document, new FindOneAndUpdateOptions().upsert(true));

		return true;
	} catch (com.mongodb.MongoWriteException error) {
		logger.error("Error occurred while locking transaction(gxid= {}).", instanceId, error);
		return false;
	} catch (RuntimeException rex) {
		logger.error("Error occurred while locking transaction(gxid= {}).", instanceId, rex);
		return false;
	}
}
 

public boolean reExitTransactionInMongoDB(TransactionXid transactionXid, String identifier) {
	byte[] global = transactionXid.getGlobalTransactionId();
	String instanceId = ByteUtils.byteArrayToString(global);

	try {
		String application = CommonUtils.getApplication(this.endpoint);
		String databaseName = application.replaceAll("\\W", "_");
		MongoDatabase mdb = this.mongoClient.getDatabase(databaseName);
		MongoCollection<Document> collection = mdb.getCollection(CONSTANTS_TB_LOCKS);

		Bson condition = Filters.eq(CONSTANTS_FD_GLOBAL, instanceId);

		Document increases = new Document();
		increases.append("times", -1);

		Document document = new Document();
		document.append("$inc", increases);

		Document target = collection.findOneAndUpdate(condition, document, new FindOneAndUpdateOptions().upsert(true));
		Integer times = target == null ? null : target.getInteger("times");

		return times == null ? true : times <= 0;
	} catch (com.mongodb.MongoWriteException error) {
		logger.error("Error occurred while locking transaction(gxid= {}).", instanceId, error);
		return true;
	} catch (RuntimeException rex) {
		logger.error("Error occurred while locking transaction(gxid= {}).", instanceId, rex);
		return true;
	}
}
 

@Override
@NonNull
public Optional<SimpleLock> lock(@NonNull LockConfiguration lockConfiguration) {
    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;
        }
    }
}
 

@Override
@NonNull
public Optional<SimpleLock> lock(@NonNull LockConfiguration lockConfiguration) {
    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
        getCollection().findOneAndUpdate(
            and(eq(ID, lockConfiguration.getName()), lte(LOCK_UNTIL, now)),
            update,
            new FindOneAndUpdateOptions().upsert(true)
        );
        return Optional.of(new MongoLock(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;
        }
    }
}
 

@Override
public Observable<TDocument> findOneAndUpdate(final Bson filter, final Bson update, final FindOneAndUpdateOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<TDocument>>() {
        @Override
        public void apply(final SingleResultCallback<TDocument> callback) {
            wrapped.findOneAndUpdate(filter, update, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Publisher<TDocument> findOneAndUpdate(final Bson filter, final Bson update, final FindOneAndUpdateOptions options) {
    return new ObservableToPublisher<TDocument>(com.mongodb.async.client.Observables.observe(
            new Block<com.mongodb.async.SingleResultCallback<TDocument>>() {
                @Override
                public void apply(final com.mongodb.async.SingleResultCallback<TDocument> callback) {
                    wrapped.findOneAndUpdate(filter, update, options, callback);
                }
            }));
}
 

@Override
public Publisher<TDocument> findOneAndUpdate(final ClientSession clientSession, final Bson filter, final Bson update,
                                             final FindOneAndUpdateOptions options) {
    return new ObservableToPublisher<TDocument>(com.mongodb.async.client.Observables.observe(
            new Block<com.mongodb.async.SingleResultCallback<TDocument>>() {
                @Override
                public void apply(final com.mongodb.async.SingleResultCallback<TDocument> callback) {
                    wrapped.findOneAndUpdate(clientSession.getWrapped(), filter, update, options, callback);
                }
            }));
}
 

@Override
public Publisher<TDocument> findOneAndUpdate(final Bson filter, final List<? extends Bson> update,
                                             final FindOneAndUpdateOptions options) {
    return new ObservableToPublisher<TDocument>(com.mongodb.async.client.Observables.observe(
            new Block<com.mongodb.async.SingleResultCallback<TDocument>>() {
                @Override
                public void apply(final com.mongodb.async.SingleResultCallback<TDocument> callback) {
                    wrapped.findOneAndUpdate(filter, update, options, callback);
                }
            }));
}
 

@Override
public Publisher<TDocument> findOneAndUpdate(final ClientSession clientSession, final Bson filter,
                                             final List<? extends Bson> update, final FindOneAndUpdateOptions options) {
    return new ObservableToPublisher<TDocument>(com.mongodb.async.client.Observables.observe(
            new Block<com.mongodb.async.SingleResultCallback<TDocument>>() {
                @Override
                public void apply(final com.mongodb.async.SingleResultCallback<TDocument> callback) {
                    wrapped.findOneAndUpdate(clientSession.getWrapped(), filter, update, options, callback);
                }
            }));
}
 

@Override
public Uni<T> findOneAndUpdate(Bson filter, Bson update, FindOneAndUpdateOptions options) {
    return Wrappers.toUni(collection.findOneAndUpdate(filter, update, options));
}
 

@Override
public Uni<T> findOneAndUpdate(ClientSession clientSession, Bson filter, Bson update,
        FindOneAndUpdateOptions options) {
    return Wrappers.toUni(collection.findOneAndUpdate(clientSession, filter, update, options));
}
 

@Override
public Observable<TDocument> findOneAndUpdate(final Bson filter, final Bson update) {
    return findOneAndUpdate(filter, update, new FindOneAndUpdateOptions());
}
 

@Override
public Publisher<TDocument> findOneAndUpdate(final Bson filter, final Bson update) {
    return findOneAndUpdate(filter, update, new FindOneAndUpdateOptions());
}
 

@Override
public Publisher<TDocument> findOneAndUpdate(final ClientSession clientSession, final Bson filter, final Bson update) {
    return findOneAndUpdate(clientSession, filter, update, new FindOneAndUpdateOptions());
}
 

@Override
public Publisher<TDocument> findOneAndUpdate(final Bson filter, final List<? extends Bson> update) {
    return findOneAndUpdate(filter, update, new FindOneAndUpdateOptions());
}
 

@Override
public Publisher<TDocument> findOneAndUpdate(final ClientSession clientSession, final Bson filter,
                                             final List<? extends Bson> update) {
    return findOneAndUpdate(clientSession, filter, update, new FindOneAndUpdateOptions());
}
 

@SuppressWarnings( "rawtypes" )
public cfData execute(cfSession _session, cfArgStructData argStruct ) throws cfmRunTimeException {
	MongoDatabase	db	= getMongoDatabase( _session, argStruct );
	
	String collection	= getNamedStringParam(argStruct, "collection", null);
	if ( collection == null )
		throwException(_session, "please specify a collection");
	
	cfData	update	= getNamedParam(argStruct, "update", null );
	if ( update == null )
		throwException(_session, "please specify update");
	
	cfData	query	= getNamedParam(argStruct, "query", null );
	if ( query == null )
		throwException(_session, "please specify query to update");

	try{
		
		MongoCollection<Document> col = db.getCollection(collection);
		FindOneAndUpdateOptions	findOneAndUpdateOptions	= new FindOneAndUpdateOptions();
		
		if ( getNamedParam(argStruct, "fields", null ) != null )
			findOneAndUpdateOptions.projection( getDocument( getNamedParam(argStruct, "fields", null ) ) );

		if ( getNamedParam(argStruct, "sort", null ) != null )
			findOneAndUpdateOptions.sort( getDocument( getNamedParam(argStruct, "sort", null ) ) );

		findOneAndUpdateOptions.upsert( getNamedBooleanParam(argStruct, "upsert", false ) );
		
		if ( getNamedBooleanParam(argStruct, "returnnew", false ) )
			findOneAndUpdateOptions.returnDocument( ReturnDocument.AFTER );
		
		Document qry = getDocument(query);
		long start = System.currentTimeMillis();
		
		Document result = col.findOneAndUpdate( qry, getDocument(update), findOneAndUpdateOptions );
		
		_session.getDebugRecorder().execMongo(col, "findandmodify", qry, System.currentTimeMillis()-start);
		
		return tagUtils.convertToCfData( (Map)result );

	} catch (MongoException me){
		throwException(_session, me.getMessage());
		return null;
	}
}
 

private <T> T findAndModify(String statement, Object parameter, ResultHandler handler, boolean upsert) {
  if (logger.isDebugEnabled()) {
    logger.debug("Execute 'findAndModify' mongodb command. Statement '" + statement + "'.");
  }

  UpdateConfig config = (UpdateConfig) configuration.getStatement(statement);
  if (config == null) {
    throw new MongoDaoException(statement, "FindAndModify statement id '" + statement + "' not found.");
  }

  String collection = config.getCollection();
  NodeEntry query = config.getQuery();
  NodeEntry action = config.getAction();
  NodeEntry field = config.getField();

  MongoDatabase db = getDatabase();
  MongoCollection<Document> coll = db.getCollection(collection);

  Map<String, Object> q = (Map<String, Object>) query.executorNode(config.getNamespace(), configuration, parameter);
  Map<String, Object> a = (Map<String, Object>) action.executorNode(config.getNamespace(), configuration, parameter);
  Map<String, Object> f = (Map<String, Object>) field.executorNode(config.getNamespace(), configuration, parameter);

  Document filter = new Document(q);
  Document update = (a == null) ? null : new Document(a);
  Document fieldDbo = (f == null) ? null : new Document(f);

  if (logger.isDebugEnabled()) {
    logger.debug("Execute 'findAndModify' mongodb command. Query '" + filter + "'.");
    logger.debug("Execute 'findAndModify' mongodb command. Action '" + update + "'.");
    logger.debug("Execute 'findAndModify' mongodb command. Field '" + fieldDbo + "'.");
  }

  Document document = coll.findOneAndUpdate(filter, update, new FindOneAndUpdateOptions().upsert(upsert));
  if (logger.isDebugEnabled()) {
    logger.debug("Execute 'findAndModify' mongodb command. Result is '" + document + "'.");
  }

  if (handler != null) {
    handler.handleResult(new ResultContext() {
      @Override
      public Object getResultObject() {
        return document;
      }

      @Override
      public int getResultCount() {
        if (document == null) {
          return 0;
        }
        return 1;
      }
    });
    return null;
  }
  return (T) helper.toResult(config.getNamespace(), field, document);
}
 

@Test
public void findAndModifyOptions() {
    scan(FindOneAndUpdateOptions.class, ModifyOptions.class, true, List.of(WriteConcern.class));
}
 

/**
 * Atomically find a document and update it.
 *
 * @param filter a document describing the query filter, which may not be null.
 * @param update a document describing the update, which may not be null. The update to apply must include only update
 *        operators.
 * @param options the options to apply to the operation
 * @return a {@link Uni} completed with the document that was updated. Depending on the value of the
 *         {@code returnOriginal}
 *         property, this will either be the document as it was before the update or as it is after the update. If no
 *         documents matched the
 *         query filter, then the uni is completed with {@code null}.
 */
Uni<T> findOneAndUpdate(Bson filter, Bson update, FindOneAndUpdateOptions options);
 

/**
 * Atomically find a document and update it.
 *
 * @param clientSession the client session with which to associate this operation
 * @param filter a document describing the query filter, which may not be null.
 * @param update a document describing the update, which may not be null. The update to apply must include only update
 *        operators.
 * @param options the options to apply to the operation
 * @return a {@link Uni} completed with the document that was updated. Depending on the value of the
 *         {@code returnOriginal}
 *         property, this will either be the document as it was before the update or as it is after the update. If no
 *         documents matched the
 *         query filter, then the uni is completed with {@code null}.
 */
Uni<T> findOneAndUpdate(ClientSession clientSession, Bson filter, Bson update,
        FindOneAndUpdateOptions options);
 

/**
 * Atomically find a document and update it.
 *
 * @param filter  a document describing the query filter, which may not be null.
 * @param update  a document describing the update, which may not be null. The update to apply must include only update operators.
 * @param options the options to apply to the operation
 * @return an Observable with a single element the document that was updated.  Depending on the value of the {@code returnOriginal}
 * property, this will either be the document as it was before the update or as it is after the update.
 * If no documents matched the query filter, then the observer will complete without emitting any items
 */
Observable<TDocument> findOneAndUpdate(Bson filter, Bson update, FindOneAndUpdateOptions options);
 

/**
 * Atomically find a document and update it.
 *
 * @param filter  a document describing the query filter, which may not be null.
 * @param update  a document describing the update, which may not be null. The update to apply must include only update operators.
 * @param options the options to apply to the operation
 * @return a publisher with a single element the document that was updated.  Depending on the value of the {@code returnOriginal}
 * property, this will either be the document as it was before the update or as it is after the update.  If no documents matched the
 * query filter, then null will be returned
 */
Publisher<TDocument> findOneAndUpdate(Bson filter, Bson update, FindOneAndUpdateOptions options);
 

/**
 * Atomically find a document and update it.
 *
 * @param clientSession the client session with which to associate this operation
 * @param filter  a document describing the query filter, which may not be null.
 * @param update  a document describing the update, which may not be null. The update to apply must include only update operators.
 * @param options the options to apply to the operation
 * @return a publisher with a single element the document that was updated.  Depending on the value of the {@code returnOriginal}
 * property, this will either be the document as it was before the update or as it is after the update.  If no documents matched the
 * query filter, then null will be returned
 * @mongodb.server.release 3.6
 * @since 1.7
 */
Publisher<TDocument> findOneAndUpdate(ClientSession clientSession, Bson filter, Bson update, FindOneAndUpdateOptions options);
 

/**
 * Atomically find a document and update it.
 *
 * <p>Note: Supports retryable writes on MongoDB server versions 3.6 or higher when the retryWrites setting is enabled.</p>
 * @param filter  a document describing the query filter, which may not be null.
 * @param update  a pipeline describing the update, which may not be null.
 * @param options the options to apply to the operation
 * @return a publisher with a single element the document that was updated.  Depending on the value of the {@code returnOriginal}
 * property, this will either be the document as it was before the update or as it is after the update.  If no documents matched the
 * query filter, then null will be returned
 * @since 1.12
 * @mongodb.server.release 4.2
 */
Publisher<TDocument> findOneAndUpdate(Bson filter, List<? extends Bson> update, FindOneAndUpdateOptions options);