(self, src_curs, dst_curs, ev)
Handle cascading events.
Handle cascading events.
Handle cascading events .
def process_remote_event(self, src_curs, dst_curs, ev):
        """Handle cascading events.
        """

        if ev.retry:
            raise Exception('CascadedWorker must not get retry events')

        # non cascade events send to CascadedConsumer to error out
        if ev.ev_type[:4] != 'pgq.':
            CascadedConsumer.process_remote_event(self, src_curs, dst_curs, ev)
            return

        # ignore cascade events if not main worker
        if not self.main_worker:
            return

        # check if for right queue
        t = ev.ev_type
        if ev.ev_extra1 != self.pgq_queue_name and t != "pgq.tick-id":
            raise Exception("bad event in queue: "+str(ev))

        self.log.debug("got cascade event: %s(%s)", t, ev.ev_data)
        st = self._worker_state
        if t == "pgq.location-info":
            node = ev.ev_data
            loc = ev.ev_extra2
            dead = ev.ev_extra3
            q = "select * from pgq_node.register_location(%s, %s, %s, %s)"
            dst_curs.execute(q, [self.pgq_queue_name, node, loc, dead])
        elif t == "pgq.unregister-location":
            node = ev.ev_data
            q = "select * from pgq_node.unregister_location(%s, %s)"
            dst_curs.execute(q, [self.pgq_queue_name, node])
        elif t == "pgq.global-watermark":
            if st.sync_watermark:
                tick_id = int(ev.ev_data)
                self.log.debug('Half-ignoring global watermark %d', tick_id)
                self.real_global_wm = tick_id
            elif st.process_global_wm:
                tick_id = int(ev.ev_data)
                q = "select * from pgq_node.set_global_watermark(%s, %s)"
                dst_curs.execute(q, [self.pgq_queue_name, tick_id])
        elif t == "pgq.tick-id":
            tick_id = int(ev.ev_data)
            if ev.ev_extra1 == self.pgq_queue_name:
                raise Exception('tick-id event for own queue?')
            if st.process_tick_event:
                q = "select * from pgq_node.set_partition_watermark(%s, %s, %s)"
                dst_curs.execute(q, [self.pgq_queue_name, ev.ev_extra1, tick_id])
        else:
            raise Exception("unknown cascade event: %s" % t)
def process_remote_event ( self , src_curs , dst_curs , ev ) : if ev . retry : raise Exception ( 'CascadedWorker must not get retry events' ) # non cascade events send to CascadedConsumer to error out if ev . ev_type [ : 4 ] != 'pgq.' : CascadedConsumer . process_remote_event ( self , src_curs , dst_curs , ev ) return # ignore cascade events if not main worker if not self . main_worker : return # check if for right queue t = ev . ev_type if ev . ev_extra1 != self . pgq_queue_name and t != "pgq.tick-id" : raise Exception ( "bad event in queue: " + str ( ev ) ) self . log . debug ( "got cascade event: %s(%s)" , t , ev . ev_data ) st = self . _worker_state if t == "pgq.location-info" : node = ev . ev_data loc = ev . ev_extra2 dead = ev . ev_extra3 q = "select * from pgq_node.register_location(%s, %s, %s, %s)" dst_curs . execute ( q , [ self . pgq_queue_name , node , loc , dead ] ) elif t == "pgq.unregister-location" : node = ev . ev_data q = "select * from pgq_node.unregister_location(%s, %s)" dst_curs . execute ( q , [ self . pgq_queue_name , node ] ) elif t == "pgq.global-watermark" : if st . sync_watermark : tick_id = int ( ev . ev_data ) self . log . debug ( 'Half-ignoring global watermark %d' , tick_id ) self . real_global_wm = tick_id elif st . process_global_wm : tick_id = int ( ev . ev_data ) q = "select * from pgq_node.set_global_watermark(%s, %s)" dst_curs . execute ( q , [ self . pgq_queue_name , tick_id ] ) elif t == "pgq.tick-id" : tick_id = int ( ev . ev_data ) if ev . ev_extra1 == self . pgq_queue_name : raise Exception ( 'tick-id event for own queue?' ) if st . process_tick_event : q = "select * from pgq_node.set_partition_watermark(%s, %s, %s)" dst_curs . execute ( q , [ self . pgq_queue_name , ev . ev_extra1 , tick_id ] ) else : raise Exception ( "unknown cascade event: %s" % t )