#NEWLINE# def run( self ) : #NEWLINE# #INDENT# while True : #NEWLINE# #INDENT# eventlet . sleep ( 0 ) #NEWLINE# try : #NEWLINE# #INDENT# for data in self . pub_sub . listen ( ) : #NEWLINE# #INDENT# if ( 'subscribe' == data [ 'type' ] ) : #NEWLINE# #INDENT# continue #NEWLINE# #UNINDENT# if ( 'unsubscribe' == data [ 'type' ] ) : #NEWLINE# #INDENT# continue #NEWLINE# #UNINDENT# if ( 'message' == data [ 'type' ] ) : #NEWLINE# #INDENT# entry = pub_sub_api . unpack_message ( data [ 'data' ] ) #NEWLINE# entry_json = jsonutils . loads ( entry ) #NEWLINE# self . db_changes_callback ( entry_json [ 'table' ] , entry_json [ 'key' ] , entry_json [ 'action' ] , entry_json [ 'value' ] , entry_json [ 'topic' ] ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# except Exception as e : #NEWLINE# #INDENT# LOG . warning ( e ) #NEWLINE# try : #NEWLINE# #INDENT# connection = self . pub_sub . connection #NEWLINE# connection . connect ( ) #NEWLINE# self . db_changes_callback ( None , None , 'sync' , None , None ) #NEWLINE# #UNINDENT# except Exception as e : #NEWLINE# #INDENT# LOG . exception ( ( _LE ( 'reconnect error %(ip)s:%(port)s' ) % { #NEWLINE# #INDENT# 'ip' : self . ip , #NEWLINE# 'port' : self . plugin_updates_port , #NEWLINE# #UNINDENT# } ) )