/**
 * {@link FluentMongoOperations#update(Class)} defines the entry point for performing modifications on potentially
 * already existing document without replacing the entire document. The domain type is used for both mapping the query
 * identifying the potential update candidates as well as the property mapping for the
 * {@link org.springframework.data.mongodb.core.query.Update} itself. <br />
 * The sample below would read something like the following using classic {@link MongoOperations}.
 *
 * <pre>
 *     <code>
 *         template.upsert(query(where("lastname").is("windu")), update("name", "mence"), Jedi.class, "star-wars");
 *     </code>
 * </pre>
 */
@Test
public void updateAndUpsert() {

	UpdateResult result = mongoOps.update(Jedi.class) // Jedi defines the collection and field mapping
			.matching(query(where("lastname").is("windu"))) // so "last" maps to "lastname".
			.apply(update("name", "mence")) // We'll update the "firstname" to "mence"
			.upsert(); // and add a new document if it does not exist already.

	assertThat(result.getMatchedCount()).isEqualTo(0);
	assertThat(result.getUpsertedId()).isNotNull();

	assertThat(
			mongoOps.query(Human.class).inCollection("star-wars").matching(query(where("firstname").is("mence"))).one())
					.contains(new Human("mence", "windu"));
}
 

protected final FluentFuture<Integer> doUpdate(
        final Constraints.ConstraintHost criteria,
        final Constraints.Constraint update,
        final UpdateOptions options) {

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

  return submit(new Callable<UpdateResult>() {
    @Override
    public UpdateResult call() {
      return collection()
          .updateMany(
          convertToBson(criteria),
          convertToBson(update),
          options);
    }
  }).lazyTransform(new Function<UpdateResult, Integer>() {
    @Override
    public Integer apply(UpdateResult input) {
      return (int) input.getModifiedCount();
    }
  });
}
 

public Mono<UpdateResult> addCoins(long gains) {
    final long coins = NumberUtils.truncateBetween(this.getCoins() + gains, 0, Config.MAX_COINS);

    // If the new coins amount is equal to the current one, no need to request an update
    if (coins == this.getCoins()) {
        LOGGER.debug("[DBMember {} / {}] Coins update useless, aborting: {} coins",
                this.getId().asLong(), this.getGuildId().asLong(), coins);
        return Mono.empty();
    }

    LOGGER.debug("[DBMember {} / {}] Coins update: {} coins", this.getId().asLong(), this.getGuildId().asLong(), coins);
    return this.update(Updates.set("members.$.coins", coins), this.toDocument().append("coins", coins))
            .then(Mono.defer(() -> {
                if (coins >= 1_000_000_000) {
                    return DatabaseManager.getUsers()
                            .getDBUser(this.getId())
                            .flatMap(dbUser -> dbUser.unlockAchievement(Achievement.CROESUS));
                }
                if (coins >= 1_000_000) {
                    return DatabaseManager.getUsers()
                            .getDBUser(this.getId())
                            .flatMap(dbUser -> dbUser.unlockAchievement(Achievement.MILLIONAIRE));
                }
                return Mono.empty();
            }));
}
 

@Override
public boolean updateOrCreate(final Collection<MongoApproval> mongoApprovals) {
    boolean result = true;
    for (MongoApproval mongoApproval : mongoApprovals) {
        final Update update = Update.update("expiresAt", mongoApproval.getExpiresAt())
                .set("status", mongoApproval.getStatus())
                .set("lastUpdatedAt", mongoApproval.getLastUpdatedAt());

        final UpdateResult upsert = mongoTemplate.upsert(byUserIdAndClientIdAndScope(mongoApproval), update, MongoApproval.class);

        if (!upsert.wasAcknowledged()) {
            result = false;
        }
    }
    return result;
}
 

private boolean takeOverTransactionInMongoDB(TransactionXid transactionXid, String source, String target) {
	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 globalFilter = Filters.eq(CONSTANTS_FD_GLOBAL, instanceId);
		Bson instIdFilter = Filters.eq("identifier", source);

		Document document = new Document("$set", new Document("identifier", target));

		UpdateResult result = collection.updateOne(Filters.and(globalFilter, instIdFilter), document);
		return result.getMatchedCount() == 1;
	} catch (RuntimeException rex) {
		logger.error("Error occurred while locking transaction(gxid= {}).", instanceId, rex);
		return false;
	}
}
 

@Override
public void updateAlert(MagicCardAlert alert) throws SQLException {
	Bson filter = new Document("alertItem.id", alert.getId());
	BasicDBObject obj = new BasicDBObject();
	obj.put(dbAlertField, alert);
	obj.put(dbIDField, IDGenerator.generate(alert.getCard()));

	UpdateResult res = db.getCollection(colAlerts, BasicDBObject.class).replaceOne(filter,
			BasicDBObject.parse(serialize(obj)));
	logger.debug(res);

}
 

/**
 * 更新多条
 */
@Test
public void updateMutTest(){
    Query query = Query.query(Criteria.where("tenant_code").is(null));
    Update update = Update.update("tenant_code", "XYS");
    UpdateResult updateResult = mongoTemplate.updateMulti(query, update, SysUser.class);
    log.info("update 影响行数："+updateResult.getModifiedCount());
}
 

@Override
public Future<@Nullable MongoClientUpdateResult> updateCollectionWithOptions(String collection, JsonObject query, JsonObject update, UpdateOptions options) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(query, "query cannot be null");
  requireNonNull(update, "update cannot be null");
  requireNonNull(options, "options cannot be null");

  MongoCollection<JsonObject> coll = getCollection(collection, options.getWriteOption());
  Bson bquery = wrap(encodeKeyWhenUseObjectId(query));
  Bson bupdate = wrap(encodeKeyWhenUseObjectId(generateIdIfNeeded(query, update, options)));


  com.mongodb.client.model.UpdateOptions updateOptions = new com.mongodb.client.model.UpdateOptions().upsert(options.isUpsert());
  if (options.getArrayFilters() != null && !options.getArrayFilters().isEmpty()) {
    final List<Bson> bArrayFilters = new ArrayList<>(options.getArrayFilters().size());
    options.getArrayFilters().getList().forEach(entry -> bArrayFilters.add(wrap(JsonObject.mapFrom(entry))));
    updateOptions.arrayFilters(bArrayFilters);
  }

  Publisher<UpdateResult> publisher;
  if (options.isMulti()) {
    publisher = coll.updateMany(bquery, bupdate, updateOptions);
  } else {
    publisher = coll.updateOne(bquery, bupdate, updateOptions);
  }

  Promise<UpdateResult> promise = vertx.promise();
  publisher.subscribe(new SingleResultSubscriber<>(promise));
  return promise.future().map(Utils::toMongoClientUpdateResult);
}
 

private void publishEvents(List<Event> sequentialEvents) {
    Preconditions.checkState(!sequentialEvents.isEmpty());
    //发布事件，先将事件hidden改为false,然后发qmq消息
    Long firstId = sequentialEvents.get(0).getId();
    Long lastId = sequentialEvents.get(sequentialEvents.size() - 1).getId();
    Query query = Query.query(Criteria.where("id").gte(firstId).lte(lastId));
    Update update = Update.update("_hidden", false);
    UpdateResult updateResult = mongoTemplate.updateMulti(query, update, Event.class);
    Preconditions.checkState(updateResult.wasAcknowledged());
    //getMatchedCount返回的是long，size返回的是int，不能使用equals
    Preconditions.checkState(updateResult.getMatchedCount() == sequentialEvents.size());
    //TODO
    logger.info("publishEvents定时任务发布了事件{}",
            sequentialEvents.stream().map(Event::getId).collect(Collectors.toList()));
}
 

private static void update() {
    MongoCollection<Person> collection = Mongo.getCollection("dianping-collection", Person.class);
    //更新一条document
    collection.updateOne(eq("name", "张三"), combine(set("age", 23), set("name", "Ada Lovelace")));

    // 更新多条document
    UpdateResult updateResult = collection.updateMany(not(eq("zip", null)), set("zip", null));
    System.out.println(updateResult.getModifiedCount());

    // 替换collection（理论上object id是不会被替换的）
    updateResult = collection.replaceOne(eq("name", "张三"), new Person("张三", 20, new Address("香柏广场", "广州", "10086")));
    System.out.println(updateResult.getModifiedCount());
}
 

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

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

/**
 * Saves the specified tile spec to the database.
 *
 * @param  stackId    stack identifier.
 * @param  tileSpec   specification to be saved.
 *
 * @throws IllegalArgumentException
 *   if any required parameters or transform spec references are missing.
 */
void saveTileSpec(final StackId stackId,
                  final TileSpec tileSpec)
        throws IllegalArgumentException {

    MongoUtil.validateRequiredParameter("stackId", stackId);
    MongoUtil.validateRequiredParameter("tileSpec", tileSpec);
    MongoUtil.validateRequiredParameter("tileSpec.tileId", tileSpec.getTileId());

    final MongoCollection<Document> tileCollection = getTileCollection(stackId);

    final String context = "tile spec with id '" + tileSpec.getTileId();
    validateTransformReferences(context, stackId, tileSpec.getTransforms());

    final Document query = new Document();
    query.put("tileId", tileSpec.getTileId());

    final Document tileSpecObject = Document.parse(tileSpec.toJson());

    final UpdateResult result = tileCollection.replaceOne(query, tileSpecObject, MongoUtil.UPSERT_OPTION);

    LOG.debug("saveTileSpec: {}.{},({}), upsertedId is {}",
              MongoUtil.fullName(tileCollection),
              MongoUtil.action(result),
              query.toJson(),
              result.getUpsertedId());
}
 

/**
 * Saves the specified transform spec to the database.
 *
 * @param  stackId        stack identifier.
 * @param  transformSpec  specification to be saved.
 *
 * @throws IllegalArgumentException
 *   if any required parameters or transform spec references are missing.
 */
void saveTransformSpec(final StackId stackId,
                       final TransformSpec transformSpec)
        throws IllegalArgumentException {

    MongoUtil.validateRequiredParameter("stackId", stackId);
    MongoUtil.validateRequiredParameter("transformSpec", transformSpec);
    MongoUtil.validateRequiredParameter("transformSpec.id", transformSpec.getId());

    final MongoCollection<Document> transformCollection = getTransformCollection(stackId);

    final String context = "transform spec with id '" + transformSpec.getId() + "'";
    validateTransformReferences(context, stackId, transformSpec);

    final Document query = new Document();
    query.put("id", transformSpec.getId());

    final Document transformSpecObject = Document.parse(transformSpec.toJson());

    final UpdateResult result = transformCollection.replaceOne(query,
                                                               transformSpecObject,
                                                               MongoUtil.UPSERT_OPTION);

    LOG.debug("saveTransformSpec: {}.{},({}), upsertedId is {}",
              MongoUtil.fullName(transformCollection),
              MongoUtil.action(result),
              query.toJson(),
              result.getUpsertedId());
}
 

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

@Test
void replaceOne() {
    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 newVillain = new Document("id", 5).append("name", "lex lutor").append("type", "villain")
            .append("stars", 3);
    Document newHeroes = new Document("id", 6).append("name", "supergirl").append("type", "heroes")
            .append("stars", 2);

    UpdateResult result = collection
            .replaceOne(new Document("id", 3), newVillain, new ReplaceOptions().bypassDocumentValidation(true))
            .await().indefinitely();
    UpdateResult result2 = collection.replaceOne(new Document("id", 2), newHeroes).await().indefinitely();
    UpdateResult result3 = collection.replaceOne(new Document("id", 50), newHeroes).await().indefinitely();

    assertThat(result.getMatchedCount()).isEqualTo(1);
    assertThat(result.getModifiedCount()).isEqualTo(1);
    assertThat(result2.getMatchedCount()).isEqualTo(1);
    assertThat(result2.getModifiedCount()).isEqualTo(1);
    assertThat(result3.getMatchedCount()).isEqualTo(0);
    assertThat(result3.getModifiedCount()).isEqualTo(0);
}
 

/**
 * 更新
 * 
 * @param dbName
 * @param collectionName
 * @param filter
 * @param document
 * @return
 */
public boolean updateOne(String dbName, String collectionName, Document filter, Document document) {
	if (filter != null && filter.size() > 0 && document != null) {
		UpdateResult result = mongoClient.getDatabase(dbName).getCollection(collectionName)
				.updateOne(new Document(filter), new Document("$set", new Document(document)));
		long modifiedCount = result.getModifiedCount();
		return modifiedCount > 0 ? true : false;
	}

	return false;
}
 

/**
 * 通过_id更新
 * 
 * @param dbName
 * @param collectionName
 * @param _id
 * @param document
 * @return
 */
public boolean updateById(String dbName, String collectionName, String _id, Document document) {
	ObjectId objectId = new ObjectId(_id);
	Bson filter = Filters.eq("_id", objectId);

	UpdateResult result = getDatabase(dbName).getCollection(collectionName).updateOne(filter,
			new Document("$set", document));
	long modifiedCount = result.getModifiedCount();

	return modifiedCount > 0 ? true : false;
}
 

@Override
public Result<UpdateResult> setAuthorListInfo(String id, String name, String desc, List<String> tag) {
    return creditOperateHandle.doCreditOperate(
            UserUtils.getUser(), CreditConstant.MODIFY_AUTHOR_LIST_INFO, id, () ->
                    mongoTemplate.updateFirst(
                            Query.query(Criteria.where("_id").is(id)),
                            Update.update("name", name).currentDate("updateTime").set("desc", desc).set("tag", tag),
                            AuthorGroup.class));
}
 

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

@POST
@Path("/{personId}")
public void updatePerson(@PathParam("personId") long id, @Valid Person p) {
    UpdateResult result = peopleCollection.replaceOne(eq("id", id), p.toDocument());
    if (result.getMatchedCount() != 1)
        personNotFound(id);
}
 

/**
 * Used to convert any result MongoOperation (either {@link DeleteResult} or {@link UpdateResult}
 * to a {@link Long}
 */
static void convertMongoResultToLong(Exchange exchange) {
    Message in = exchange.getIn();
    if (in.getBody() instanceof DeleteResult) {
        Long docsDeleted = in.getBody(DeleteResult.class).getDeletedCount();
        in.setBody(docsDeleted);
    } else if (in.getBody() instanceof UpdateResult) {
        Long docsUpdated = in.getBody(UpdateResult.class).getModifiedCount();
        in.setBody(docsUpdated);
    } else {
        LOGGER.warn("Impossible to convert the body, type was {}", in.getBody() == null ? null : in.getBody().getClass());
    }
}
 

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

@Test
public void mongoSaveNewAutogeneratedIDTest() {
    // When
    // Given
    String saveArguments = "{\"someText\":\"new\"}";
    UpdateResult result = template.requestBody("direct:start", saveArguments, UpdateResult.class);
    // Then
    List<Document> docsFound = collection.find(Filters.eq("_id",
        result.getUpsertedId().asObjectId())).into(new ArrayList<>());
    Assertions.assertThat(docsFound).hasSize(1);
    Assertions.assertThat(docsFound.get(0).getString("test")).isEqualTo("new");
    Assertions.assertThat(result.getUpsertedId().asObjectId().getValue()).isEqualTo(docsFound.get(0).getObjectId("_id"));
}
 

/**
 * {@code value} must be serializable or serialized.
 */
public <T> Mono<UpdateResult> updateSetting(Setting setting, T value) {
    LOGGER.debug("[DBGuild {}] Setting update: {}={}", this.getId().asLong(), setting, value);

    return Mono.from(DatabaseManager.getGuilds()
            .getCollection()
            .updateOne(
                    Filters.eq("_id", this.getId().asString()),
                    Updates.set(String.format("settings.%s", setting), value),
                    new UpdateOptions().upsert(true)))
            .doOnNext(result -> LOGGER.trace("[DBGuild {}] Setting update result: {}",
                    this.getId().asLong(), result))
            .doOnTerminate(() -> DB_REQUEST_COUNTER.labels(GuildsCollection.NAME).inc());
}
 

public Mono<UpdateResult> addToJackpot(long coins) {
    final long value = (long) Math.ceil(coins / 100.0f);

    LOGGER.debug("[Lottery] Jackpot update: {} coins", value);

    return Mono.from(this.getCollection()
            .updateOne(Filters.eq("_id", "jackpot"),
                    Updates.inc("jackpot", value),
                    new UpdateOptions().upsert(true)))
            .doOnNext(result -> LOGGER.trace("[Lottery] Jackpot update result: {}", result))
            .doOnTerminate(() -> DB_REQUEST_COUNTER.labels(LotteryCollection.NAME).inc());
}
 

public SyncUpdateResult execute(@Nullable final CoreStitchServiceClient service) {
  final UpdateResult localResult = this.dataSynchronizer.updateOne(
      namespace,
      filter,
      update,
      new UpdateOptions().upsert(this.syncUpdateOptions.isUpsert()));

  return new SyncUpdateResult(
      localResult.getMatchedCount(),
      localResult.getModifiedCount(),
      localResult.getUpsertedId()
  );
}
 

public SyncUpdateResult execute(@Nullable final CoreStitchServiceClient service) {
  final UpdateResult localResult = this.dataSynchronizer.updateMany(
      namespace,
      filter,
      update,
      new UpdateOptions().upsert(this.syncUpdateOptions.isUpsert()));

  return new SyncUpdateResult(
      localResult.getMatchedCount(),
      localResult.getModifiedCount(),
      localResult.getUpsertedId()
  );
}
 

@Override
long executeQuery(int threadId, long threadRunCount, long globalRunCount, long selectorId, long randomId){

    final Document doc = new Document("$set", new Document(RANDOM_LONG, randomId))
            .append("$inc", new Document(VERSION, 1));

    final UpdateResult res = THREAD_RUN_COUNT.equals(queriedField)?mongoCollection.updateMany(eq(queriedField, selectorId), doc)
            :ID.equals(queriedField)?mongoCollection.updateOne(eq(queriedField, selectorId), doc):null;
    return res!=null?res.getModifiedCount():0l;
}
 

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