@Override
public void apply(ClusterSettings.Builder builder) {
    Optional<String> maybeConnectionString = config.connectionString;
    if (!maybeConnectionString.isPresent()) {
        // Parse hosts
        List<ServerAddress> hosts = parseHosts(config.hosts);
        builder.hosts(hosts);

        if (hosts.size() == 1 && !config.replicaSetName.isPresent()) {
            builder.mode(ClusterConnectionMode.SINGLE);
        } else {
            builder.mode(ClusterConnectionMode.MULTIPLE);
        }
    }
    if (config.localThreshold.isPresent()) {
        builder.localThreshold(config.localThreshold.get().toMillis(), TimeUnit.MILLISECONDS);
    }

    config.replicaSetName.ifPresent(builder::requiredReplicaSetName);

    if (config.serverSelectionTimeout.isPresent()) {
        builder.serverSelectionTimeout(config.serverSelectionTimeout.get().toMillis(), TimeUnit.MILLISECONDS);
    }
}
 

@Override
public void afterPropertiesSet() throws Exception {

    final IMongodConfig mongodConfig = new MongodConfigBuilder()
            .version(Version.Main.PRODUCTION).build();

    IRuntimeConfig runtimeConfig = new RuntimeConfigBuilder()
            .defaultsWithLogger(Command.MongoD, logger)
            .processOutput(ProcessOutput.getDefaultInstanceSilent())
            .build();

    MongodStarter runtime = MongodStarter.getInstance(runtimeConfig);

    MongodExecutable mongodExecutable = runtime.prepare(mongodConfig);
    mongod = mongodExecutable.start();

    // cluster configuration
    ClusterSettings clusterSettings = ClusterSettings.builder().hosts(Collections.singletonList(new ServerAddress(mongodConfig.net().getServerAddress().getHostName(), mongodConfig.net().getPort()))).build();
    // codec configuration
    CodecRegistry pojoCodecRegistry = fromRegistries(MongoClients.getDefaultCodecRegistry(),
            fromProviders(PojoCodecProvider.builder().automatic(true).build()));

    MongoClientSettings settings = MongoClientSettings.builder().clusterSettings(clusterSettings).codecRegistry(pojoCodecRegistry).writeConcern(WriteConcern.ACKNOWLEDGED).build();
    mongoClient = MongoClients.create(settings);
    mongoDatabase = mongoClient.getDatabase(databaseName);
}
 

@Override
public void afterPropertiesSet() throws Exception {
    final IMongodConfig mongodConfig = new MongodConfigBuilder()
            .version(Version.Main.PRODUCTION).build();

    IRuntimeConfig runtimeConfig = new RuntimeConfigBuilder()
            .defaultsWithLogger(Command.MongoD, logger)
            .processOutput(ProcessOutput.getDefaultInstanceSilent())
            .build();

    MongodStarter runtime = MongodStarter.getInstance(runtimeConfig);

    MongodExecutable mongodExecutable = runtime.prepare(mongodConfig);
    mongod = mongodExecutable.start();

    // cluster configuration
    ClusterSettings clusterSettings = ClusterSettings.builder().hosts(Collections.singletonList(new ServerAddress(mongodConfig.net().getServerAddress().getHostName(), mongodConfig.net().getPort()))).build();
    // codec configuration
    CodecRegistry pojoCodecRegistry = fromRegistries(MongoClients.getDefaultCodecRegistry(),
            fromProviders(PojoCodecProvider.builder().automatic(true).build()));

    MongoClientSettings settings = MongoClientSettings.builder().clusterSettings(clusterSettings).codecRegistry(pojoCodecRegistry).writeConcern(WriteConcern.ACKNOWLEDGED).build();
    mongoClient = MongoClients.create(settings);
    mongoDatabase = mongoClient.getDatabase(databaseName);
}
 

@Bean
public MongoClient mongoClient() throws IOException {
  ServerAddress serverAddress = getServerAddress();
  MongoClientSettings settings = MongoClientSettings.builder()
                                                    .clusterSettings(ClusterSettings.builder()
                                                                                    .hosts(singletonList(serverAddress))
                                                                                    .requiredClusterType(STANDALONE)
                                                                                    .build()).build();
  return MongoClients.create(settings);
}
 

private MongoDatabase createDatabase(CodecRegistry registry) {
    if (uri == null) throw new Error("uri must not be null");
    String database = uri.getDatabase();
    if (database == null) throw new Error("uri must have database, uri=" + uri);
    var watch = new StopWatch();
    try {
        connectionPoolSettings.maxWaitTime(timeoutInMs, TimeUnit.MILLISECONDS); // pool checkout timeout
        var socketSettings = SocketSettings.builder()
                                           .connectTimeout((int) timeoutInMs, TimeUnit.MILLISECONDS)
                                           .readTimeout((int) timeoutInMs, TimeUnit.MILLISECONDS)
                                           .build();
        var clusterSettings = ClusterSettings.builder()
                                             .serverSelectionTimeout(timeoutInMs * 3, TimeUnit.MILLISECONDS)    // able to try 3 servers
                                             .build();
        var settings = MongoClientSettings.builder()
                                          .applicationName(LogManager.APP_NAME)
                                          .codecRegistry(registry)
                                          .applyToConnectionPoolSettings(builder -> builder.applySettings(connectionPoolSettings.build()))
                                          .applyToSocketSettings(builder -> builder.applySettings(socketSettings))
                                          .applyToClusterSettings(builder -> builder.applySettings(clusterSettings))
                                          .applyConnectionString(uri)
                                          .build();
        mongoClient = MongoClients.create(settings);
        return mongoClient.getDatabase(database);
    } finally {
        logger.info("create mongo client, uri={}, elapsed={}", uri, watch.elapsed());
    }
}
 

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

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

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

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