Java 类com.mongodb.MongoClientURI 实例源码

@Override
public Connection connect(String url, Properties info) throws SQLException {
    MongoClientURI mcu = null;
    if ((mcu = parseURL(url, info)) == null) {
        return null;
    }

    MongoConnection result = null;
    //System.out.print(info);
    try{
        result = new MongoConnection(mcu, url);
    }catch (Exception e){
        throw new SQLException("Unexpected exception: " + e.getMessage(), e);
    }

    return result;
}

private MongoClientURI parseURL(String url, Properties defaults) {
    if (url == null) {
        return null;
    }

    if (!StringUtils.startsWithIgnoreCase(url, PREFIX)) {   
        return null;
    }

    //删掉开头的 jdbc:
    //url = url.replace(URL_JDBC, "");

    try {
        //FIXME 判断defaults中的参数,写入URL中?
        return new MongoClientURI(url);
    } catch (Exception e) {
        LOGGER.error("parseURLError",e);
        return null;
    }

}

public void Connect(String user, String pass) {
    try {
        logger.info("Connecting to " + connectionString);

        // Authaenticate
        // MongoCredential credential =
        // MongoCredential.createCredential(user,
        // "admin",
        // pass); //Only users on admin as that will be mandatory in 3.6

        mongoClient = new MongoClient(new MongoClientURI(connectionString));

        mongoClient.getDatabase("admin")
                .runCommand(new Document("ping", 1));

    } catch (Exception e) {
        logger.error("Unable to connect to MongoDB");
        logger.error(e.getMessage());
        System.exit(1);
    }
    this.user = user;
    this.pass = pass;
}

public BingoChessChallenge(String[] args) {
    announcer = new Chatter(args[0]);
    lichs = new HashMap<String,Lichesser>();
    chessplayers = new HashMap<String,ChessPlayer>();
    chessgames = new HashMap<String,LichessGame>();
    BingoPlayer.SQUARE_BAG = new Vector<Dimension>();
    for (int x=0;x<8;x++)
    for (int y=0;y<8;y++)
    BingoPlayer.SQUARE_BAG.add(new Dimension(x,y));
    initIRC(args[0], args[1], args[2], args[3]);
    loadAdmins("res/admins.txt");
    serv = new BingoServ(Integer.parseInt(args[4]),this);
    serv.startSrv();
    bingoURL = args[5];
    MongoClientURI connStr = new MongoClientURI("mongodb://bingobot:" + args[6] + "@localhost:27017/BingoBase");
    MongoClient mongoClient = new MongoClient(connStr);
    MongoDatabase bingoBase = mongoClient.getDatabase("BingoBase");
    playData = bingoBase.getCollection("players");
}

public BingoChess(String[] args) {
    tv_client = new GameClient();
    twits = new HashMap<String,Chatter>();
    bingoers = new HashMap<String,BingoPlayer>();
    BingoPlayer.SQUARE_BAG = new Vector<Dimension>();
    for (int x=0;x<8;x++)
    for (int y=0;y<8;y++)
    BingoPlayer.SQUARE_BAG.add(new Dimension(x,y));
    initIRC(args[0], args[1], args[2], args[3]);
    loadAdmins("res/admins.txt");
    serv = new BingoServ(Integer.parseInt(args[4]),this);
    serv.startSrv();
    bingoURL = args[5];
    followTVGame();
    MongoClientURI connStr = new MongoClientURI("mongodb://bingobot:" + args[6] + "@localhost:27017/BingoBase");
    mongoClient = new MongoClient(connStr);
    bingoBase = mongoClient.getDatabase("BingoBase");
    playData = bingoBase.getCollection("players");
}

public BingoChessChallenge(String[] args) {
    announcer = new Chatter(args[0]);
    lichs = new HashMap<String,Lichesser>();
    chessplayers = new HashMap<String,ChessPlayer>();
    chessgames = new HashMap<String,LichessGame>();
    BingoPlayer.SQUARE_BAG = new Vector<Dimension>();
    for (int x=0;x<8;x++)
    for (int y=0;y<8;y++)
    BingoPlayer.SQUARE_BAG.add(new Dimension(x,y));
    initIRC(args[0], args[1], args[2], args[3]);
    loadAdmins("res/admins.txt");
    serv = new BingoServ(Integer.parseInt(args[4]),this);
    serv.startSrv();
    bingoURL = args[5];
    MongoClientURI connStr = new MongoClientURI("mongodb://bingobot:" + args[6] + "@localhost:27017/BingoBase");
    MongoClient mongoClient = new MongoClient(connStr);
    MongoDatabase bingoBase = mongoClient.getDatabase("BingoBase");
    playData = bingoBase.getCollection("players");
}

@Override
public void create(JSONObject config) {
    String key = config.getString("key");
    if (mongos.containsKey(key))
        return;

    String schema = config.getString("schema");
    if (validator.isEmpty(schema))
        throw new NullPointerException("未设置schema值[" + config + "]！");

    JSONArray array = config.getJSONArray("ips");
    if (array == null || array.size() == 0)
        throw new NullPointerException("未设置ips值[" + config + "]！");

    String username = config.getString("username");
    String password = config.getString("password");
    MongoClientOptions.Builder builder = MongoClientOptions.builder().connectionsPerHost(maxActive).maxWaitTime(maxWait);
    List<MongoClient> list = new ArrayList<>();
    for (int i = 0; i < array.size(); i++)
        list.add(new MongoClient(new MongoClientURI("mongodb://" + username + ":" + password + "@" + array.getString(i) + "/" + schema, builder)));
    schemas.put(key, schema);
    mongos.put(key, list);

    if (logger.isDebugEnable())
        logger.debug("Mongo数据库[{}]初始化完成。", config);
}

public static boolean connectionTest(MongoDBConfig mongoDBConfig) {
    Logging.disableMongoDBLogging();
    boolean success = true;
    MongoClient mongoClient = null;
    try {
        mongoClient = new MongoClient(new MongoClientURI("mongodb://" + mongoDBConfig.getIp() + ":" + mongoDBConfig.getPort()));
        mongoClient.getDatabaseNames();
    } catch (MongoException e) {
        success = false;
    } finally {
        if (mongoClient != null) {
            mongoClient.close();
        }
        Logging.enableMongoDBLogging();
    }
    return success;
}

/**
 * 指定使用的mongouri key
 *
 * @return key
 */
private MongoClientURI determineTargetDataSource() {
    String currentKey = MongoDataSourceHolder.get();
    MongoClientURI mongoURI;

    mongoURI = mongoURIs.get(currentKey);

    if (null == mongoURI) {
        mongoURI = defaultMongoURI;
    }

    if (null == mongoURI) {
        throw new CannotGetMongoDbConnectionException(String.format("determine current lookup key '%s' not exist!", currentKey));
    }

    return mongoURI;
}

public DBManager(String host, int port, String dbname, String username, String password) {
    // Construct mongo url
    if (dbname == null || dbname.isEmpty()) dbname = "Grimoire";
    if (host == null || host.isEmpty()) host = "127.0.0.1";
    if (port <= 0 || port >= 65535) port = 27017;
    String mongoURL = host + ":" + port + "/" + dbname;
    if (username != null && !username.isEmpty()) {
        String auth = username;
        if (password != null && !password.isEmpty()) auth += ":" + password;
        mongoURL = auth + "@" + mongoURL;
    }
    mongoURL = "mongodb://" + mongoURL;

    // Construct client
    MongoClient client = new MongoClient(new MongoClientURI(mongoURL));

    // Wrap with jongo
    jongo = new Jongo(client.getDB(dbname));
}

public static void createUser() throws IOException, InterruptedException {
    String mongoHost = MultiValueConfigLoader.getConfig("MONGODB-integrationTest");
    log.info("Host is " + mongoHost);
    if (mongoHost != null) {
        MongoClientURI uri = new MongoClientURI(mongoHost);
        List<String> hosts = uri.getHosts();
        for (String host : hosts) {
            String[] cmdarray = createSetupCommand(host, uri.getDatabase(), uri.getUsername(), new String(uri.getPassword()));
            Process process = Runtime.getRuntime().exec(cmdarray);
            int retVal = process.waitFor();
            log.info(String.format("retVal=%s", retVal));
            log.info("output is " + IOUtils.toString(process.getInputStream()));
            if (retVal != 0) {
                log.info("error is " + IOUtils.toString(process.getErrorStream()));
            }
        }
    } else {
        log.error("mongo host is not defined!");
    }
}

@PostConstruct
public void initConnection() {
    String mongoHost = env.getProperty("mongodb.server.host", "127.0.0.1"); // env var MONGODB_SERVER_HOST takes precedence
    String mongoPort = env.getProperty("mongodb.server.port", "27017"); // env var MONGODB_SERVER_PORT takes precedence
    String mongoUser = env.getProperty("mongodb.user", "mongodb"); // env var MONGODB_USER takes precedence
    String mongoPassword = env.getProperty("mongodb.password", "mongodb"); // env var MONGODB_PASSWORD takes precedence
    String mongoDBName = env.getProperty("mongodb.database", "mongodb"); // env var MONGODB_DATABASE takes precedence

    try {
        String mongoURI = "mongodb://" + mongoUser + ":" + mongoPassword + "@" + mongoHost + ":" + mongoPort + "/" + mongoDBName;
        System.out.println("[INFO] Connection string: " + mongoURI);
        MongoClient mongoClient = new MongoClient(new MongoClientURI(mongoURI));
        mongoDB = mongoClient.getDatabase(mongoDBName);
    } catch (Exception e) {
        System.out.println("[ERROR] Creating the mongoDB. " + e.getMessage());
        mongoDB = null;
    }
}

@Override
protected Integer doInBackground(User... params) {
    try {
        MongoClientURI mongoClientURI = new MongoClientURI(Variables.mongo_uri);
        MongoClient mongoClient = new MongoClient(mongoClientURI);
        MongoDatabase dbMongo = mongoClient.getDatabase(mongoClientURI.getDatabase());
        MongoCollection<Document> coll = dbMongo.getCollection("users");
        User local_user = params[0];
        if (!local_user.getEmail().equals(original_email)) {
            Document user = coll.find(eq("email", local_user.getEmail())).first();
            if (user != null) {
                return 1; // Repeated email
            }
        }

        Document search = new Document("_id", new ObjectId(local_user.getId()));
        Document replacement = new Document("$set", local_user.getRegisterDocument());
        // We update some fields of the documents without affecting the rest
        coll.updateOne(search, replacement);
        mongoClient.close();
        return 0; //Successfully saved
    } catch (Exception e) {
        return 2; // Error
    }
}

@Override
protected Integer doInBackground(User... params) {
    try {
        MongoClientURI mongoClientURI = new MongoClientURI(Variables.mongo_uri);
        MongoClient mongoClient = new MongoClient(mongoClientURI);
        MongoDatabase dbMongo = mongoClient.getDatabase(mongoClientURI.getDatabase());
        MongoCollection<Document> coll = dbMongo.getCollection("users");
        User local_user = params[0];
        Document user = coll.find(eq("email", local_user.getEmail())).first();
        mongoClient.close();
        if (user == null || !(user.get("pin").equals(local_user.getPin()))) {
            return 1; // Wrong data
        }
        Date d = (Date) user.get("birthDate");
        Calendar cal = Calendar.getInstance();
        cal.setTime(d);
        // WARNING: Calendar.MONTH starts in 0 Calendar.DAY_OF_MONTH starts in 1
        local_user.completeSignIn((String) user.get("name"), cal.get(Calendar.DAY_OF_MONTH) - 1, cal.get(Calendar.MONTH), cal.get(Calendar.YEAR), (Boolean) user.get("gender"), user.getObjectId("_id").toString());
        return 0; //Successfully saved
    } catch (Exception e) {
        return 2; // Error
    }
}

@Override
protected Integer doInBackground(User... params) {
    try {
        MongoClientURI mongoClientURI = new MongoClientURI(Variables.mongo_uri);
        MongoClient mongoClient = new MongoClient(mongoClientURI);
        MongoDatabase dbMongo = mongoClient.getDatabase(mongoClientURI.getDatabase());
        MongoCollection<Document> coll = dbMongo.getCollection("users");
        User local_user = params[0];
        if (coll.find(eq("email", local_user.getEmail())).first() != null) {
            mongoClient.close();
            return 1; // Repeated email
        }
        Document document = local_user.getRegisterDocument();
        coll.insertOne(document);
        local_user.setId(document.getObjectId("_id").toString());
        mongoClient.close();
        return 0; //Successfully saved
    } catch (Exception e) {
        return 2; // Error
    }
}

@SuppressWarnings("deprecation")
    @Override
    public void init(String configurationName, Map<String, String> properties) {
        if (mongoClient == null) {
            this.configurationName = configurationName;
            this.properties = new HashMap<>(properties);

            try {
                String mongoClientURI = "mongodb://" + property("user") + ":" + property("pass") + "@" + property("host") + ":" + property("port") + "/" + property("name") + "?authSource=admin";

                MongoClientURI uri = new MongoClientURI(mongoClientURI);
                mongoClient = new MongoClient(uri);                


//                MongoClientOptions options = MongoClientOptions.builder().connectionsPerHost(100).autoConnectRetry(true)
//                    .connectTimeout(30000).socketTimeout(60000).socketKeepAlive(true).build();
//                mongoClient = new MongoClient(new ServerAddress(property("host"), Integer.parseInt(property("port"))),
//                    options);
            } catch (Throwable t) {
                throw new IllegalStateException(t);
            }
        }
    }

private void startMangoDb() throws InterruptedException {
    startInNewThread(() -> {
        try {
            MongodStarter starter = MongodStarter.getDefaultInstance();
            IMongodConfig mongodConfig = new MongodConfigBuilder()
                    .version(Version.Main.PRODUCTION)
                    .net(new Net(12345, Network.localhostIsIPv6()))
                    .pidFile(new File("target/process.pid").getAbsolutePath())
                    .replication(new Storage(new File("target/tmp/mongodb/").getAbsolutePath(), null, 0))
                    .build();
            logger.debug("Would download MongoDB if not yet downloaded.");
            MongodExecutable mongodExecutable = starter.prepare(mongodConfig);
            logger.debug("Done with downloading MongoDB exec.");
            mongodExecutable.start();

            MongoClientURI uri = new MongoClientURI("mongodb://localhost:12345/eventStreamAnalytics");
            MongoClient client = new MongoClient(uri);
            MongoDatabase mongoDatabase = client.getDatabase(uri.getDatabase());
            mongoDatabase.createCollection("events");
        } catch (Exception ex) {
            logger.error("Failed to start MongoDB", ex);
            throw new RuntimeException(ex);
        }
    }, "MangoDB").join();
    logger.debug("Successfully Started MongoDB.");
}

@Override
public boolean start() {
  Read spec = source.spec;
  MongoClientOptions.Builder optionsBuilder = new MongoClientOptions.Builder();
  optionsBuilder.maxConnectionIdleTime(spec.maxConnectionIdleTime());
  optionsBuilder.socketKeepAlive(spec.keepAlive());
  client = new MongoClient(new MongoClientURI(spec.uri(), optionsBuilder));

  MongoDatabase mongoDatabase = client.getDatabase(spec.database());

  MongoCollection<Document> mongoCollection = mongoDatabase.getCollection(spec.collection());

  if (spec.filter() == null) {
    cursor = mongoCollection.find().iterator();
  } else {
    Document bson = Document.parse(spec.filter());
    cursor = mongoCollection.find(bson).iterator();
  }

  return advance();
}

@Override
public void prepare() {
    // 初始化增量的开始时间
    startTime = new StartTime(sinceTime, lastRunMetadataPath);

    // 获取用户自定义的筛选条件
    queryDocument = parseQueryDocument(query);

    // 获取是否需要转换Binary对象为byte[]
    if (binaryFields != null && binaryFields.size() > 0) {
        needConvertBin = true;
    }

    // 连接client
    mongoClient = new MongoClient(new MongoClientURI(uri));
    database = mongoClient.getDatabase(dbName);
    coll = database.getCollection(collection);
}

@Override
public void prepare() {
    // 获取是否需要转换Binary
    if (bin_fields != null && bin_fields.size() > 0) {
        convertBin = true;
    }

    // 准备since_time
    prepareSinceTime();

    // 将filter查询语句转换为Document对象
    filterDocument = parseFilterDocument(filter);

    // 连接client
    mongoClient = new MongoClient(new MongoClientURI(uri));
    database = mongoClient.getDatabase(db_name);
    coll = database.getCollection(collection);
}

@BeforeClass
public static void setUpAll() throws Exception {
    morphiaBundle = new MorphiaPackageBundle<DummyConfiguration>(DummyEntity.class.getPackage().getName(), false) {
        @Override
        protected MongoConfiguration getMongo(DummyConfiguration configuration) {
            UriMongoConfiguration mongoConfiguration = new UriMongoConfiguration();
            mongoConfiguration.setDbName("test");
            mongoConfiguration.setStoreEmpties(false);
            mongoConfiguration.setStoreNulls(false);
            mongoConfiguration.setIgnoreFinals(false);
            mongoConfiguration.setUseLowerCaseCollectionNames(false);
            mongoConfiguration.setUri(new MongoClientURI(String.format("mongodb://localhost:%d", port)));

            return mongoConfiguration;
        }
    };
}

public void initialize(IMongoClientOptionsHandler optionsHandler, MongoDataSourceCfgMeta cfgMeta) throws Exception {
    __cfgMeta = cfgMeta;
    MongoClientOptions.Builder _builder = null;
    if (optionsHandler != null) {
        _builder = optionsHandler.handler(cfgMeta.getName());
    }
    if (_builder == null) {
        _builder = MongoClientOptions.builder();
    }
    if (StringUtils.isNotBlank(cfgMeta.getConnectionUrl())) {
        __mongoClient = new MongoClient(new MongoClientURI(cfgMeta.getConnectionUrl(), _builder));
    } else {
        String _username = StringUtils.trimToNull(cfgMeta.getUserName());
        String _password = StringUtils.trimToNull(cfgMeta.getPassword());
        if (_username != null && _password != null) {
            if (__cfgMeta.isPasswordEncrypted() && __cfgMeta.getPasswordClass() != null) {
                _password = __cfgMeta.getPasswordClass().newInstance().decrypt(_password);
            }
            MongoCredential _credential = MongoCredential.createCredential(cfgMeta.getUserName(), cfgMeta.getDatabaseName(), _password == null ? null : _password.toCharArray());
            __mongoClient = new MongoClient(cfgMeta.getServers(), Collections.singletonList(_credential), _builder.build());
        } else {
            __mongoClient = new MongoClient(cfgMeta.getServers(), _builder.build());
        }
    }
}

MongoApplicationStructure(final String applicationName)
{
    super( applicationName );

    // Turn off the really annoying MongoDB spam :/
    {
        LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
        Logger rootLogger = loggerContext.getLogger( "org.mongodb.driver" );
        rootLogger.setLevel( Level.OFF );
    }

    this.mongoClient = new MongoClient( new MongoClientURI( CoreConfig.MongoDB.uri ) );
    this.accountManager = createNewAccountManager();
    xyz.kvantum.server.api.logging.Logger.info( "Initialized MongoApplicationStructure: {}", this
            .applicationName );

    this.morphia = new Morphia();
    this.morphia.mapPackage( "com.github.intellectualsites.kvantum.implementation" );
    this.morphiaDatastore = morphia.createDatastore( this.mongoClient, CoreConfig.MongoDB.dbMorphia );
}

@Override
public void startUp() {
    Logger.getLogger("org.mongodb").setLevel(Level.SEVERE);
    Logger.getLogger("com.mongodb").setLevel(Level.SEVERE);

    AppConfiguration configuration = AppConfiguration.instance();
    this.databaseName = configuration.getProperty("database.name");
    if (this.databaseName == null) {
        throw new VerbumDominiException("Property database.name not found in app-configuration.properties file.");
    }

    String connectionUrl = configuration.getProperty("mongodb.connection.url");

    if (Environments.TEST.equals(configuration.getEnvironment())) {
        this.databaseName = "verbum_domini_test";
        connectionUrl = "mongodb://localhost";
    } else if (Environments.PRODUCTION.equals(configuration.getEnvironment())) {
        this.databaseName = System.getenv("MONGOLAB_DB_NAME");
        connectionUrl = System.getenv("MONGOLAB_URI");
    }

    MongoClientOptions.Builder options = this.buildOptions(configuration);
    MongoClientURI uri = new MongoClientURI(connectionUrl, options);

    this.mongoClient = new MongoClient(uri);
}

/**
 * Connects to the database
 */
private void connectToDatabase() {
    try {
        logger.debug("Connect MongoDB");
        this.cl = new MongoClient(new MongoClientURI(this.url));
        mongoCollection = cl.getDB(this.db).getCollection(this.collection);

        BasicDBObject idx = new BasicDBObject();
        idx.append(FIELD_TIMESTAMP, 1).append(FIELD_ITEM, 1);
        this.mongoCollection.createIndex(idx);
        logger.debug("Connect MongoDB ... done");
    } catch (Exception e) {
        logger.error("Failed to connect to database {}", this.url);
        throw new RuntimeException("Cannot connect to database", e);
    }
}

public MongoDbConnection connect(ConnectionSettings connectionSettings) {
  StringBuilder authString = new StringBuilder();

  String user = connectionSettings.getUser();
  if (user != null && !user.isEmpty()) {
    authString.append(user);
    String password = connectionSettings.getPassword();
    if (password != null && !password.isEmpty()) {
      authString.append(":").append(password);
    }
    authString.append("@");
  }
  String uri = String.format("mongodb://%s%s", authString, connectionSettings.getHost());
  Builder options = MongoClientOptions.builder().serverSelectionTimeout(10000);
  MongoClient client = new MongoClient(new MongoClientURI(uri, options));
  MongoConnection mongoConnection = new MongoConnection(client);
  return new MongoDbConnection(mongoConnection, connectionSettings);
}

private MongoDatabase connect(final PluginTask task) throws UnknownHostException, MongoException
{
    MongoClient mongoClient;
    String database;

    if (!task.getUri().isPresent() && !task.getHosts().isPresent()) {
        throw new ConfigException("'uri' or 'hosts' is required");
    }

    if (task.getUri().isPresent()) {
        MongoClientURI uri = new MongoClientURI(task.getUri().get());
        database = uri.getDatabase();
        mongoClient = new MongoClient(uri);
    }
    else {
        mongoClient = createClientFromParams(task);
        database = task.getDatabase().get();
    }

    MongoDatabase db = mongoClient.getDatabase(database);
    // Get collection count for throw Exception
    db.getCollection(task.getCollection()).count();
    return db;
}

@Test
public void testRunWithConnectionParams() throws Exception
{
    MongoClientURI uri = new MongoClientURI(MONGO_URI);
    String host = uri.getHosts().get(0);
    Integer port = (host.split(":")[1] != null) ? Integer.valueOf(host.split(":")[1]) : 27017;
    ConfigSource config = Exec.newConfigSource()
            .set("hosts", Arrays.asList(ImmutableMap.of("host", host.split(":")[0], "port", port)))
            .set("user", uri.getUsername())
            .set("password", uri.getPassword())
            .set("database", uri.getDatabase())
            .set("collection", MONGO_COLLECTION);
    PluginTask task = config.loadConfig(PluginTask.class);

    dropCollection(task, MONGO_COLLECTION);
    createCollection(task, MONGO_COLLECTION);
    insertDocument(task, createValidDocuments());

    plugin.transaction(config, new Control());
    assertValidRecords(getFieldSchema(), output);
}

private void startMangoDb() throws InterruptedException {
    startInNewThread(() -> {
        try {
            MongodStarter starter = MongodStarter.getDefaultInstance();
            IMongodConfig mongodConfig = new MongodConfigBuilder()
                    .version(Version.Main.PRODUCTION)
                    .net(new Net(12345, Network.localhostIsIPv6()))
                    .pidFile(new File("target/process.pid").getAbsolutePath())
                    .replication(new Storage(new File("target/tmp/mongodb/").getAbsolutePath(), null, 0))
                    .build();
            logger.debug("Would download MongoDB if not yet downloaded.");
            MongodExecutable mongodExecutable = starter.prepare(mongodConfig);
            logger.debug("Done with downloading MongoDB exec.");
            mongodExecutable.start();

            MongoClientURI uri = new MongoClientURI("mongodb://localhost:12345/eventStreamAnalytics");
            MongoClient client = new MongoClient(uri);
            MongoDatabase mongoDatabase = client.getDatabase(uri.getDatabase());
            mongoDatabase.createCollection("events");
        } catch (Exception ex) {
            logger.error("Failed to start MongoDB", ex);
            throw new RuntimeException(ex);
        }
    }, "MangoDB").join();
    logger.debug("Successfully Started MongoDB.");
}

/**
 * @return A {@link TenantInformationProvider} that gets the database name from {@link #dataBaseName}.
 */
@Bean
public TenantInformationProvider tenantProvider() {
    return new TenantInformationProvider() {
        @Override
        public String getDatabaseName() {
            if("failure".equals(dataBaseName.get())) {
                throw new RuntimeException("On purpose thrown exception.");
            }
            return dataBaseName.get();
        }

        @Override
        public MongoClientURI getDatabaseURI() {
            String uri = "mongodb://fake_host:27017/" + dataBaseName.get();
            return new MongoClientURI(uri);
        }
    };
}

public synchronized void open() throws Exception {


    if ( mongouri == null ){

        if ( clientMongoMap.containsKey( ip + port ) ){
            mdb = clientMongoMap.get( ip + port ).open().getDatabase( db );
        }else{
            mongoclient = newClient( server + ":" + port, user, pass, db );
            MongoClientWrapper mcw = new MongoClientWrapper(mongoclient);
            clientMongoMap.put(  ip + port, mcw );
            mdb = mcw.open().getDatabase( db );
        }

    }else{
        MongoClientURI clientURI = new MongoClientURI(mongouri);
        mongoclient = new MongoClient( clientURI );
        mdb     = mongoclient.getDatabase( clientURI.getDatabase() );
    }

    lastUsed    = System.currentTimeMillis();
}

@Before
public void init() throws MongobeeException, UnknownHostException {
  fakeDb = new Fongo("testServer").getDB("mongobeetest");
  fakeMongoDatabase = new Fongo("testServer").getDatabase("mongobeetest");
  when(dao.connectMongoDb(any(MongoClientURI.class), anyString()))
      .thenReturn(fakeMongoDatabase);
  when(dao.getDb()).thenReturn(fakeDb);
  when(dao.getMongoDatabase()).thenReturn(fakeMongoDatabase);
  doCallRealMethod().when(dao).save(any(ChangeEntry.class));
  doCallRealMethod().when(dao).setChangelogCollectionName(anyString());
  doCallRealMethod().when(dao).setIndexDao(any(ChangeEntryIndexDao.class));
  dao.setIndexDao(indexDao);
  dao.setChangelogCollectionName(CHANGELOG_COLLECTION_NAME);

  runner.setDbName("mongobeetest");
  runner.setEnabled(true);
  runner.setChangeLogsScanPackage(MongobeeTestResource.class.getPackage().getName());
}

@Before
public void init() throws Exception {
  fakeDb = new Fongo("testServer").getDB("mongobeetest");
  fakeMongoDatabase = new Fongo("testServer").getDatabase("mongobeetest");

  when(dao.connectMongoDb(any(MongoClientURI.class), anyString()))
      .thenReturn(fakeMongoDatabase);
  when(dao.getDb()).thenReturn(fakeDb);
  when(dao.getMongoDatabase()).thenReturn(fakeMongoDatabase);
  when(dao.acquireProcessLock()).thenReturn(true);
  doCallRealMethod().when(dao).save(any(ChangeEntry.class));
  doCallRealMethod().when(dao).setChangelogCollectionName(anyString());
  doCallRealMethod().when(dao).setIndexDao(any(ChangeEntryIndexDao.class));
  dao.setIndexDao(indexDao);
  dao.setChangelogCollectionName(CHANGELOG_COLLECTION_NAME);

  runner.setDbName("mongobeetest");
  runner.setEnabled(true);
}

@Before
public void init() throws Exception {
  fakeDb = new Fongo("testServer").getDB("mongobeetest");
  fakeMongoDatabase = new Fongo("testServer").getDatabase("mongobeetest");

  when(dao.connectMongoDb(any(MongoClientURI.class), anyString()))
      .thenReturn(fakeMongoDatabase);
  when(dao.getDb()).thenReturn(fakeDb);
  when(dao.getMongoDatabase()).thenReturn(fakeMongoDatabase);
  when(dao.acquireProcessLock()).thenReturn(true);
  doCallRealMethod().when(dao).save(any(ChangeEntry.class));
  doCallRealMethod().when(dao).setChangelogCollectionName(anyString());
  doCallRealMethod().when(dao).setIndexDao(any(ChangeEntryIndexDao.class));
  dao.setIndexDao(indexDao);
  dao.setChangelogCollectionName(CHANGELOG_COLLECTION_NAME);

  runner.setDbName("mongobeetest");
  runner.setEnabled(true);
}

public ServiceManager(Map<String, Object> svcConfig, MongoClientURI defaultUri) {

        this.svcConfig = svcConfig;
        this.factory = new ServiceFactory();
        this.defaultDbUri = defaultUri;

        logger.info("Initializing configured services");
        // Load the configured AsyncService implementation
        Map<String, Object> asyncServiceConfig = getServiceConfig(ASYNC_SERVICE_KEY, DEFAULT_ASYNC_SERVICE);
        if(asyncServiceConfig != null){
            factory.createAndRegisterService(
                    AsyncService.class, asyncServiceConfig, this.defaultDbUri);
        }

        // Load the configured UserGraphService implementation
        Map<String, Object> channelServiceConfig = getServiceConfig(CHANNEL_SERVICE_KEY, DEFAULT_CHANNEL_SERVICE);
        factory.createAndRegisterService(
                ChannelService.class, channelServiceConfig, this.defaultDbUri);

    }

public DefaultChannelService(
        final MongoClientURI dbUri, 
        final ChannelServiceConfiguration config){
    super(dbUri, config);

    this.config = config;

    if(this.config.channel_task_thread_pool_size > 0){
        this.taskExecutor = Executors.newScheduledThreadPool(
                this.config.channel_task_thread_pool_size);
    }
    else {
        this.taskExecutor = null;
    }

}

@Override
protected void setUpValve(Tomcat tomcat) throws UnknownHostException {
  // remove AccessLogValve
  for (Valve vl : tomcat.getHost().getPipeline().getValves()) {
    if (vl.getClass().equals(AccessLogValve.class)) {
      tomcat.getHost().getPipeline().removeValve(vl);
    }
  }

  mongoClient = new MongoClient(new MongoClientURI(url));
  db = mongoClient.getDB(dbName);

  MongoAccessLogValve mavl = new MongoAccessLogValve();
  mavl.setUri(url);
  mavl.setDbName(dbName);
  mavl.setCollName(collName);
  mavl.setPattern(pattern);

  tomcat.getHost().getPipeline().addValve(mavl);
}

@Test
public void getLastCheckPointWithValidCollection() throws UnknownHostException, ParseException {
    String mongoHost = getMongoHost();
    when(context.get("field"))
            .thenReturn("date");
    when(context.get("type"))
            .thenReturn("java.util.Date");
    when(context.get("format"))
            .thenReturn(DATE_FORMAT_YYYY_MM_DD_T_HH_MM_SS_XXX);
    when(context.get("mongoUri"))
            .thenReturn("mongodb://" + mongoHost + "/" + DB_TEST + ".validCollection");
    handler = new MongoFilterHandler();
    mongoClient = new MongoClient(new MongoClientURI("mongodb://" + mongoHost));
    mongoClient.getDB(DB_TEST).createCollection("validCollection", null);
    mongoClient.getDB(DB_TEST).getCollection("validCollection").save(populateDocument());
    when(context.get("filterType"))
            .thenReturn("com.stratio.ingestion.source.rest.url.filter.type.DateCheckpointType");
    handler = spy(new MongoFilterHandler());
    doReturn(context).when(handler).loadCheckpointContext(context);
    handler.configure(context);

    final Map<String, String> lastCheckpoint = handler.getLastFilter(context);
    assertThat(lastCheckpoint).isNotNull();
}