import com.mongodb.client.model.ReplaceOneModel; //导入依赖的package包/类
@Override
public void fsync() {
    super.fsync();

    val count = new MutableInt();

    Stream.of( data.values()
        .stream()
        .filter( m -> m.modified >= lastFsync ) )
        .grouped( bulkSize )
        .forEach( list -> {
            count.add( list.size() );

            final List<? extends WriteModel<Metadata<T>>> bulk = Lists.map( list,
                metadata -> {
                    val id = identifier.get( metadata.object );
                    return new ReplaceOneModel<>( eq( "_id", new ObjectId( id ) ), metadata, UPDATE_OPTIONS_UPSERT );
                } );
            collection.bulkWrite( bulk );

        } );

    log.info( "[{}] fsync total: {}, modified: {}", collection.getNamespace(), size(), count.intValue() );
    lastFsync = System.currentTimeMillis();
}
 

import com.mongodb.client.model.ReplaceOneModel; //导入依赖的package包/类
public void Save(Document doc) {
	if (!doc.containsKey("_id")) {
		Create(doc);
		return;
	}
	Document find = new Document("_id", doc.get("_id"));
	UpdateOptions uo = new UpdateOptions();
	uo.upsert(true);
	ops.add(new ReplaceOneModel<Document>(find, doc, uo));
	FlushOpsIfFull();
}
 

import com.mongodb.client.model.ReplaceOneModel; //导入依赖的package包/类
/**
 * Inserts new resources to DB, or else updates existing ones descriptions if existing ID`s was used
 *
 * @param resource resources to be upserted
 */
public synchronized void insert(Set<Resource> resource) {
  List<ReplaceOneModel<Document>> upserts =
      resource.stream()
          .map(
              res -> new ReplaceOneModel<Document>(
                  eq("_id", res.getId()),
                  resourceToDocument(res),
                  new UpdateOptions().upsert(true)
              )
          ).collect(Collectors.toList());

  this.thisCollection.bulkWrite(upserts);
}
 

import com.mongodb.client.model.ReplaceOneModel; //导入依赖的package包/类
@Override
public void update(List<Node> nodes) {
	if (CollectionUtils.isNotEmpty(nodes)) {
		collection.bulkWrite(nodes
				.stream()
				.map(node -> {
					try {
						return new ReplaceOneModel<Document>(new Document("_id", node.getId()), Document.parse(mapper.writeValueAsString(node)),
								new UpdateOptions().upsert(true));
					} catch (Exception e) {
						throw Throwables.propagate(e);
					}
				}).collect(Collectors.toList()));
	}
}
 

import com.mongodb.client.model.ReplaceOneModel; //导入依赖的package包/类
@Override
public void save(List<Document> data) {
	if (data != null) {
		List<ReplaceOneModel<org.bson.Document>> updates = data.stream().map(toBson)
				.map(doc -> new ReplaceOneModel<org.bson.Document>(Filters.eq("_id", doc.getString("_id")), doc, new UpdateOptions().upsert(true)))
				.collect(Collectors.toList());
		collection.bulkWrite(updates);
	}
}
 

import com.mongodb.client.model.ReplaceOneModel; //导入依赖的package包/类
/**
 * Saves the specified tile spec to the database.
 *
 * @param  stackId            stack identifier.
 * @param  resolvedTileSpecs  collection of resolved tile specs (with referenced transforms).
 *
 * @throws IllegalArgumentException
 *   if any required parameters or transform spec references are missing.
 */
public void saveResolvedTiles(final StackId stackId,
                              final ResolvedTileSpecCollection resolvedTileSpecs)
        throws IllegalArgumentException {

    MongoUtil.validateRequiredParameter("stackId", stackId);
    MongoUtil.validateRequiredParameter("resolvedTileSpecs", resolvedTileSpecs);

    final Collection<TransformSpec> transformSpecs = resolvedTileSpecs.getTransformSpecs();
    final Collection<TileSpec> tileSpecs = resolvedTileSpecs.getTileSpecs();

    if (transformSpecs.size() > 0) {
        saveResolvedTransforms(stackId, transformSpecs);
    }

    if (tileSpecs.size() > 0) {

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

        final List<WriteModel<Document>> modelList = new ArrayList<>(tileSpecs.size());
        Document query = new Document();
        Document tileSpecObject;
        for (final TileSpec tileSpec : tileSpecs) {
            query = new Document("tileId", tileSpec.getTileId());
            tileSpecObject = Document.parse(tileSpec.toJson());
            modelList.add(new ReplaceOneModel<>(query, tileSpecObject, MongoUtil.UPSERT_OPTION));
        }

        final BulkWriteResult result = tileCollection.bulkWrite(modelList, MongoUtil.UNORDERED_OPTION);

        if (LOG.isDebugEnabled()) {
            final String bulkResultMessage = MongoUtil.toMessage("tile specs", result, tileSpecs.size());
            LOG.debug("saveResolvedTiles: {} using {}.initializeUnorderedBulkOp()",
                      bulkResultMessage, MongoUtil.fullName(tileCollection), query.toJson());
        }
    }

}
 

import com.mongodb.client.model.ReplaceOneModel; //导入依赖的package包/类
private List<WriteModel<JsonObject>> convertBulkOperations(List<BulkOperation> operations) {
  List<WriteModel<JsonObject>> result = new ArrayList<>(operations.size());
  for (BulkOperation bulkOperation : operations) {
    switch (bulkOperation.getType()) {
    case DELETE:
      Bson bsonFilter = toBson(encodeKeyWhenUseObjectId(bulkOperation.getFilter()));
      if (bulkOperation.isMulti()) {
        result.add(new DeleteManyModel<>(bsonFilter));
      } else {
        result.add(new DeleteOneModel<>(bsonFilter));
      }
      break;
    case INSERT:
      result.add(new InsertOneModel<>(encodeKeyWhenUseObjectId(bulkOperation.getDocument())));
      break;
    case REPLACE:
      result.add(new ReplaceOneModel<>(toBson(encodeKeyWhenUseObjectId(bulkOperation.getFilter())), bulkOperation.getDocument(),
          new com.mongodb.client.model.UpdateOptions().upsert(bulkOperation.isUpsert())));
      break;
    case UPDATE:
      Bson filter = toBson(encodeKeyWhenUseObjectId(bulkOperation.getFilter()));
      Bson document = toBson(encodeKeyWhenUseObjectId(bulkOperation.getDocument()));
      com.mongodb.client.model.UpdateOptions updateOptions = new com.mongodb.client.model.UpdateOptions()
          .upsert(bulkOperation.isUpsert());
      if (bulkOperation.isMulti()) {
        result.add(new UpdateManyModel<>(filter, document, updateOptions));
      } else {
        result.add(new UpdateOneModel<>(filter, document, updateOptions));
      }
      break;
    default:
      throw new IllegalArgumentException("Unknown bulk operation type: " + bulkOperation.getClass());
    }
  }
  return result;
}
 

import com.mongodb.client.model.ReplaceOneModel; //导入依赖的package包/类
public void saveMatches(final MatchCollectionId collectionId,
                        final List<CanvasMatches> matchesList)
        throws IllegalArgumentException {

    MongoUtil.validateRequiredParameter("collectionId", collectionId);
    MongoUtil.validateRequiredParameter("matchesList", matchesList);

    LOG.debug("saveMatches: entry, collectionId={}, matchesList.size()={}",
              collectionId, matchesList.size());

    if (matchesList.size() > 0) {

        final MongoCollection<Document> collection =
                matchDatabase.getCollection(collectionId.getDbCollectionName());

        ensureMatchIndexes(collection);

        final List<WriteModel<Document>> modelList = new ArrayList<>(matchesList.size());

        final UpdateOptions upsertOption = new UpdateOptions().upsert(true);
        Document filter;
        Document matchesObject;
        for (final CanvasMatches canvasMatches : matchesList) {
            canvasMatches.normalize();
            filter = new Document(
                    "pGroupId", canvasMatches.getpGroupId()).append(
                    "pId", canvasMatches.getpId()).append(
                    "qGroupId", canvasMatches.getqGroupId()).append(
                    "qId", canvasMatches.getqId());
            matchesObject = Document.parse(canvasMatches.toJson());
            modelList.add(new ReplaceOneModel<>(filter, matchesObject, upsertOption));
        }

        final BulkWriteResult result = collection.bulkWrite(modelList, MongoUtil.UNORDERED_OPTION);

        if (LOG.isDebugEnabled()) {
            final String bulkResultMessage = MongoUtil.toMessage("matches", result, matchesList.size());
            LOG.debug("saveMatches: {} using {}.initializeUnorderedBulkOp()",
                      bulkResultMessage, MongoUtil.fullName(collection));
        }
    }
}