<s> import traceback <EOL> import eventlet <EOL> from eventlet import event <EOL> from . base import check_callable <EOL> from . sync import SyncConsumer <EOL> from . . utils import json <EOL> class ChangeConsumer ( object ) : <EOL> def __init__ ( self , db , callback , ** params ) : <EOL> self . process_change = callback <EOL> self . params = params <EOL> self . db = db <EOL> self . stop_event = event . Event ( ) <EOL> def wait ( self ) : <EOL> eventlet . spawn_n ( self . _run ) <EOL> self . stop_event . wait ( ) <EOL> def wait_async ( self ) : <EOL> eventlet . spawn_n ( self . _run ) <EOL> def _run ( self ) : <EOL> while True : <EOL> try : <EOL> resp = self . db . res . get ( "<STR_LIT>" , ** self . params ) <EOL> return self . consume ( resp ) <EOL> except ( SystemExit , KeyboardInterrupt ) : <EOL> eventlet . sleep ( <NUM_LIT:5> ) <EOL> break <EOL> except : <EOL> traceback . print_exc ( ) <EOL> eventlet . sleep ( <NUM_LIT:5> ) <EOL> break <EOL> self . stop_event . send ( True ) <EOL> def consume ( self , resp ) : <EOL> raise NotImplementedError <EOL> class ContinuousChangeConsumer ( ChangeConsumer ) : <EOL> def consume ( self , resp ) : <EOL> with resp . body_stream ( ) as body : <EOL> while True : <EOL> line = body . readline ( ) <EOL> if not line : <EOL> break <EOL> if line . endswith ( "<STR_LIT:\r\n>" ) : <EOL> line = line [ : - <NUM_LIT:2> ] <EOL> else : <EOL> line = line [ : - <NUM_LIT:1> ] <EOL> if not line : <EOL> continue <EOL> self . process_change ( line ) <EOL> self . stop_event . send ( True ) <EOL> class LongPollChangeConsumer ( ChangeConsumer ) : <EOL> def consume ( self , resp ) : <EOL> with resp . body_stream ( ) as body : <EOL> buf = [ ] <EOL> while True : <EOL> data = body . read ( ) <EOL> if not data : <EOL> break <EOL> buf . append ( data ) <EOL> change = "<STR_LIT>" . join ( buf ) <EOL> try : <EOL> change = json . loads ( change ) <EOL> except ValueError : <EOL> pass <EOL> self . process_change ( change ) <EOL> self . stop_event . send ( True ) <EOL> class EventletConsumer ( SyncConsumer ) : <EOL> def __init__ ( self , db ) : <EOL> eventlet . monkey_patch ( socket = True ) <EOL> super ( EventletConsumer , self ) . __init__ ( db ) <EOL> def _fetch ( self , cb , ** params ) : <EOL> resp = self . db . res . get ( "<STR_LIT>" , ** params ) <EOL> cb ( resp . json_body ) <EOL> def fetch ( self , cb = None , ** params ) : <EOL> if cb is None : <EOL> return super ( EventletConsumer , self ) . wait_once ( ** params ) <EOL> eventlet . spawn_n ( self . _fetch , cb , ** params ) <EOL> def wait_once ( self , cb = None , ** params ) : <EOL> if cb is None : <EOL> return super ( EventletConsumer , self ) . wait_once ( ** params ) <EOL> check_callable ( cb ) <EOL> params . update ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> consumer = LongPollChangeConsumer ( self . db , callback = cb , <EOL> ** params ) <EOL> consumer . wait ( ) <EOL> def wait ( self , cb , ** params ) : <EOL> params . update ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> consumer = ContinuousChangeConsumer ( self . db , callback = cb , <EOL> ** params ) <EOL> consumer . wait ( ) <EOL> def wait_once_async ( self , cb , ** params ) : <EOL> check_callable ( cb ) <EOL> params . update ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> consumer = LongPollChangeConsumer ( self . db , callback = cb , <EOL> ** params ) <EOL> return consumer . wait_async ( ) <EOL> def wait_async ( self , cb , ** params ) : <EOL> check_callable ( cb ) <EOL> params . update ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> consumer = ContinuousChangeConsumer ( self . db , callback = cb , <EOL> ** params ) <EOL> return consumer . wait_async ( ) </s>
