(self, provider_loc = None)
Register consumer on source node first, then target node.
Register consumer on source node first, then target node.
Register consumer on source node first then target node .
def register_consumer(self, provider_loc = None):
        """Register consumer on source node first, then target node."""

        if not provider_loc:
            provider_loc = self.options.provider
        if not provider_loc:
            self.log.error('Please give provider location with --provider=')
            sys.exit(1)

        dst_db = self.get_database(self.target_db)
        dst_curs = dst_db.cursor()
        src_db = self.get_database(PDB, connstr = provider_loc, profile = 'remote')
        src_curs = src_db.cursor()

        # check target info
        q = "select * from pgq_node.get_node_info(%s)"
        res = self.exec_cmd(src_db, q, [ self.queue_name ])
        pnode = res[0]['node_name']
        if not pnode:
            raise Exception('parent node not initialized?')

        # source queue
        BaseConsumer.register_consumer(self)

        # fetch pos
        q = "select last_tick from pgq.get_consumer_info(%s, %s)"
        src_curs.execute(q, [self.queue_name, self.consumer_name])
        last_tick = src_curs.fetchone()['last_tick']
        if not last_tick:
            raise Exception('registration failed?')
        src_db.commit()

        # target node
        q = "select * from pgq_node.register_consumer(%s, %s, %s, %s)"
        self.exec_cmd(dst_db, q, [self.queue_name, self.consumer_name, pnode, last_tick])
def register_consumer ( self , provider_loc = None ) : if not provider_loc : provider_loc = self . options . provider if not provider_loc : self . log . error ( 'Please give provider location with --provider=' ) sys . exit ( 1 ) dst_db = self . get_database ( self . target_db ) dst_curs = dst_db . cursor ( ) src_db = self . get_database ( PDB , connstr = provider_loc , profile = 'remote' ) src_curs = src_db . cursor ( ) # check target info q = "select * from pgq_node.get_node_info(%s)" res = self . exec_cmd ( src_db , q , [ self . queue_name ] ) pnode = res [ 0 ] [ 'node_name' ] if not pnode : raise Exception ( 'parent node not initialized?' ) # source queue BaseConsumer . register_consumer ( self ) # fetch pos q = "select last_tick from pgq.get_consumer_info(%s, %s)" src_curs . execute ( q , [ self . queue_name , self . consumer_name ] ) last_tick = src_curs . fetchone ( ) [ 'last_tick' ] if not last_tick : raise Exception ( 'registration failed?' ) src_db . commit ( ) # target node q = "select * from pgq_node.register_consumer(%s, %s, %s, %s)" self . exec_cmd ( dst_db , q , [ self . queue_name , self . consumer_name , pnode , last_tick ] )