(self)
Syncer main function.
Syncer main function.
Syncer main function .
def work(self):
        """Syncer main function."""

        # 'SELECT 1' and COPY must use same snapshot, so change isolation level.
        dst_db = self.get_database('db', isolation_level = skytools.I_REPEATABLE_READ)
        pnode, ploc = self.get_provider_location(dst_db)

        dst_tables, names = self.get_tables(dst_db)

        if len(self.args) > 2:
            tlist = self.args[2:]
        else:
            tlist = names

        for tbl in tlist:
            tbl = skytools.fq_name(tbl)
            if not tbl in dst_tables:
                self.log.warning('Table not subscribed: %s', tbl)
                continue
            t2 = dst_tables[tbl]
            if t2.merge_state != 'ok':
                self.log.warning('Table %s not synced yet, no point', tbl)
                continue

            pnode, ploc, wname = find_copy_source(self, self.queue_name, tbl, pnode, ploc)
            self.log.info('%s: Using node %s as provider', tbl, pnode)

            if wname is None:
                wname = self.consumer_name
            self.downstream_worker_name = wname

            self.process_one_table(tbl, t2, dst_db, pnode, ploc)

        # signal caller about bad tables
        sys.exit(self.bad_tables)
def work ( self ) : # 'SELECT 1' and COPY must use same snapshot, so change isolation level. dst_db = self . get_database ( 'db' , isolation_level = skytools . I_REPEATABLE_READ ) pnode , ploc = self . get_provider_location ( dst_db ) dst_tables , names = self . get_tables ( dst_db ) if len ( self . args ) > 2 : tlist = self . args [ 2 : ] else : tlist = names for tbl in tlist : tbl = skytools . fq_name ( tbl ) if not tbl in dst_tables : self . log . warning ( 'Table not subscribed: %s' , tbl ) continue t2 = dst_tables [ tbl ] if t2 . merge_state != 'ok' : self . log . warning ( 'Table %s not synced yet, no point' , tbl ) continue pnode , ploc , wname = find_copy_source ( self , self . queue_name , tbl , pnode , ploc ) self . log . info ( '%s: Using node %s as provider' , tbl , pnode ) if wname is None : wname = self . consumer_name self . downstream_worker_name = wname self . process_one_table ( tbl , t2 , dst_db , pnode , ploc ) # signal caller about bad tables sys . exit ( self . bad_tables )