import com.mongodb.rx.client.MongoClients; //导入依赖的package包/类
public SimpleMongoAsyncTest() throws IOException {
    // Start embeded mongo
    config = new MongodConfigBuilder()
            .version(Version.Main.PRODUCTION)
            .net(new Net(27018, Network.localhostIsIPv6()))
            .build();
    exe = MongodStarter.getDefaultInstance().prepare(config);
    mongoProcess = exe.start();

    AtomicBoolean started = new AtomicBoolean(false);

    // Start mongo client
    client = MongoClients.create("mongodb://localhost:27018");
    database = client.getDatabase("test");
    collection = database.getCollection(COLLECTION_NAME);

    morphia = new Morphia().mapPackage("com.querydsl.mongodb.domain");
}
 

import com.mongodb.rx.client.MongoClients; //导入依赖的package包/类
@Test
public void populate() {
  int simulationCount = 1;

  MongoClient mongoClient = MongoClients.create("mongodb://localhost:27017");

  MongoDatabase database = mongoClient.getDatabase(databaseName);
  MongoCollection<Document> collection = database.getCollection(sample);

  for (int i = 0; i < simulationCount; i++) {
    Document d = new Document("name", "Person " + i).append("updatedAt", new Date());
    collection.insertOne(d).toList().toBlocking().single();
  }
}
 

import com.mongodb.rx.client.MongoClients; //导入依赖的package包/类
public MongoSongStore() {
    String connectionString = System.getProperty(CONNECTION_STRING_PROPERTY, "mongodb://localhost");
    String databaseName = System.getProperty(DATABASE_NAME_PROPERTY, "default");
    String collectionName = System.getProperty(COLLECTION_NAME_PROPERTY, "default");

    mongoClient = MongoClients.create(new ConnectionString(connectionString));
    mongoDatabase = mongoClient.getDatabase(databaseName);
    mongoCollection = mongoDatabase.getCollection(collectionName);

    logger.info("Initialized Mongo Driver with databaseName=" + databaseName + " collectionName=" + collectionName);

}
 

import com.mongodb.rx.client.MongoClients; //导入依赖的package包/类
@Override
public Observable<Document> startImport() {
	MongoClient client = MongoClients.create(connectionString);
	MongoDatabase db = client.getDatabase(dbName);
	return db.getCollection(collectionName).find().toObservable()
			.map(new Func1<org.bson.Document, Document>() {
				public Document call(org.bson.Document mongoDoc) {
					mongoDoc.put(typeField, type);
					RawJsonDocument d = RawJsonDocument.create(mongoDoc
							.getObjectId("_id").toHexString(), mongoDoc
							.toJson());
					return d;
				};
			});
}
 

import com.mongodb.rx.client.MongoClients; //导入依赖的package包/类
/**
 * 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);
}
 

import com.mongodb.rx.client.MongoClients; //导入依赖的package包/类
protected MongoClient getMongoClient() {
  if (mongoClient == null) {
    mongoClient = MongoClients.create("mongodb://localhost:" + port);
  }
  return mongoClient;
}
 

import com.mongodb.rx.client.MongoClients; //导入依赖的package包/类
@SuppressWarnings({"rawtypes", "unchecked"})
@Override
public void configure(final Env env, final Config conf, final Binder binder) {
  /** connection string */
  ConnectionString cstr = Try.apply(() -> new ConnectionString(db))
      .orElseGet(() -> new ConnectionString(conf.getString(db)));

  log.debug("Starting {}", cstr);

  boolean first = instances.getAndIncrement() == 0;
  Throwing.Function3<Class, String, Object, Void> bind = (type, name, value) -> {
    binder.bind(Key.get(type, Names.named(name))).toInstance(value);
    if (first) {
      binder.bind(Key.get(type)).toInstance(value);
    }
    return null;
  };

  /** settings */
  MongoClientSettings.Builder settings = settings(cstr, dbconf(db, conf));
  if (configurer != null) {
    configurer.accept(settings, conf);
  }
  MongoClient client = MongoClients.create(settings.build());
  bind.apply(MongoClient.class, db, client);

  /** bind database */
  Optional.ofNullable(cstr.getDatabase()).ifPresent(dbname -> {
    // observable adapter
    MongoDatabase predb = adapter
        .map(a -> client.getDatabase(dbname).withObservableAdapter(a))
        .orElseGet(() -> client.getDatabase(dbname));
    // codec registry
    MongoDatabase database = codecRegistry
        .map(predb::withCodecRegistry)
        .orElse(predb);

    bind.apply(MongoDatabase.class, dbname, database);

    /** bind collection */
    Optional.ofNullable(cstr.getCollection()).ifPresent(cname -> {
      MongoCollection<Document> collection = database.getCollection(cname);
      bind.apply(MongoCollection.class, cname, collection);
    });
  });

  /** mapper */
  env.router()
      .map(mapper());

  log.info("Started {}", cstr);

  env.onStop(() -> {
    log.debug("Stopping {}", cstr);
    client.close();
    log.info("Stopped {}", cstr);
  });
}