(self, dst_curs, ev, filtered_copy)
Add event to copy buffer.
Add event to copy buffer.
Add event to copy buffer .
def copy_event(self, dst_curs, ev, filtered_copy):
        """Add event to copy buffer.
        """
        if not self.main_worker:
            return
        if filtered_copy:
            if ev.type[:4] == "pgq.":
                return
        if len(self.ev_buf) >= self.max_evbuf:
            self.flush_events(dst_curs)

        if ev.type == 'pgq.global-watermark':
            st = self._worker_state
            if st.sync_watermark:
                # replace payload with synced global watermark
                row = ev._event_row.copy()
                row['ev_data'] = str(st.global_watermark)
                ev = Event(self.queue_name, row)
        self.ev_buf.append(ev)
def copy_event ( self , dst_curs , ev , filtered_copy ) : if not self . main_worker : return if filtered_copy : if ev . type [ : 4 ] == "pgq." : return if len ( self . ev_buf ) >= self . max_evbuf : self . flush_events ( dst_curs ) if ev . type == 'pgq.global-watermark' : st = self . _worker_state if st . sync_watermark : # replace payload with synced global watermark row = ev . _event_row . copy ( ) row [ 'ev_data' ] = str ( st . global_watermark ) ev = Event ( self . queue_name , row ) self . ev_buf . append ( ev )