(self, cstr1, cstr2, queue_name, consumer_name, table_name)
return (src_db, dst_db)
Syncer main function.

        Returns (src_db, dst_db) that are in transaction
        where table should be in sync.
Syncer main function.
Syncer main function .
def sync_table(self, cstr1, cstr2, queue_name, consumer_name, table_name):
        """Syncer main function.

        Returns (src_db, dst_db) that are in transaction
        where table should be in sync.
        """

        setup_db = self.get_database('setup_db', connstr = cstr1, autocommit = 1)
        lock_db = self.get_database('lock_db', connstr = cstr1)

        src_db = self.get_database('src_db', connstr = cstr1,
                isolation_level = skytools.I_REPEATABLE_READ)
        dst_db = self.get_database('dst_db', connstr = cstr2,
                isolation_level = skytools.I_REPEATABLE_READ)

        lock_curs = lock_db.cursor()
        setup_curs = setup_db.cursor()
        src_curs = src_db.cursor()
        dst_curs = dst_db.cursor()

        self.check_consumer(setup_curs, queue_name, consumer_name)

        # lock table in separate connection
        self.log.info('Locking %s' % table_name)
        self.set_lock_timeout(lock_curs)
        lock_time = time.time()
        lock_curs.execute("LOCK TABLE %s IN SHARE MODE" % skytools.quote_fqident(table_name))

        # now wait until consumer has updated target table until locking
        self.log.info('Syncing %s' % table_name)

        # consumer must get further than this tick
        tick_id = self.force_tick(setup_curs, queue_name)
        # try to force second tick also
        self.force_tick(setup_curs, queue_name)

        # take server time
        setup_curs.execute("select to_char(now(), 'YYYY-MM-DD HH24:MI:SS.MS')")
        tpos = setup_curs.fetchone()[0]
        # now wait
        while 1:
            time.sleep(0.5)

            q = "select now() - lag > timestamp %s, now(), lag from pgq.get_consumer_info(%s, %s)"
            setup_curs.execute(q, [tpos, queue_name, consumer_name])
            res = setup_curs.fetchall()
            if len(res) == 0:
                raise Exception('No such consumer: %s/%s' % (queue_name, consumer_name))
            row = res[0]
            self.log.debug("tpos=%s now=%s lag=%s ok=%s" % (tpos, row[1], row[2], row[0]))
            if row[0]:
                break

            # limit lock time
            if time.time() > lock_time + self.lock_timeout:
                self.log.error('Consumer lagging too much, exiting')
                lock_db.rollback()
                sys.exit(1)

        # take snapshot on provider side
        src_db.commit()
        src_curs.execute("SELECT 1")

        # take snapshot on subscriber side
        dst_db.commit()
        dst_curs.execute("SELECT 1")

        # release lock
        lock_db.commit()

        self.close_database('setup_db')
        self.close_database('lock_db')

        return (src_db, dst_db)
def sync_table ( self , cstr1 , cstr2 , queue_name , consumer_name , table_name ) : setup_db = self . get_database ( 'setup_db' , connstr = cstr1 , autocommit = 1 ) lock_db = self . get_database ( 'lock_db' , connstr = cstr1 ) src_db = self . get_database ( 'src_db' , connstr = cstr1 , isolation_level = skytools . I_REPEATABLE_READ ) dst_db = self . get_database ( 'dst_db' , connstr = cstr2 , isolation_level = skytools . I_REPEATABLE_READ ) lock_curs = lock_db . cursor ( ) setup_curs = setup_db . cursor ( ) src_curs = src_db . cursor ( ) dst_curs = dst_db . cursor ( ) self . check_consumer ( setup_curs , queue_name , consumer_name ) # lock table in separate connection self . log . info ( 'Locking %s' % table_name ) self . set_lock_timeout ( lock_curs ) lock_time = time . time ( ) lock_curs . execute ( "LOCK TABLE %s IN SHARE MODE" % skytools . quote_fqident ( table_name ) ) # now wait until consumer has updated target table until locking self . log . info ( 'Syncing %s' % table_name ) # consumer must get further than this tick tick_id = self . force_tick ( setup_curs , queue_name ) # try to force second tick also self . force_tick ( setup_curs , queue_name ) # take server time setup_curs . execute ( "select to_char(now(), 'YYYY-MM-DD HH24:MI:SS.MS')" ) tpos = setup_curs . fetchone ( ) [ 0 ] # now wait while 1 : time . sleep ( 0.5 ) q = "select now() - lag > timestamp %s, now(), lag from pgq.get_consumer_info(%s, %s)" setup_curs . execute ( q , [ tpos , queue_name , consumer_name ] ) res = setup_curs . fetchall ( ) if len ( res ) == 0 : raise Exception ( 'No such consumer: %s/%s' % ( queue_name , consumer_name ) ) row = res [ 0 ] self . log . debug ( "tpos=%s now=%s lag=%s ok=%s" % ( tpos , row [ 1 ] , row [ 2 ] , row [ 0 ] ) ) if row [ 0 ] : break # limit lock time if time . time ( ) > lock_time + self . lock_timeout : self . log . error ( 'Consumer lagging too much, exiting' ) lock_db . rollback ( ) sys . exit ( 1 ) # take snapshot on provider side src_db . commit ( ) src_curs . execute ( "SELECT 1" ) # take snapshot on subscriber side dst_db . commit ( ) dst_curs . execute ( "SELECT 1" ) # release lock lock_db . commit ( ) self . close_database ( 'setup_db' ) self . close_database ( 'lock_db' ) return ( src_db , dst_db )