import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Creates a large collection of documents that test should only read from.
 */
protected static void disableBalancer() {

    final MongoClientConfiguration config = new MongoClientConfiguration();
    config.addServer(createAddress());

    final MongoClient mongoClient = MongoFactory.createClient(config);
    try {
        // Turn off the balancer - Can confuse the test counts.
        final boolean upsert = true;
        mongoClient
        .getDatabase("config")
        .getCollection("settings")
        .update(where("_id").equals("balancer"),
                d(e("$set", d(e("stopped", true)))), false, upsert);
    }
    finally {
        IOUtils.close(mongoClient);
    }
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Test method for {@link X509Authenticator}.
 *
 * @throws GeneralSecurityException
 *             On a failure to load the credentials.
 */
@Test
public void testAuthenticateReadWrite() throws GeneralSecurityException {
    final MongoClientConfiguration config = new MongoClientConfiguration(
            URI);
    config.addCredential(Credential.builder()
            .userName("CN=test-rsa.example.com,OU=example,OU=example2")
            .x509());
    config.setSocketFactory(mySocketFactory);

    final MongoClient client = MongoFactory.createClient(config);

    final MongoCollection collection = client.getDatabase("test")
            .getCollection("test");

    collection.insert(Durability.ACK, BuilderFactory.start());

    final MongoIterator<Document> iter = collection.find(BuilderFactory
            .start());
    assertTrue(iter.hasNext());

    collection.delete(MongoCollection.ALL);

    IOUtils.close(client);
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Test method for {@link ScramSha1Authenticator}.
 *
 * @throws GeneralSecurityException
 *             On a failure to load the credentials.
 */
@Test
public void testAuthenticateReadWrite() throws GeneralSecurityException {
    final MongoClientConfiguration config = new MongoClientConfiguration(
            URI);
    config.addCredential(Credential.builder().userName("test_scram")
            .password("test_scram_bad_password".toCharArray()).scramSha1());

    final MongoClient client = MongoFactory.createClient(config);

    final MongoCollection collection = client.getDatabase("test")
            .getCollection("test");

    collection.insert(Durability.ACK, BuilderFactory.start());

    final MongoIterator<Document> iter = collection.find(BuilderFactory
            .start());
    assertTrue(iter.hasNext());

    collection.delete(MongoCollection.ALL);

    IOUtils.close(client);
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Creates a new GroupManager.
 * 
 * @param executor
 *            Used for updating a registration in MongoDB.
 * @param mongoClient
 *            The client for the watcher.
 * @param collection
 *            The collection being watched.
 * @param rootContext
 *            The context for the {@code _id} of items to watch.
 */
public GroupManager(final ScheduledExecutorService executor,
        final MongoClient mongoClient, final MongoCollection collection,
        final String rootContext) {
    myExecutor = executor;
    myMongoClient = mongoClient;
    myCollection = collection;
    myRootContext = rootContext;

    myMembers = new ConcurrentHashMap<String, WeakReference<GroupMember>>();
    myListeners = new CopyOnWriteArrayList<GroupListener>();
    myListener = new GroupWatchListener();

    myWatcher = new Watcher(myMongoClient, myCollection,
            Pattern.compile(myRootContext + ".*"), myListener);
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Performs the document updates using the legacy driver.
 * <p>
 * The main draw back here (other than those discussed in
 * {@link #doSynchronously()}) is the difficulty creating the GeoJSON
 * documents.
 * </p>
 * 
 * @throws UnknownHostException
 *             On an invalid URI.
 */
protected static void doLegacy() throws UnknownHostException {
    // Execute the query to find all of the documents and then
    // update them.
    final com.mongodb.MongoClient legacyClient = new com.mongodb.MongoClient(
            new MongoClientURI(URI));
    final com.mongodb.DBCollection legacyCollection = legacyClient.getDB(
            theCollection.getDatabaseName()).getCollection(
            theCollection.getName());
    try {
        int count = 0;
        for (final DBObject doc : legacyCollection.find()) {
            final Object id = doc.get("_id");
            final Number lat = (Number) doc.get("latitude_deg");
            final Number lon = (Number) doc.get("longitude_deg");

            final BasicDBObject query = new BasicDBObject();
            query.append("_id", id);

            final ArrayList<Double> coordinates = new ArrayList<>();
            coordinates.add(lon.doubleValue());
            coordinates.add(lat.doubleValue());
            final BasicDBObject geojson = new BasicDBObject("type", "Point");
            geojson.append("coordinates", coordinates);
            final BasicDBObject set = new BasicDBObject("loc", geojson);
            final BasicDBObject update = new BasicDBObject("$set", set);

            legacyCollection.update(query, update, /* upsert= */false,
            /* multi= */false, WriteConcern.ACKNOWLEDGED);

            count += 1;
        }
        System.out.printf("Updated %d documents via the legacy driver.%n",
                count);
    }
    finally {
        // Always close the client.
        legacyClient.close();
    }
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * <p>
 * Overridden to create a new Mongo instance around a SerialClientImpl.
 * </p>
 */
@Override
public MongoClient asSerializedClient() {
    if (myClient instanceof SerialClientImpl) {
        return this;
    }

    return new MongoClientImpl(new SerialClientImpl((ClientImpl) myClient));
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Creates the base set of objects for the test.
 */
@Before
public void setUp() {
    myMockMongoClient = EasyMock.createMock(MongoClient.class);
    myMockClient = EasyMock.createMock(Client.class);

    myTestInstance = new MongoDatabaseImpl(myMockMongoClient, myMockClient,
            "test");

    expect(myMockClient.getConfig()).andReturn(
            new MongoClientConfiguration()).anyTimes();
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Test method for {@link MongoClientImpl#asSerializedClient()} .
 */
@Test
public void testAsSerializedClient() {
    final MongoClientImpl impl = new MongoClientImpl(
            new MongoClientConfiguration());
    assertThat(impl.getClient(), instanceOf(ClientImpl.class));
    impl.close();

    final MongoClient serial = impl.asSerializedClient();
    assertThat(serial, instanceOf(MongoClientImpl.class));
    final MongoClientImpl serialImpl = (MongoClientImpl) serial;
    assertThat(serialImpl.getClient(), instanceOf(SerialClientImpl.class));

    assertSame(serial, serial.asSerializedClient());
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Test method for
 * {@link com.allanbank.mongodb.client.MongoImpl#asSerializedClient()} .
 */
@Test
public void testAsSerializedClient() {
    final MongoImpl impl = new MongoImpl(
            new com.allanbank.mongodb.MongoDbConfiguration());
    assertThat(impl.getClient(), instanceOf(ClientImpl.class));
    impl.close();

    final MongoClient serial = impl.asSerializedClient();
    assertThat(serial, instanceOf(MongoClientImpl.class));
    final MongoClientImpl serialImpl = (MongoClientImpl) serial;
    assertThat(serialImpl.getClient(), instanceOf(SerialClientImpl.class));

    assertSame(serial, serial.asSerializedClient());
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Tests that the parsing of an ObjectId in the shell matches that by the
 * driver.
 */
@Test
public void testObjectIdParsingMatches() {
    final MongoClientConfiguration config = new MongoClientConfiguration();
    config.addServer(new InetSocketAddress("127.0.0.1", 27017));
    config.setMaxConnectionCount(1);

    final MongoClient mongo = MongoFactory.createClient(config);
    final MongoDatabase db = mongo.getDatabase("test");
    final MongoCollection collection = db.getCollection("test");

    // Make sure the collection is created and empty.
    db.createCollection("test", BuilderFactory.start());
    collection.delete(BuilderFactory.start(), Durability.ACK);

    final ObjectId id = new ObjectId(0x11223344, 0x5566778899001122L);
    final ObjectIdElement element = new ObjectIdElement("_id", id);
    assertEquals("'_id' : ObjectId('112233445566778899001122')",
            element.toString());

    final Document doc = BuilderFactory.start().add(element).build();
    assertEquals("{ '_id' : ObjectId('112233445566778899001122') }",
            doc.toString());

    final ManagedProcess mp = ourTestSupport
            .run(null, "mongo", "localhost:27017/test", "-eval",
                    "db.test.insert( { '_id' : ObjectId('112233445566778899001122') } );");
    mp.waitFor();

    // Now pull the document back and make sure it matches what we expect.
    final Document result = collection.findOne(BuilderFactory.start());

    assertEquals(doc, result);

    IOUtils.close(mongo);
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Runs a create/modify/delete on a collection.
 * 
 * @param args
 *            Command line arguments. Expect the MongoDB URL, database and
 *            collection name as the first, second and third argument.
 * @throws InterruptedException
 *             If the thread is interrupted.
 * @throws IOException
 *             On a failure to close the MongoDB client.
 */
public static void main(String[] args) throws InterruptedException,
        IOException {
    if (args.length < 3) {
        System.out.println("Usage: java " + ModifyIt.class.getName()
                + " <mongodb-url> <database> <collection>");
        System.exit(1);
    }
    MongoClient client = MongoFactory.createClient(args[0]);
    MongoDatabase database = client.getDatabase(args[1]);
    MongoCollection collection = database.getCollection(args[2]);

    String id = new ObjectId().toHexString();

    Document query = BuilderFactory.start().add("_id", id).build();
    collection.insert(Durability.ACK, query);
    for (int i = 0; i < 10; ++i) {
        DocumentBuilder update = BuilderFactory.start();
        update.push("$inc").add("i", 1);
        collection.update(query, update, false, false, Durability.ACK);

        Thread.sleep(100);
    }

    collection.delete(query, Durability.ACK);

    client.close();
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Runs a single watcher on a collection.
 * 
 * @param args
 *            Command line arguments. Expect the MongoDB URL, database and
 *            collection name.
 * @throws InterruptedException
 *             If the thread is interrupted.
 */
public static void main(String[] args) throws InterruptedException {
    if (args.length < 3) {
        System.out.println("Usage: java " + WatchIt.class.getName()
                + " <mongodb-url> <database> <collection>");
        System.exit(1);
    }

    MongoClient client = MongoFactory.createClient(args[0]);
    MongoDatabase database = client.getDatabase(args[1]);
    MongoCollection collection = database.getCollection(args[2]);

    Watcher watcher = new Watcher(client, collection,
            Pattern.compile(".*"), new WatchListener() {
                @Override
                public void changed(Operation op, String context,
                        Document document) {
                    if (op == Operation.DELETE) {
                        System.out.println(context + ": " + op);
                    }
                    else {
                        System.out.println(context + ": " + op + ": "
                                + document);
                    }
                }
            });
    watcher.start();

    while (true) {
        Thread.sleep(10000000L);
    }
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Runs a single consumer on the queue.
 * 
 * @param args
 *            Command line arguments. Expect the MongoDB URL, database and
 *            collection/queue name.
 * @throws InterruptedException
 *             If interrupted.
 */
public static void main(String[] args) throws InterruptedException {
    if (args.length < 3) {
        System.out.println("Usage: java " + Consumer.class.getName()
                + " <mongodb-url> <database> <queue/collection>");
        System.exit(1);
    }

    MongoClient client = MongoFactory.createClient(args[0]);
    MongoCollection index = client.getDatabase(args[1]).getCollection(
            "lookup");
    Document queueLookupDoc = index.findOne(BuilderFactory.start().add(
            "_id", args[2]));
    DocumentElement cursorElement = queueLookupDoc.get(
            DocumentElement.class, "cursor");

    MongoIterator<Document> iter = client.restart(cursorElement
            .getDocument());
    iter.setBatchSize(3);

    long lastCount = 0;
    while (iter.hasNext()) {
        Document doc = iter.next();
        NumericElement count = doc.get(NumericElement.class, "count");
        if (count == null) {
            System.out.println(iter.next());
        }
        else {
            if ((lastCount + 1) != count.getLongValue()) {
                System.out.println(lastCount);
                System.out.print(count.getLongValue());
                System.out.print("...");
            }
            lastCount = count.getLongValue();
        }

        TimeUnit.MILLISECONDS.sleep(200);

    }
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Runs a single producer on the queue.
 * 
 * @param args
 *            The for the producer. Expect the MongoDBURL, database and
 *            collection/queue name
 * @throws InterruptedException
 *             If the producer is interrupted.
 */
public static void main(String[] args) throws InterruptedException {
    if (args.length < 3) {
        System.out.println("Usage: java " + Producer.class.getName()
                + " <mongodb-url> <database> <queue/collection>");
        System.exit(1);
    }

    MongoClient client = MongoFactory.createClient(args[0]);

    String dbName = args[1];
    String collectionName = args[2];

    MongoDatabase db = client.getDatabase(dbName);
    MongoCollection collection = db.getCollection(collectionName);

    int count = 0;
    UUID producerIdentifier = UUID.randomUUID();
    DocumentBuilder builder = BuilderFactory.start();
    while (true) {
        builder.reset();
        builder.add("_id", new ObjectId());
        builder.add("producer", producerIdentifier);
        builder.add("count", count++);

        collection.insert(builder);

        TimeUnit.MILLISECONDS.sleep(100);

        if ((count % 10) == 0) {
            System.out.println(count);
        }
    }
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Creates a builder for executing background process scripts.
 */
@Before
public void setUp() {
    try {
        final MongoClientConfiguration config = new MongoClientConfiguration(
                ourMongoServerUri);
        config.setMaxConnectionCount(1);
        config.setMaxPendingOperationsPerConnection(10 * 1024);
        config.setLockType(LockType.LOW_LATENCY_SPIN);

        myAsyncMongo = MongoFactory.createClient(config);
        myAsyncDb = myAsyncMongo.getDatabase("asyncTest");
        myAsyncCollection = myAsyncDb.getCollection("test");

        final MongoClientOptions options = new MongoClientOptions.Builder()
                .connectionsPerHost(1).build();

        mySyncMongo = new com.mongodb.MongoClient(
                ourMongoServerUri.getHostName() + ":"
                        + ourMongoServerUri.getPort(), options);
        mySyncDb = mySyncMongo.getDB("syncTest");
        mySyncCollection = mySyncDb.getCollection("test");
    }
    catch (final IOException ioe) {
        final AssertionError error = new AssertionError(ioe.getMessage());
        error.initCause(ioe);
        throw error;
    }
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Run the demo.
 * 
 * @throws IOException
 *             On a failure closing the MongoCLient.
 * @throws InterruptedException
 *             On a failure waiting for a future.
 */
public static void runDemoWithLegacyDriver() throws IOException,
        InterruptedException {
    com.mongodb.MongoClient client2 = new com.mongodb.MongoClient(
            "localhost", 27017);
    DBCollection collection = client2.getDB("db").getCollection(
            "collection");

    // Before we start lets make sure there is not already a document.
    collection.remove(new BasicDBObject());

    // Insert a few documents.
    collection.insert(new BasicDBObject("loc", new int[] { 0, 0 }));
    collection.insert(new BasicDBObject("loc", new int[] { 10, 10 }));
    collection.insert(new BasicDBObject("loc", new int[] { -10, 10 }));
    collection.insert(new BasicDBObject("loc", new int[] { 10, -10 }));
    collection.insert(new BasicDBObject("loc", new int[] { -10, -10 }));
    collection.insert(new BasicDBObject("loc", new int[] { 40, 40 }));
    collection.insert(new BasicDBObject("loc", new int[] { -40, 40 }));
    collection.insert(new BasicDBObject("loc", new int[] { 40, -40 }));
    collection.insert(new BasicDBObject("loc", new int[] { -40, -40 }));
    collection.insert(new BasicDBObject("loc", new int[] { 100, 100 }));
    collection.insert(new BasicDBObject("loc", new int[] { -100, 100 }));
    collection.insert(new BasicDBObject("loc", new int[] { 100, -100 }));
    collection.insert(new BasicDBObject("loc", new int[] { -100, -100 }));

    // Speed up the query.
    collection.createIndex(new BasicDBObject("loc", "2d"));

    // Create the query.
    BasicDBObject geometery = new BasicDBObject("$centerSphere", asList(
            asList(40, -40), 10 / 3963.2));
    BasicDBObject operator = new BasicDBObject("$geoWithin", geometery);
    BasicDBObject query = new BasicDBObject("loc", operator);
    try (DBCursor iter = collection.find(query)) {
        for (DBObject result : iter) {
            System.out.println(result);
        }
    }

    // Always close the client.
    client2.close();
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Tests that the parsing of an ObjectId in the shell matches that by the
 * driver.
 */
@Test
public void testMongoTimeStampParsingMatches() {
    int seconds = 0x12345 * 1000; // Times are truncated.
    final int offset = 0xAB;

    final MongoClientConfiguration config = new MongoClientConfiguration();
    config.addServer(new InetSocketAddress("127.0.0.1", 27017));
    config.setMaxConnectionCount(1);

    final MongoClient mongo = MongoFactory.createClient(config);
    final MongoDatabase db = mongo.getDatabase("test");
    final MongoCollection collection = db.getCollection("test");

    // Make sure the collection is created and empty.
    db.createCollection("test", BuilderFactory.start());
    collection.delete(BuilderFactory.start(), Durability.ACK);

    final MongoTimestampElement element = new MongoTimestampElement("_id",
            0x00012345000000ABL);
    assertEquals("'_id' : Timestamp(" + seconds + ", " + offset + ")",
            element.toString());

    final Document doc = BuilderFactory.start().add(element).build();
    assertEquals("{ '_id' : Timestamp(" + seconds + ", " + offset + ") }",
            doc.toString());

    ManagedProcess mp = ourTestSupport.run(null, "mongo",
            "localhost:27017/test", "-eval",
            "db.test.insert( { '_id' : Timestamp(" + seconds + ", "
                    + offset + ") } );");
    mp.waitFor();

    // Now pull the document back and make sure it matches what we expect.
    Document result = collection.findOne(BuilderFactory.start());

    // For 2.4 the shell parsing changes to be seconds.
    if (!doc.equals(result)) {
        seconds = 0x12345;
        collection.delete(BuilderFactory.start(), Durability.ACK);
        mp = ourTestSupport.run(null, "mongo", "localhost:27017/test",
                "-eval", "db.test.insert( { '_id' : Timestamp(" + seconds
                        + ", " + offset + ") } );");
        mp.waitFor();

        result = collection.findOne(BuilderFactory.start());
    }
    assertEquals(doc, result);

    IOUtils.close(mongo);
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Initializes the queue by creating a new capped collection and creating a
 * new cursor on it. Prints the cursor document to standard out.
 * 
 * @param args
 *            Command line arguments. Expect the MongoDB URL, database and
 *            collection/queue name.
 * @throws IOException
 *             On a failure to close the connection to MongoDB.
 */
public static void main(String[] args) throws IOException {
    if (args.length < 3) {
        System.out.println("Usage: java " + InitializeQueue.class.getName()
                + " <mongodb-url> <database> <queue/collection>");
        System.exit(1);
    }

    MongoClient client = MongoFactory.createClient(args[0]);

    String dbName = args[1];
    String collectionName = args[2];

    MongoDatabase db = client.getDatabase(dbName);
    MongoCollection collection = db.getCollection(collectionName);

    collection.drop();
    db.createCappedCollection(collectionName, 100000000L);

    // For a tailable cursor to initialize we need at least a single
    // document in the collection.
    collection.insert(BuilderFactory.start().add("_id", "seed"));

    Find.Builder builder = new Find.Builder(BuilderFactory.start());
    builder.setTailable(true);
    // builder.setAwaitData(false); // Uncomment if not using patch for SERVER-8602
    MongoIterator<Document> cursor = collection.find(builder.build());

    // Graceful shutdown of the iterator locally but not on the server.
    cursor.stop();
    while (cursor.hasNext()) {
        System.out.println(cursor.next());
    }

    collection = db.getCollection("lookup");
    collection.delete(BuilderFactory.start().add("_id", collectionName));
    collection.insert(BuilderFactory.start().add("_id", collectionName)
            .add("cursor", cursor.asDocument()));

    // Printout the cursor document.
    System.out.println("Queue created: " + collectionName);
    System.out.println(collection
            .findOne(BuilderFactory.start().add("_id", collectionName))
            .get("cursor").getValueAsObject());

    client.close();
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Create a new MongoDatabaseClient.
 *
 * @param mongoClient
 *            The {@link MongoClient}.
 * @param client
 *            The client for interacting with MongoDB.
 * @param name
 *            The name of the database we interact with.
 */
public MongoDatabaseImpl(final MongoClient mongoClient,
        final Client client, final String name) {
    myMongoClient = mongoClient;
    myClient = client;
    myName = name;
    myDurability = null;
    myReadPreference = null;
    myCollections = new ConcurrentHashMap<String, Reference<MongoCollection>>();
}
 

import com.allanbank.mongodb.MongoClient; //导入依赖的package包/类
/**
 * Creates a new Watcher.
 * 
 * @param mongoClient
 *            The client for the watcher.
 * @param collection
 *            The collection being watched.
 * @param context
 *            The context for the {@code _id} of items to watch.
 * @param listener
 *            The listener to notify that something has changed.
 */
public Watcher(final MongoClient mongoClient,
        final MongoCollection collection, final Pattern context,
        final WatchListener listener) {
    myMongoClient = mongoClient;
    myCollection = collection;
    myContext = context;
    myListener = listener;
    myControls = null;
    myLastTs = null;
}