'Start the oplog worker.'

@log_fatal_exceptions
def run(self):

    ReplicationLagLogger(self, 30).start()
    LOG.debug('OplogThread:    Run    thread    started')
    while (self.running is True):
        LOG.debug('OplogThread:    Getting    cursor')
        (cursor, cursor_empty) = retry_until_ok(self.init_cursor)
        if ((cursor is None) and (self.checkpoint is not None)):
            err_msg = 'OplogThread:    Last    entry    no    longer    in    oplog'
            effect = 'cannot    recover!'
            LOG.error(('%s    %s    %s' % (err_msg, effect, self.oplog)))
            self.running = False
            continue
        if cursor_empty:
            LOG.debug('OplogThread:    Last    entry    is    the    one    we    already    processed.        Up    to    date.        Sleeping.')
            time.sleep(1)
            continue
        last_ts = None
        remove_inc = 0
        upsert_inc = 0
        update_inc = 0
        try:
            LOG.debug('OplogThread:    about    to    process    new    oplog    entries')
            while (cursor.alive and self.running):
                LOG.debug('OplogThread:    Cursor    is    still    alive    and    thread    is    still    running.')
                for (n, entry) in enumerate(cursor):
                    if (not self.running):
                        break
                    LOG.debug(('OplogThread:    Iterating    through    cursor,    document    number    in    this    cursor    is    %d' % n))
                    (skip, is_gridfs_file) = self._should_skip_entry(entry)
                    if skip:
                        last_ts = entry['ts']
                        continue
                    operation = entry['op']
                    ns = entry['ns']
                    timestamp = util.bson_ts_to_long(entry['ts'])
                    for docman in self.doc_managers:
                        try:
                            LOG.debug(('OplogThread:    Operation    for    this    entry    is    %s' % str(operation)))
                            if (operation == 'd'):
                                docman.remove(entry['o']['_id'], ns, timestamp)
                                remove_inc += 1
                            elif (operation == 'i'):
                                doc = entry.get('o')
                                if is_gridfs_file:
                                    (db, coll) = ns.split('.', 1)
                                    gridfile = GridFSFile(self.primary_client[db][coll], doc)
                                    docman.insert_file(gridfile, ns, timestamp)
                                else:
                                    docman.upsert(doc, ns, timestamp)
                                upsert_inc += 1
                            elif (operation == 'u'):
                                docman.update(entry['o2']['_id'], entry['o'], ns, timestamp)
                                update_inc += 1
                            elif (operation == 'c'):
                                doc = entry.get('o')
                                docman.handle_command(doc, entry['ns'], timestamp)
                        except errors.OperationFailed:
                            LOG.exception(('Unable    to    process    oplog    document    %r' % entry))
                        except errors.ConnectionFailed:
                            LOG.exception(('Connection    failed    while    processing    oplog    document    %r' % entry))
                    if ((((remove_inc + upsert_inc) + update_inc) % 1000) == 0):
                        LOG.debug(('OplogThread:    Documents    removed:    %d,    inserted:    %d,    updated:    %d    so    far' % (remove_inc, upsert_inc, update_inc)))
                    LOG.debug('OplogThread:    Doc    is    processed.')
                    last_ts = entry['ts']
                    if ((n % self.batch_size) == 1):
                        self.update_checkpoint(last_ts)
                        last_ts = None
                if (last_ts is not None):
                    LOG.debug('OplogThread:    updating    checkpoint    after    processing    new    oplog    entries')
                    self.update_checkpoint(last_ts)
        except (pymongo.errors.AutoReconnect, pymongo.errors.OperationFailure, pymongo.errors.ConfigurationError):
            LOG.exception('Cursor    closed    due    to    an    exception.    Will    attempt    to    reconnect.')
        if (last_ts is not None):
            LOG.debug('OplogThread:    updating    checkpoint    after    an    Exception,    cursor    closing,    or    join()    on    thisthread.')
            self.update_checkpoint(last_ts)
        LOG.debug(('OplogThread:    Sleeping.    Documents    removed:    %d,    upserted:    %d,    updated:    %d' % (remove_inc, upsert_inc, update_inc)))
        time.sleep(2)
