(self, dst_db)
On root node send global watermark downstream.
On root node send global watermark downstream.
On root node send global watermark downstream .
def process_root_node(self, dst_db):
        """On root node send global watermark downstream.
        """

        CascadedConsumer.process_root_node(self, dst_db)

        t = time.time()
        if t - self.global_wm_publish_time < self.global_wm_publish_period:
            return

        self.log.debug("Publishing global watermark")
        dst_curs = dst_db.cursor()
        q = "select * from pgq_node.set_global_watermark(%s, NULL)"
        dst_curs.execute(q, [self.pgq_queue_name])
        dst_db.commit()
        self.global_wm_publish_time = t
def process_root_node ( self , dst_db ) : CascadedConsumer . process_root_node ( self , dst_db ) t = time . time ( ) if t - self . global_wm_publish_time < self . global_wm_publish_period : return self . log . debug ( "Publishing global watermark" ) dst_curs = dst_db . cursor ( ) q = "select * from pgq_node.set_global_watermark(%s, NULL)" dst_curs . execute ( q , [ self . pgq_queue_name ] ) dst_db . commit ( ) self . global_wm_publish_time = t