Java 类com.mongodb.connection.ClusterSettings 实例源码

@Override
public CompletableFuture<SecurityDefinitionStore> open() {
  List<ServerAddress> hostList =
      Arrays.stream(hosts).map(h -> new ServerAddress(h)).collect(Collectors.toList());
  ClusterSettings clusterSettings = ClusterSettings.builder().hosts(hostList).build();
  MongoClientSettings settings =
      MongoClientSettings.builder().clusterSettings(clusterSettings).build();
  mongoClient = MongoClients.create(settings);

  database = mongoClient.getDatabase(DATABASE_NAME);
  collection = database.getCollection(SECDEF_COLLECTION_NAME);

  // In the case of MongoDB, open is synchronous because it doesn't
  // actually communicate with the server until a query is invoked.
  return CompletableFuture.completedFuture(this);
}

private List<ServerAddress> extractServerAddresses(MongoClient client) {
    Cluster cluster = (Cluster) ReflectionTestUtils.getField(client, "cluster");
    ClusterSettings clusterSettings = (ClusterSettings) ReflectionTestUtils
            .getField(cluster, "settings");
    List<ServerAddress> allAddresses = clusterSettings.getHosts();
    return allAddresses;
}

/**
 * Create an asynchronous MongoDB client
 * @return the client
 */
private com.mongodb.async.client.MongoClient createAsyncClient() {
  ClusterSettings clusterSettings = ClusterSettings.builder()
      .hosts(Arrays.asList(mongoConnector.serverAddress))
      .build();
  MongoClientSettings settings = MongoClientSettings.builder()
      .clusterSettings(clusterSettings).build();
  return MongoClients.create(settings);
}

private List<ServerAddress> extractServerAddresses(MongoClient client) {
    Cluster cluster = (Cluster) ReflectionTestUtils.getField(client, "cluster");
    ClusterSettings clusterSettings = (ClusterSettings) ReflectionTestUtils
            .getField(cluster, "settings");
    List<ServerAddress> allAddresses = clusterSettings.getHosts();
    return allAddresses;
}

/**
 * {@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 ClusterSettingsParser(ConnectionString connectionString, JsonObject config) {
  ClusterSettings.Builder settings = ClusterSettings.builder();
  // ConnectionString takes precedence
  if (connectionString != null) {
    settings.applyConnectionString(connectionString);
  } else {
    // hosts
    List<ServerAddress> hosts = parseHosts(config);
    settings.hosts(hosts);

    // replica set / mode
    String replicaSet = config.getString("replicaSet");
    if (hosts.size() == 1 && replicaSet == null) {
      settings.mode(ClusterConnectionMode.SINGLE);
    } else {
      settings.mode(ClusterConnectionMode.MULTIPLE);
    }
    if (replicaSet != null) {
      settings.requiredReplicaSetName(replicaSet);
    }

    // serverSelectionTimeoutMS
    Long serverSelectionTimeoutMS = config.getLong("serverSelectionTimeoutMS");
    if(serverSelectionTimeoutMS != null) {
      settings.serverSelectionTimeout(serverSelectionTimeoutMS, MILLISECONDS);
    }

    Integer waitQueueMultiple = config.getInteger("waitQueueMultiple");
    if (waitQueueMultiple != null) {
      Integer waitQueueSize = waitQueueMultiple * DEFAULT_MONGO_DRIVER_WAIT_Q_SIZE;
      settings.maxWaitQueueSize(waitQueueSize);
    }
  }

  this.settings = settings.build();
}

private static void assertSingleHost(ClusterConnectionMode mode, ClusterSettings settings) {
  List<ServerAddress> hosts = settings.getHosts();
  assertNotNull(hosts);
  assertEquals(1, hosts.size());
  assertEquals(new ServerAddress("single.host", 1111), hosts.get(0));
  assertEquals(mode, settings.getMode());
}

private static void assertMultipleHosts(ClusterSettings settings) {
  List<ServerAddress> hosts = settings.getHosts();
  assertNotNull(hosts);
  assertEquals(2, hosts.size());
  assertEquals(new ServerAddress("multiple.1", 2222), hosts.get(0));
  assertEquals(new ServerAddress("multiple.2", 3333), hosts.get(1));
  assertEquals(ClusterConnectionMode.MULTIPLE, settings.getMode());
}

private Block cluster(final String db) {
  return unit -> {
    ClusterSettings settings = unit.mock(ClusterSettings.class);
    unit.registerMock(ClusterSettings.class, settings);

    ClusterSettings.Builder builder = unit.mock(ClusterSettings.Builder.class);
    expect(builder.applyConnectionString(new ConnectionString(db))).andReturn(builder);
    expect(builder.build()).andReturn(settings);

    unit.mockStatic(ClusterSettings.class);
    expect(ClusterSettings.builder()).andReturn(builder);
  };
}

@Start
public void start() {
    final ClusterSettings clusterSettings = ClusterSettings.builder().hosts(asList(new ServerAddress(host, port))).build();
    final MongoClientSettings settings = MongoClientSettings.builder().clusterSettings(clusterSettings).build();
    mongoClient = MongoClients.create(settings);
    db = mongoClient.getDatabase(this.database);
    launchConsumers();
}

/**
 * Lazily instantiate the {@link MongoClient} instance.
 *
 * @return
 */
private MongoClient createMongoClient() {
  String host = applicationConfiguration.getMongoHost();
  int port = applicationConfiguration.getMongoPort();
  ConnectionString connectionString = new ConnectionString("mongodb://" + host + ":" + port);

  logger.info("Creating Mongo client for: {}:{}", host, port);

  MongoClientSettings mongoClientSettings =
      MongoClientSettings.builder()
          .applicationName("dragoman")
          .serverSettings(
              ServerSettings.builder()
                  .applyConnectionString(connectionString)
                  .addServerMonitorListener(new LoggingServerMonitorListener())
                  .addServerListener(new LoggingServerListener())
                  .build())
          .clusterSettings(
              ClusterSettings.builder()
                  .applyConnectionString(connectionString)
                  .serverSelectionTimeout(
                      applicationConfiguration.getMongoServerSelectionTimeout(), MILLISECONDS)
                  .addClusterListener(new LoggingClusterListener())
                  .build())
          .connectionPoolSettings(
              ConnectionPoolSettings.builder()
                  .applyConnectionString(connectionString)
                  .maxWaitTime(
                      applicationConfiguration.getConnectionPoolMaxWaitTime(), MILLISECONDS)
                  .minSize(applicationConfiguration.getConnectionPoolMinSize())
                  .maxSize(applicationConfiguration.getConnectionPoolMaxSize())
                  .addConnectionPoolListener(new LoggingConnectionPoolListener())
                  .build())
          .socketSettings(
              SocketSettings.builder()
                  .applyConnectionString(connectionString)
                  .connectTimeout(
                      applicationConfiguration.getMongoSocketConnectionTimeout(), MILLISECONDS)
                  .readTimeout(applicationConfiguration.getMongoReadTimeout(), MILLISECONDS)
                  .build())
          .build();

  return MongoClients.create(mongoClientSettings);
}

private MongoClientSettings mongoSettings(String host, int port) {
  return MongoClientSettings.builder()
      .clusterSettings(
          ClusterSettings.builder().hosts(newArrayList(new ServerAddress(host, port))).build())
      .build();
}

public ClusterSettings settings() {
  return settings;
}

@Test
public void testReplicaSetSingleHost() {
  ClusterSettings settings = settings(singleHost().put("replicaSet", "foo"));
  assertSingleHost(ClusterConnectionMode.MULTIPLE, settings);
  assertEquals("foo", settings.getRequiredReplicaSetName());
}

@Test
public void testReplicaSetMultipleHosts() {
  ClusterSettings settings = settings(multipleHosts().put("replicaSet", "foobar"));
  assertMultipleHosts(settings);
  assertEquals("foobar", settings.getRequiredReplicaSetName());
}

@Test
public void testServerSelectionTimeoutMS() {
  ClusterSettings settings = settings(multipleHosts().put("serverSelectionTimeoutMS", 7533L));
  assertMultipleHosts(settings);
  assertEquals(7533L, settings.getServerSelectionTimeout(TimeUnit.MILLISECONDS));
}

@Test
public void testMaxWaitQueueSize() {
  ClusterSettings settings = settings(multipleHosts().put("waitQueueMultiple", 10));
  assertEquals(ClusterSettingsParser.DEFAULT_MONGO_DRIVER_WAIT_Q_SIZE * 10, settings.getMaxWaitQueueSize());
}

private static ClusterSettings settings(JsonObject config) {
  return new ClusterSettingsParser(null, config).settings();
}

private void connect() throws ActivecheckReporterException {
    if (mongoClient == null) {
        logger.debug("Cannot run query. MongoDB is not connected. Trying to (re)connect.");
        try {
            // configure credentials
            List<MongoCredential> credentialsList = new ArrayList<MongoCredential>();
            String username = properties
                    .getString("mongodb.username", null);
            String password = properties
                    .getString("mongodb.password", null);
            if (username != null && password != null) {
                credentialsList.add(MongoCredential.createPlainCredential(
                        username, "*", password.toCharArray()));
            }

            // configure server addresses
            List<ServerAddress> addressList = new ArrayList<ServerAddress>();
            String socketPath = properties.getString("socket", null);
            if (socketPath != null) {
                addressList.add(new ServerAddress(new AFUNIXSocketAddress(
                        new File(socketPath))));
            } else {
                String url = properties.getString("url",
                        ServerAddress.defaultHost());
                int port = ServerAddress.defaultPort();
                String[] urlParts = url.split(":");
                if (urlParts.length > 1) {
                    port = Integer.parseInt(urlParts[1]);
                }
                addressList.add(new ServerAddress(urlParts[0], port));
            }
            ServerSelector serverSelector = new ReadPreferenceServerSelector(
                    MONGO_READ_PREFERENCE);
            ClusterSettings clusterSettings = ClusterSettings.builder()
                    .hosts(addressList).serverSelector(serverSelector)
                    .build();

            // actually configure and (re)create mongoClient
            ConnectionPoolSettings connectionPoolSettings = ConnectionPoolSettings
                    .builder().maxSize(MONGO_POOL_SIZE).build();
            MongoClientSettings settings = MongoClientSettings.builder()
                    .readPreference(MONGO_READ_PREFERENCE)
                    .credentialList(credentialsList)
                    .clusterSettings(clusterSettings)
                    .connectionPoolSettings(connectionPoolSettings).build();
            mongoClient = MongoClients.create(settings);
        } catch (Exception e) {
            mongoClient = null;
            String errorMessage = "MongodbReporter Configuration Error for service '"
                    + getOverallServiceName() + "': " + e.getMessage();

            logger.error(errorMessage);
            logger.trace(e.getMessage(), e);

            // set report and status
            setOverallServiceReport(NagiosServiceStatus.CRITICAL,
                    errorMessage);
            throw new ActivecheckReporterException(e);
        }
    }
}