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

private boolean testConnection(MongoDatabase database) {
    CountDownLatch waiter = new CountDownLatch(1);
    final AtomicReference<Throwable> error = new AtomicReference<>();
    boolean erroneous = false;
    try {
        database.getCollection(RegionStorageAdapter.COLLECTION_NAME).count(new OperationResultCallback<Long>(error, waiter));
        waiter.await();
        Throwable realError = error.get();
        if (realError != null)
            throw realError;
    } catch (MongoTimeoutException ignore) {
        getLogger().severe("Cannot connect to MongoDB server.");
        erroneous = true;
    } catch (Throwable throwable) {
        getLogger().log(Level.SEVERE, "An error occurred while connecting to database.", throwable);
        erroneous = true;
    }

    if (erroneous) {
        getLogger().severe("An error was encountered. Disabling plugin and NOT injecting into WorldGuard.");
        getServer().getPluginManager().disablePlugin(this);
        return false;
    }

    return true;
}

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);
}

/**
 * DataWriter constructor
 *
 * @throws IllegalStateException if database configuration is not set
 */
private DatabaseWriter() {
    try {
        final MongoClient mongoClient = MongoClients.create(PROPERTIES_MANAGER.getProperty("database.host"));
        final MongoDatabase mongoDatabase = mongoClient.getDatabase(PROPERTIES_MANAGER.getProperty("database.datasource"));
        this.mongoCollection = mongoDatabase.getCollection(PROPERTIES_MANAGER.getProperty("database.collection"));
    } catch (IllegalArgumentException e) {
        LOGGER.error(e.getMessage());
        throw new IllegalStateException(e.getMessage());
    }
    LOGGER.info("DatabaseWriter has been instantiate");
}

/**
 * Get or create the MongoDB database
 * @return the MongoDB client
 */
private MongoDatabase getDB() {
  if (database == null) {
    database = getMongoClient().getDatabase(databaseName);
  }
  return database;
}

/**
 * {@inheritDoc}
 */
@Override
public void onEnable() {
    saveDefaultConfig();

    CodecRegistry codecRegistry = createCodecRegistry();
    MongoClientSettings settings = MongoClientSettings.builder()
            .clusterSettings(ClusterSettings.builder().applyConnectionString(new ConnectionString(getConfig().getString("mongodb.uri"))).build())
            .codecRegistry(codecRegistry)
            .build();
    client = MongoClients.create(settings);
    MongoDatabase database = client.getDatabase(getConfig().getString("mongodb.database"));
    if (!testConnection(database))
        return;
    RegionStorageAdapter storageAdapter = new RegionStorageAdapter(database);
    MongoRegionDriver driver = new MongoRegionDriver(getServer(), storageAdapter);

    WorldGuardPlugin wgPlugin = WorldGuardPlugin.inst();
    if (getConfig().getBoolean("mongodb.use_oplog")) {
        getLogger().info("OpLog usage enabled.");
        WorldGuardOpLogHandler opLogHandler = new WorldGuardOpLogHandler(codecRegistry.get(ProcessingProtectedRegion.class), storageAdapter, wgPlugin);
        getServer().getScheduler().runTaskAsynchronously(this, new OpLogRetriever(
                OpLogUtils.getCollection(client),
                new OpLogParser(opLogHandler),
                getConfig().getString("mongodb.database") + "." + RegionStorageAdapter.COLLECTION_NAME
        ));
        storageAdapter.setListener(opLogHandler);
    }

    ConfigurationManager config = wgPlugin.getGlobalStateManager();
    RegionContainer container = wgPlugin.getRegionContainer();
    InjectionUtils.injectRegionDriver(container, driver);
    InjectionUtils.callUnload(container);
    InjectionUtils.callLoadWorlds(container);
    config.selectedRegionStoreDriver = driver;
}

public abstract void execute(MongoDatabase db, String collection,
final CountDownLatch latch);

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

    // define callback for handling the result
    SingleResultCallback<Document> callback = new SingleResultCallback<Document>() {
        @Override
        public void onResult(Document document, Throwable t) {
            if (t != null) {
                checkResult
                        .setStatusMoreSevere(NagiosServiceStatus.WARNING);
                checkResult.addMessage(t.getMessage());
            } else if (document == null) {
                checkResult
                        .setStatusMoreSevere(NagiosServiceStatus.WARNING);
                checkResult.addMessage("stats did not return a result");
            } else {
                for (String metric : stats) {
                    String name = StringUtils.join(nameArray, ".") + "."
                            + metric;
                    Object value = document.get(metric);
                    if (value != null) {
                        // set status message and perfdata
                        // TODO: move to compare function?
                        checkResult.setStatusMoreSevere(compare(value));
                        checkResult.addMessage(name + "=" + value);
                        addPerformanceData(name, value);
                    } else {
                        checkResult
                                .setStatusMoreSevere(NagiosServiceStatus.WARNING);
                        checkResult.addMessage(name + " does not exist");
                    }
                }
            }
            latch.countDown();
        }
    };

    // retrieve stats document
    if (collection == null || collection.isEmpty()) {
        db.runCommand(new Document("dbStats", 1), callback);
    } else {
        nameArray.add(collection);
        db.runCommand(new Document("collStats", collection), callback);
    }
}

@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();
        }
    });
}

/**
 * Constructor.
 *
 * @param database The {@link MongoDatabase} to use
 */
public RegionStorageAdapter(MongoDatabase database) {
    this.database = Preconditions.checkNotNull(database, "database must be not null.");
}