(self, curs, queue_name, consumer_name)
Before locking anything check if consumer is working ok.
Before locking anything check if consumer is working ok.
Before locking anything check if consumer is working ok .
def check_consumer(self, curs, queue_name, consumer_name):
        """ Before locking anything check if consumer is working ok.
        """
        self.log.info("Queue: %s Consumer: %s" % (queue_name, consumer_name))

        curs.execute('select current_database()')
        self.log.info('Actual db: %s' % curs.fetchone()[0])

        # get ticker lag
        q = "select extract(epoch from ticker_lag) from pgq.get_queue_info(%s);"
        curs.execute(q, [queue_name])
        ticker_lag = curs.fetchone()[0]
        self.log.info("Ticker lag: %s" % ticker_lag)
        # get consumer lag
        q = "select extract(epoch from lag) from pgq.get_consumer_info(%s, %s);"
        curs.execute(q, [queue_name, consumer_name])
        res = curs.fetchall()
        if len(res) == 0:
            self.log.error('check_consumer: No such consumer: %s/%s' % (queue_name, consumer_name))
            sys.exit(1)
        consumer_lag = res[0][0]

        # check that lag is acceptable
        self.log.info("Consumer lag: %s" % consumer_lag)
        if consumer_lag > ticker_lag + 10:
            self.log.error('Consumer lagging too much, cannot proceed')
            sys.exit(1)
def check_consumer ( self , curs , queue_name , consumer_name ) : self . log . info ( "Queue: %s Consumer: %s" % ( queue_name , consumer_name ) ) curs . execute ( 'select current_database()' ) self . log . info ( 'Actual db: %s' % curs . fetchone ( ) [ 0 ] ) # get ticker lag q = "select extract(epoch from ticker_lag) from pgq.get_queue_info(%s);" curs . execute ( q , [ queue_name ] ) ticker_lag = curs . fetchone ( ) [ 0 ] self . log . info ( "Ticker lag: %s" % ticker_lag ) # get consumer lag q = "select extract(epoch from lag) from pgq.get_consumer_info(%s, %s);" curs . execute ( q , [ queue_name , consumer_name ] ) res = curs . fetchall ( ) if len ( res ) == 0 : self . log . error ( 'check_consumer: No such consumer: %s/%s' % ( queue_name , consumer_name ) ) sys . exit ( 1 ) consumer_lag = res [ 0 ] [ 0 ] # check that lag is acceptable self . log . info ( "Consumer lag: %s" % consumer_lag ) if consumer_lag > ticker_lag + 10 : self . log . error ( 'Consumer lagging too much, cannot proceed' ) sys . exit ( 1 )