import com.mongodb.MongoURI; //导入依赖的package包/类
public Datastore getDataStore() throws UnknownHostException {
    Logger.getLogger( "org.mongodb" ).setLevel(Level.OFF);

    Morphia morphia = getMorphiaInstance();
    MongoURI mongoClientURI = new MongoURI(uri);
    Mongo mongoClient = new Mongo(mongoClientURI);

    Datastore ds = morphia.createDatastore(mongoClient, mongoClientURI.getDatabase());
    ds.ensureIndexes();
    ds.ensureCaps();

    return ds;
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
@Override
public State makeState(Map conf, int partitionIndex, int numPartitions) {
    MongoURI mongoURI = new MongoURI(mongoUri);
    Mongo mongo;
    try {
        mongo = new Mongo(mongoURI);
    } catch (UnknownHostException e) {
        throw new RuntimeException(e);
    }
    JacksonProcessor processor = new JacksonProcessor() {
        @Override
        public <T> T unmarshall(String json, Class<T> clazz) {
            if (TransactionalValue.class.equals(clazz) || OpaqueValue.class.equals(clazz)) {
                return (T) super.unmarshall(json, entityClass);
            }
            return super.unmarshall(json, clazz);
        }
    };
    Jongo jongo = new Jongo(mongo.getDB(mongoURI.getDatabase()), processor, processor);
    collection = jongo.getCollection(mongoURI.getCollection());

    MapState<T> mapState;
    switch (type) {
        case NON_TRANSACTIONAL:
            mapState = buildNonTransactional();
            break;
        case TRANSACTIONAL:
            collection.getDBCollection().setWriteConcern(WriteConcern.SAFE);
            mapState = buildTransactional();
            break;
        case OPAQUE:
            mapState = buildOpaque();
            break;
        default:
            throw new RuntimeException("Unknown state type: " + type);
    }

    return new SnapshottableMap<T>(mapState, Arrays.<Object>asList(opts.globalKey));
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
public void addJob(String mongoURI) {
  @SuppressWarnings("deprecation")
  MongoURI uri = new MongoURI(mongoURI);

  String database = uri.getDatabase(), collection = uri.getCollection();
  String tableName = getTranslator().mapNamespaceToHBaseTable(database, collection);

  if (tableName == null) {
    log.info("Skipping namespace '" + database + "." + collection + "' because it doesn't map to an HBase table");
    return;
  }

  addJob(uri, tableName);
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
private byte[][] calculateRegionSplits(MongoURI uri, String tableName) throws Exception {
  DBCollection collection = uri.connectDB().getCollection(uri.getCollection());
  long size = collection.getStats().getLong("size");
  long regionSize = ConfigUtil.getPresplitTableRegionSize(_conf);
  int numRegions = (int) Math.min((size / regionSize) + 1, 4096);

  if (numRegions > 1) {
    log.info("Pre-splitting " + tableName + " into " + numRegions + " regions");
    RegionSplitter.UniformSplit splitter = new RegionSplitter.UniformSplit();
    return splitter.split(numRegions);
  } else {
    log.info("Not splitting " + tableName + ", because the data can fit into a single region");
    return new byte[0][0];
  }
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
public BulkImportJob(Configuration conf, MongoURI mongoURI, String tableName) throws IOException {
  Configuration cloned = new Configuration(conf);
  _uuid = UUID.randomUUID().toString();
  _job = new Job(cloned, tableName + "." + _uuid);

  _mongoURI = mongoURI;
  _tableName = tableName;

  String tmpPath = ConfigUtil.getTemporaryHFilePath(conf);
  _hfilePath = new Path(tmpPath, _uuid);

  setupJob();
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
Mongo setupMongo() {
  return uri() == null ? new Mongo() : new Mongo(new MongoURI(uri()));
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
/**
 * Create an instance of <code>Mongo</code> based on the information
 * provided in the configuration files located into <code>WEB-INF/conf</code>, if the instance has already been
 * created using the same information, so it uses the same instance.
 *
 * @param context <code>ServletContext</code> of a web application.
 * @return an instance of a <code>CollectionManager</code>.
 */
public static CollectionManager setup(ServletContext context) {
    try {
        File props = getPropertiesFile(context);
        if (props == null) {
            throw new FileNotFoundException("application or database configuration file not found.");
        }
        InputStream in = new FileInputStream(props);
        Properties properties = new Properties();
        properties.load(in);
        StringBuilder builder = new StringBuilder();
        builder.append(MongoURI.MONGODB_PREFIX);
        String user, password, host, port, dbName;
        user = properties.containsKey("mongocom.user") ? properties.getProperty("mongocom.user") : "";
        password = properties.containsKey("mongocom.password") ? properties.getProperty("mongocom.password") : "";
        host = properties.containsKey("mongocom.host") ? properties.getProperty("mongocom.host") : "";
        port = properties.containsKey("mongocom.port") ? properties.getProperty("mongocom.port") : "";
        dbName = properties.containsKey("mongocom.database") ? properties.getProperty("mongocom.database") : "";
        if (!user.equals("")) {
            builder.append(user).append(":").append(password).append("@");
        }
        if (host.equals("")) {
            builder.append("localhost");
        } else {
            builder.append(host);
        }
        if (!port.equals("")) {
            builder.append(":");
            builder.append(port);
        }
        builder.append("/");
        if (!dbName.equals("")) {
            builder.append(dbName);
        }
        LOG.log(Level.INFO, "Mongo URI: {0}", builder.toString());
        MongoURI uri = new MongoURI(builder.toString());
        client = MongoClient.Holder.singleton().connect(uri);
        return new CollectionManager(client, dbName);
    } catch (IOException ex) {
        LOG.log(Level.SEVERE, null, ex);
    }
    return null;
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
@Bean
public MongoURI mongoURI() throws Exception {
	return new MongoURI(env.getProperty("dbUrl"));
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
public ZettlMongoDbFactory(MongoURI uri) throws UnknownHostException {
	mongoDbFactory = new SimpleMongoDbFactory(uri);
	mongoDbFactory.setWriteConcern(WriteConcern.ACKNOWLEDGED);
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
public void addJobsForNamespace(String rawURI, String database, String collection) {
  Mongo mongo;

  @SuppressWarnings("deprecation")
  MongoURI uri = new MongoURI(rawURI);

  try {
    mongo = uri.connect();
  } catch (UnknownHostException e) {
    throw new IllegalArgumentException("Failed to connect to MongoDB using URI: " + uri, e);
  }

  if (database != null && collection != null) {
    if(!mongo.getDB(database).collectionExists(collection)) {
      throw new IllegalArgumentException("Couldn't find namespace: " + database + "." + collection);
    }

    buildJob(uri, database, collection);
    return;
  } else if (database == null && collection != null) {
    throw new IllegalArgumentException("You can't specify a MongoDB collection without also specifying a database");
  }

  List<String> databases;
  if (database != null) {
    databases = new ArrayList<String>(1);
    databases.add(database);
  } else {
    databases = mongo.getDatabaseNames();
  }

  for (String db : databases) {
    if (db.equals("local") || db.equals("admin")) continue;

    Set<String> collections = mongo.getDB(db).getCollectionNames();
    for (String coll : collections) {
      if (coll.startsWith("system.")) continue;
      buildJob(uri, db, coll);
    }
  }
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
private void buildJob(MongoURI uri, String database, String collection) {
  UriBuilder builder = UriBuilder.fromUri(uri.toString());
  String builtURI = builder.path("{db}.{coll}").build(database, collection).toString();

  addJob(builtURI);
}
 

import com.mongodb.MongoURI; //导入依赖的package包/类
public MongoURI getMongoURI() {
  return _mongoURI;
}