Java 类com.mongodb.async.client.MongoCollection 实例源码

/**
 * Returns the timestamp of the latest oplog entry.
 *
 * @param collection The oplog {@link MongoCollection}
 * @return The latest timestamp or {@code null} if no entry is available
 */
public static BsonTimestamp getLatestOplogTimestamp(MongoCollection<BsonDocument> collection) {
    final AtomicReference<BsonTimestamp> timestamp = new AtomicReference<>();
    final AtomicReference<Throwable> error = new AtomicReference<>();
    final CountDownLatch waiter = new CountDownLatch(1);
    collection.find().sort(new Document("$natural", -1)).limit(1).first(new SingleResultCallback<BsonDocument>() {
        @Override
        public void onResult(BsonDocument document, Throwable throwable) {
            if (throwable != null)
                error.set(throwable);
            if (document != null)
                timestamp.set(document.getTimestamp("ts"));
            waiter.countDown();
        }
    });
    ConcurrentUtils.safeAwait(waiter);
    Throwable realError = error.get();
    if (realError != null)
        throw Throwables.propagate(realError);
    return timestamp.get();
}

/**
 * Saves a set of {@link ProtectedRegion} for the specified world to database.
 *
 * @param world The name of the world
 * @param set The {@link Set} of regions
 * @throws StorageException Thrown if something goes wrong during database query
 */
public void saveAll(final String world, Set<ProtectedRegion> set) throws StorageException {
    MongoCollection<ProcessingProtectedRegion> collection = getCollection();
    final AtomicReference<Throwable> lastError = new AtomicReference<>();
    final CountDownLatch waiter = new CountDownLatch(set.size());
    for (final ProtectedRegion region : set) {
        if (listener != null)
            listener.beforeDatabaseUpdate(world, region);
        collection.findOneAndUpdate(
                Filters.and(Filters.eq("name", region.getId()), Filters.eq("world", world)),
                new Document("$set", new ProcessingProtectedRegion(region, world)),
                new FindOneAndUpdateOptions().upsert(true).returnDocument(ReturnDocument.AFTER),
                OperationResultCallback.create(lastError, waiter, new UpdateCallback(world))
        );
    }
    ConcurrentUtils.safeAwait(waiter);
    Throwable realLastError = lastError.get();
    if (realLastError != null)
        throw new StorageException("An error occurred while saving or updating in MongoDB.", realLastError);
}

@Override
public io.vertx.ext.mongo.MongoClient saveWithOptions(String collection, JsonObject document, @Nullable WriteOption writeOption, Handler<AsyncResult<String>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(document, "document cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  MongoCollection<JsonObject> coll = getCollection(collection, writeOption);
  Object id = document.getValue(ID_FIELD);
  if (id == null) {
    coll.insertOne(document, convertCallback(resultHandler, wr -> useObjectId ? document.getJsonObject(ID_FIELD).getString(JsonObjectCodec.OID_FIELD) : document.getString(ID_FIELD)));
  } else {
    JsonObject filter = new JsonObject();
    JsonObject encodedDocument = encodeKeyWhenUseObjectId(document);
    filter.put(ID_FIELD, encodedDocument.getValue(ID_FIELD));

    com.mongodb.client.model.UpdateOptions updateOptions = new com.mongodb.client.model.UpdateOptions()
        .upsert(true);

    coll.replaceOne(wrap(filter), encodedDocument, updateOptions, convertCallback(resultHandler, result -> null));
  }
  return this;
}

@Override
public io.vertx.ext.mongo.MongoClient insertWithOptions(String collection, JsonObject document, @Nullable WriteOption writeOption, Handler<AsyncResult<String>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(document, "document cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  JsonObject encodedDocument = encodeKeyWhenUseObjectId(document);
  boolean hasCustomId = document.containsKey(ID_FIELD);

  MongoCollection<JsonObject> coll = getCollection(collection, writeOption);
  coll.insertOne(encodedDocument, convertCallback(resultHandler, wr -> {
    if (hasCustomId) return null;

    JsonObject decodedDocument = decodeKeyWhenUseObjectId(encodedDocument);
    return decodedDocument.getString(ID_FIELD);
  }));
  return this;
}

@Override
public io.vertx.ext.mongo.MongoClient updateCollectionWithOptions(String collection, JsonObject query, JsonObject update, UpdateOptions options, Handler<AsyncResult<MongoClientUpdateResult>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(query, "query cannot be null");
  requireNonNull(update, "update cannot be null");
  requireNonNull(options, "options cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  update = generateIdIfNeeded(query, update, options);

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


  if (options.isMulti()) {
    coll.updateMany(bquery, bupdate, mongoUpdateOptions(options), toMongoClientUpdateResult(resultHandler));
  } else {
    coll.updateOne(bquery, bupdate, mongoUpdateOptions(options), toMongoClientUpdateResult(resultHandler));
  }
  return this;
}

@Override
public io.vertx.ext.mongo.MongoClient findOneAndUpdateWithOptions(String collection, JsonObject query, JsonObject update, FindOptions findOptions, UpdateOptions updateOptions, Handler<AsyncResult<JsonObject>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(query, "query cannot be null");
  requireNonNull(update, "update cannot be null");
  requireNonNull(findOptions, "find options cannot be null");
  requireNonNull(updateOptions, "update options cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  JsonObject encodedQuery = encodeKeyWhenUseObjectId(query);

  Bson bquery = wrap(encodedQuery);
  Bson bupdate = wrap(update);
  FindOneAndUpdateOptions foauOptions = new FindOneAndUpdateOptions();
  foauOptions.sort(wrap(findOptions.getSort()));
  foauOptions.projection(wrap(findOptions.getFields()));
  foauOptions.upsert(updateOptions.isUpsert());
  foauOptions.returnDocument(updateOptions.isReturningNewDocument() ? ReturnDocument.AFTER : ReturnDocument.BEFORE);

  MongoCollection<JsonObject> coll = getCollection(collection);
  coll.findOneAndUpdate(bquery, bupdate, foauOptions, wrapCallback(resultHandler));
  return this;
}

@Override
public io.vertx.ext.mongo.MongoClient findOneAndReplaceWithOptions(String collection, JsonObject query, JsonObject replace, FindOptions findOptions, UpdateOptions updateOptions, Handler<AsyncResult<JsonObject>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(query, "query cannot be null");
  requireNonNull(findOptions, "find options cannot be null");
  requireNonNull(updateOptions, "update options cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  JsonObject encodedQuery = encodeKeyWhenUseObjectId(query);

  Bson bquery = wrap(encodedQuery);
  FindOneAndReplaceOptions foarOptions = new FindOneAndReplaceOptions();
  foarOptions.sort(wrap(findOptions.getSort()));
  foarOptions.projection(wrap(findOptions.getFields()));
  foarOptions.upsert(updateOptions.isUpsert());
  foarOptions.returnDocument(updateOptions.isReturningNewDocument() ? ReturnDocument.AFTER : ReturnDocument.BEFORE);

  MongoCollection<JsonObject> coll = getCollection(collection);
  coll.findOneAndReplace(bquery, replace, foarOptions, wrapCallback(resultHandler));
  return this;
}

@Override
public io.vertx.ext.mongo.MongoClient findOneAndDeleteWithOptions(String collection, JsonObject query, FindOptions findOptions, Handler<AsyncResult<JsonObject>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(query, "query cannot be null");
  requireNonNull(findOptions, "find options cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  JsonObject encodedQuery = encodeKeyWhenUseObjectId(query);

  Bson bquery = wrap(encodedQuery);
  FindOneAndDeleteOptions foadOptions = new FindOneAndDeleteOptions();
  foadOptions.sort(wrap(findOptions.getSort()));
  foadOptions.projection(wrap(findOptions.getFields()));

  MongoCollection<JsonObject> coll = getCollection(collection);
  coll.findOneAndDelete(bquery, foadOptions, wrapCallback(resultHandler));
  return this;
}

private FindIterable<JsonObject> doFind(String collection, JsonObject query, FindOptions options) {
  MongoCollection<JsonObject> coll = getCollection(collection);
  Bson bquery = wrap(encodeKeyWhenUseObjectId(query));
  FindIterable<JsonObject> find = coll.find(bquery, JsonObject.class);
  if (options.getLimit() != -1) {
    find.limit(options.getLimit());
  }
  if (options.getSkip() > 0) {
    find.skip(options.getSkip());
  }
  if (options.getSort() != null) {
    find.sort(wrap(options.getSort()));
  }
  if (options.getFields() != null) {
    find.projection(wrap(options.getFields()));
  }
  return find;
}

private void loop() {
    final MongoDatabase database = mongoClient.getDatabase(this.database);
    final MongoCollection<Document> collection = database.getCollection(this.collection);
    final Bson filter = and(or(not(exists(RECEIVED_FIELD)), eq(RECEIVED_FIELD, false)), in("port", localInputs));
    final Document update = new Document("$set", new Document(RECEIVED_FIELD, true));
    final SingleResultCallback<Document> callback = (message, throwable) -> {
        if (message != null) {
            // TODO : keep it dry.
            final String portName = message.getString("port");
            final String payload = message.getString("payload");
            for (Port p : localInputs) {
                if (p.getPath().equals(portName)) {
                    p.send(payload);
                    break;
                }
            }
            // once treated the message is removed from the mongodb.
            collection.deleteOne(eq("_id", message.getObjectId("_id")), null);
            loop();
        }
    };
    collection.findOneAndUpdate(filter, update, callback);
}

@Override
public MongoClient replaceDocumentsWithOptions(String collection, JsonObject query, JsonObject replace, UpdateOptions options, Handler<AsyncResult<MongoClientUpdateResult>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(query, "query cannot be null");
  requireNonNull(replace, "update cannot be null");
  requireNonNull(options, "options cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  MongoCollection<JsonObject> coll = getCollection(collection, options.getWriteOption());
  Bson bquery = wrap(encodeKeyWhenUseObjectId(query));
  coll.replaceOne(bquery, encodeKeyWhenUseObjectId(replace), mongoUpdateOptions(options), toMongoClientUpdateResult(resultHandler));
  return this;
}

@Override
public io.vertx.ext.mongo.MongoClient count(String collection, JsonObject query, Handler<AsyncResult<Long>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(query, "query cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  Bson bquery = wrap(encodeKeyWhenUseObjectId(query));
  MongoCollection<JsonObject> coll = getCollection(collection);
  coll.count(bquery, wrapCallback(resultHandler));
  return this;
}

@Override
public MongoClient removeDocumentsWithOptions(String collection, JsonObject query, @Nullable WriteOption writeOption, Handler<AsyncResult<MongoClientDeleteResult>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(query, "query cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  MongoCollection<JsonObject> coll = getCollection(collection, writeOption);
  Bson bquery = wrap(encodeKeyWhenUseObjectId(query));
  coll.deleteMany(bquery, toMongoClientDeleteResult(resultHandler));
  return this;
}

@Override
public MongoClient removeDocumentWithOptions(String collection, JsonObject query, @Nullable WriteOption writeOption, Handler<AsyncResult<MongoClientDeleteResult>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(query, "query cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  MongoCollection<JsonObject> coll = getCollection(collection, writeOption);
  Bson bquery = wrap(encodeKeyWhenUseObjectId(query));
  coll.deleteOne(bquery, toMongoClientDeleteResult(resultHandler));
  return this;
}

@Override
public MongoClient bulkWriteWithOptions(String collection, List<BulkOperation> operations, BulkWriteOptions bulkWriteOptions, Handler<AsyncResult<MongoClientBulkWriteResult>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(operations, "operations cannot be null");
  requireNonNull(bulkWriteOptions, "bulkWriteOptions cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");
  MongoCollection<JsonObject> coll = getCollection(collection, bulkWriteOptions.getWriteOption());
  List<WriteModel<JsonObject>> bulkOperations = convertBulkOperations(operations);
  coll.bulkWrite(bulkOperations, mongoBulkWriteOptions(bulkWriteOptions),
      toMongoClientBulkWriteResult(resultHandler));

  return this;
}

@Override
public io.vertx.ext.mongo.MongoClient dropCollection(String collection, Handler<AsyncResult<Void>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");

  MongoCollection<JsonObject> coll = getCollection(collection);
  coll.drop(wrapCallback(resultHandler));
  return this;
}

@Override
public io.vertx.ext.mongo.MongoClient createIndexWithOptions(String collection, JsonObject key, IndexOptions options, Handler<AsyncResult<Void>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(key, "fieldName cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");
  MongoCollection<JsonObject> coll = getCollection(collection);
  com.mongodb.client.model.IndexOptions driverOpts = new com.mongodb.client.model.IndexOptions()
          .background(options.isBackground())
          .unique(options.isUnique())
          .name(options.getName())
          .sparse(options.isSparse())
          .expireAfter(options.getExpireAfter(TimeUnit.SECONDS), TimeUnit.SECONDS)
          .version(options.getVersion())
          .weights(toBson(options.getWeights()))
          .defaultLanguage(options.getDefaultLanguage())
          .languageOverride(options.getLanguageOverride())
          .textVersion(options.getTextVersion())
          .sphereVersion(options.getSphereVersion())
          .bits(options.getBits())
          .min(options.getMin())
          .max(options.getMax())
          .bucketSize(options.getBucketSize())
          .storageEngine(toBson(options.getStorageEngine()))
          .partialFilterExpression(toBson(options.getPartialFilterExpression()));
  coll.createIndex(wrap(key), driverOpts, wrapCallback(toVoidAsyncResult(resultHandler)));
  return this;
}

@Override
public io.vertx.ext.mongo.MongoClient listIndexes(String collection, Handler<AsyncResult<JsonArray>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");
  MongoCollection<JsonObject> coll = getCollection(collection);
  ListIndexesIterable indexes = coll.listIndexes(JsonObject.class);
  if (indexes != null) {
    convertMongoIterable(indexes, resultHandler);
  }
  return this;
}

@Override
public MongoClient dropIndex(String collection, String indexName, Handler<AsyncResult<Void>> resultHandler) {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(indexName, "indexName cannot be null");
  requireNonNull(resultHandler, "resultHandler cannot be null");
  MongoCollection<JsonObject> coll = getCollection(collection);
  coll.dropIndex(indexName, wrapCallback(resultHandler));
  return this;
}

private DistinctIterable<?> findDistinctValuesWithQuery(String collection, String fieldName, String resultClassname, JsonObject query) throws ClassNotFoundException {
  requireNonNull(collection, "collection cannot be null");
  requireNonNull(fieldName, "fieldName cannot be null");
  requireNonNull(query, "query cannot be null");

  JsonObject encodedQuery = encodeKeyWhenUseObjectId(query);

  Bson bquery = wrap(encodedQuery);

  MongoCollection<JsonObject> mongoCollection = getCollection(collection);
  Class<?> resultClass = this.getClass().getClassLoader().loadClass(resultClassname);
  return mongoCollection.distinct(fieldName, bquery, resultClass);
}

private MongoCollection<JsonObject> getCollection(String name, @Nullable WriteOption writeOption) {
  MongoCollection<JsonObject> coll = holder.db.getCollection(name, JsonObject.class);
  if (coll != null && writeOption != null) {
    coll = coll.withWriteConcern(WriteConcern.valueOf(writeOption.name()));
  }
  return coll;
}

private MongoCollection<ProcessingProtectedRegion> getCollection() {
    return database.getCollection(COLLECTION_NAME, ProcessingProtectedRegion.class);
}

private MongoCollection<JsonObject> getCollection(String name) {
  return getCollection(name, null);
}

@Override
public void execute(MongoDatabase db, String collection,
        final CountDownLatch latch) {
    Validate.notNull(collection);
    final List<String> nameArray = new ArrayList<String>();

    nameArray.add(db.getName());
    nameArray.add(collection);

    MongoCollection<Document> mongoCollection = db
            .getCollection(collection);
    MongoIterable<Document> documentList = mongoCollection.find(queryWhere)
            .projection(queryFields);

    emptyResult = true;
    documentList.forEach(new Block<Document>() {
        @Override
        public void apply(Document document) {
            for (Entry<String, Object> entry : document.entrySet()) {
                String name = StringUtils.join(nameArray, "_") + "_"
                        + entry.getKey();
                Object value = entry.getValue();

                // set status message and perfdata
                checkResult.setStatusMoreSevere(compare(value));
                checkResult.addMessage(name + "=" + value);
                addPerformanceData(name, value);
                emptyResult = false;
            }
        }
    }, new SingleResultCallback<Void>() {
        @Override
        public void onResult(Void arg0, Throwable t) {
            if (t != null) {
                checkResult
                        .setStatusMoreSevere(NagiosServiceStatus.WARNING);
                checkResult.addMessage(t.getMessage());
            } else if (emptyResult) {
                checkResult
                        .setStatusMoreSevere(NagiosServiceStatus.WARNING);
                checkResult.addMessage(String.format(
                        "%s.find(%s) did not return a result",
                        StringUtils.join(nameArray, "."), getQuery()));
            }
            latch.countDown();
        }
    });
}

/**
 * Returns the {@code MongoCollection} which contains the oplog.
 *
 * @param client The {@link MongoClient}
 * @return The {@code MongoCollection} which contains the oplog
 */
public static MongoCollection<BsonDocument> getCollection(MongoClient client) {
    return client.getDatabase(OPLOG_DATABASE).getCollection(OPLOG_COLLECTION, BsonDocument.class);
}

/**
 * Constructor.
 *
 * @param oplog The oplog collection
 * @param parser An instance of {@link OpLogParser}
 * @param namespace The namespace for which should be listened
 */
public OpLogRetriever(MongoCollection<BsonDocument> oplog, OpLogParser parser, String namespace) {
    this.oplog = Preconditions.checkNotNull(oplog, "oplog must be not null.");
    this.parser = Preconditions.checkNotNull(parser, "parser must be not null.");
    this.namespace = Preconditions.checkNotNull(namespace, "namespace must be not null.");
}