(self, src_db, tick_id, event_list, dst_db)
Worker-specific event processing.
Worker-specific event processing.
Worker - specific event processing .
def process_remote_batch(self, src_db, tick_id, event_list, dst_db):
        """Worker-specific event processing."""
        self.ev_buf = []
        max_id = 0
        st = self._worker_state

        if st.wait_behind:
            self.wait_for_tick(dst_db, tick_id)

        src_curs = src_db.cursor()
        dst_curs = dst_db.cursor()
        for ev in event_list:
            if st.copy_events:
                self.copy_event(dst_curs, ev, st.filtered_copy)
            if ev.ev_type.split('.', 1)[0] in ("pgq", "londiste"):
                # process cascade events even on waiting leaf node
                self.process_remote_event(src_curs, dst_curs, ev)
            else:
                if st.process_events:
                    self.process_remote_event(src_curs, dst_curs, ev)
            if ev.ev_id > max_id:
                max_id = ev.ev_id
        if max_id > self.cur_max_id:
            self.cur_max_id = max_id
def process_remote_batch ( self , src_db , tick_id , event_list , dst_db ) : self . ev_buf = [ ] max_id = 0 st = self . _worker_state if st . wait_behind : self . wait_for_tick ( dst_db , tick_id ) src_curs = src_db . cursor ( ) dst_curs = dst_db . cursor ( ) for ev in event_list : if st . copy_events : self . copy_event ( dst_curs , ev , st . filtered_copy ) if ev . ev_type . split ( '.' , 1 ) [ 0 ] in ( "pgq" , "londiste" ) : # process cascade events even on waiting leaf node self . process_remote_event ( src_curs , dst_curs , ev ) else : if st . process_events : self . process_remote_event ( src_curs , dst_curs , ev ) if ev . ev_id > max_id : max_id = ev . ev_id if max_id > self . cur_max_id : self . cur_max_id = max_id