MongoDBOplogSourceIT.java 文件源码

@Before
public void createCollection() throws Exception {
  MongoDatabase db = mongoClient.getDatabase(DATABASE);
  testCollectionName = name.getMethodName();
  db.createCollection(testCollectionName);
  final long currentTime = System.currentTimeMillis();
  //To make sure that oplog is read on each method after we created the above collection.
  //We let this current second pass, before we get the initial timestamp seconds.
  Awaitility.await().untilTrue(new AtomicBoolean((System.currentTimeMillis() - currentTime) > 1000));
  //So we can skip old oplogs and just start with whatever this test is producing
  initialTs = getInitialTsFromCurrentTime();
  testDocuments = mongoClient.getDatabase(DATABASE).getCollection(testCollectionName);
  mongoCursorFindIterable = mongoClient.getDatabase("local").getCollection(OPLOG_COLLECTION)
      .find()
      //As the collection is a capped collection we use Tailable cursor which will return results in natural order in this case
      //based on ts timestamp field.
      //Tailable Await does not return and blocks, so we are using tailable.
      .cursorType(CursorType.Tailable);
}