'Dumps collection into the target system.
This method is called when we\'re initializing the cursor and have no
configs i.e. when we\'re starting for the first time.'

def dump_collection(self):

    timestamp = retry_until_ok(self.get_last_oplog_timestamp)
    if (timestamp is None):
        return None
    long_ts = util.bson_ts_to_long(timestamp)
    dump_cancelled = [False]
    def get_all_ns():
        ns_set = []
        gridfs_ns_set = []
        db_list = self.namespace_config.get_included_databases()
        if (not db_list):
            db_list = retry_until_ok(self.primary_client.database_names)
        for database in db_list:
            if ((database == 'config') or (database == 'local')):
                continue
            coll_list = retry_until_ok(self.primary_client[database].collection_names)
            for coll in coll_list:
                if coll.startswith('system.'):
                    continue
                if coll.endswith('.chunks'):
                    continue
                if coll.endswith('.files'):
                    namespace = ('%s.%s' % (database, coll))
                    namespace = namespace[:(- len('.files'))]
                    if self.namespace_config.gridfs_namespace(namespace):
                        gridfs_ns_set.append(namespace)
                else:
                    namespace = ('%s.%s' % (database, coll))
                    if self.namespace_config.map_namespace(namespace):
                        ns_set.append(namespace)
        return (ns_set, gridfs_ns_set)
    (dump_set, gridfs_dump_set) = get_all_ns()
    LOG.debug(('OplogThread:    Dumping    set    of    collections    %s    ' % dump_set))
    def docs_to_dump(from_coll):
        last_id = None
        attempts = 0
        projection = self.namespace_config.projection(from_coll.full_name)
        while (attempts < 60):
            if (last_id is None):
                cursor = retry_until_ok(from_coll.find, projection=projection, sort=[('_id', pymongo.ASCENDING)])
            else:
                cursor = retry_until_ok(from_coll.find, {'_id': {'$gt': last_id}}, projection=projection, sort=[('_id', pymongo.ASCENDING)])
            try:
                for doc in cursor:
                    if (not self.running):
                        dump_cancelled[0] = True
                        raise StopIteration
                    last_id = doc['_id']
                    (yield doc)
                break
            except (pymongo.errors.AutoReconnect, pymongo.errors.OperationFailure):
                attempts += 1
                time.sleep(1)
    def upsert_each(dm):
        num_failed = 0
        for namespace in dump_set:
            from_coll = self.get_collection(namespace)
            mapped_ns = self.namespace_config.map_namespace(namespace)
            total_docs = retry_until_ok(from_coll.count)
            num = None
            for (num, doc) in enumerate(docs_to_dump(from_coll)):
                try:
                    dm.upsert(doc, mapped_ns, long_ts)
                except Exception:
                    if self.continue_on_error:
                        LOG.exception(('Could    not    upsert    document:    %r' % doc))
                        num_failed += 1
                    else:
                        raise
                if ((num % 10000) == 0):
                    LOG.info("Upserted    %d    out    of    approximately    %d    docs    from    collection    '%s'", (num + 1), total_docs, namespace)
            if (num is not None):
                LOG.info("Upserted    %d    out    of    approximately    %d    docs    from    collection    '%s'", (num + 1), total_docs, namespace)
        if (num_failed > 0):
            LOG.error(('Failed    to    upsert    %d    docs' % num_failed))
    def upsert_all(dm):
        try:
            for namespace in dump_set:
                from_coll = self.get_collection(namespace)
                total_docs = retry_until_ok(from_coll.count)
                mapped_ns = self.namespace_config.map_namespace(namespace)
                LOG.info("Bulk    upserting    approximately    %d    docs    from    collection    '%s'", total_docs, namespace)
                dm.bulk_upsert(docs_to_dump(from_coll), mapped_ns, long_ts)
        except Exception:
            if self.continue_on_error:
                LOG.exception('OplogThread:    caught    exception    during    bulk    upsert,    re-upserting    documents    serially')
                upsert_each(dm)
            else:
                raise
    def do_dump(dm, error_queue):
        try:
            LOG.debug('OplogThread:    Using    bulk    upsert    function    for    collection    dump')
            upsert_all(dm)
            if gridfs_dump_set:
                LOG.info('OplogThread:    dumping    GridFS    collections:    %s', gridfs_dump_set)
            for gridfs_ns in gridfs_dump_set:
                mongo_coll = self.get_collection(gridfs_ns)
                from_coll = self.get_collection((gridfs_ns + '.files'))
                dest_ns = self.namespace_config.map_namespace(gridfs_ns)
                for doc in docs_to_dump(from_coll):
                    gridfile = GridFSFile(mongo_coll, doc)
                    dm.insert_file(gridfile, dest_ns, long_ts)
        except:
            error_queue.put(sys.exc_info())
    dumping_threads = []
    dump_success = True
    errors = queue.Queue()
    if (len(self.doc_managers) == 1):
        do_dump(self.doc_managers[0], errors)
    else:
        for dm in self.doc_managers:
            t = threading.Thread(target=do_dump, args=(dm, errors))
            dumping_threads.append(t)
            t.start()
        for t in dumping_threads:
            t.join()
    try:
        while True:
            LOG.critical('Exception    during    collection    dump', exc_info=errors.get_nowait())
            dump_success = False
    except queue.Empty:
        pass
    if (not dump_success):
        err_msg = 'OplogThread:    Failed    during    dump    collection'
        effect = 'cannot    recover!'
        LOG.error(('%s    %s    %s' % (err_msg, effect, self.oplog)))
        self.running = False
        return None
    if dump_cancelled[0]:
        LOG.warning('Initial    collection    dump    was    interrupted.    Will    re-run    the    collection    dump    on    next    startup.')
        return None
    return timestamp
