<s> from frontera import Backend <EOL> from frontera . core import OverusedBuffer <EOL> from codecs . msgpack import Encoder , Decoder <EOL> from frontera . utils . misc import load_object <EOL> class MessageBusBackend ( Backend ) : <EOL> def __init__ ( self , manager ) : <EOL> self . _manager = manager <EOL> settings = self . _manager . settings <EOL> messagebus = load_object ( settings . get ( '<STR_LIT>' ) ) <EOL> self . mb = messagebus ( settings ) <EOL> store_content = settings . get ( '<STR_LIT>' ) <EOL> self . _encoder = Encoder ( manager . request_model , send_body = store_content ) <EOL> self . _decoder = Decoder ( manager . request_model , manager . response_model ) <EOL> self . spider_log_producer = self . mb . spider_log ( ) . producer ( ) <EOL> spider_feed = self . mb . spider_feed ( ) <EOL> self . partition_id = int ( settings . get ( '<STR_LIT>' ) ) <EOL> self . consumer = spider_feed . consumer ( partition_id = self . partition_id ) <EOL> self . _get_timeout = float ( settings . get ( '<STR_LIT>' ) ) <EOL> self . _buffer = OverusedBuffer ( self . _get_next_requests , <EOL> manager . logger . manager . debug ) <EOL> @ classmethod <EOL> def from_manager ( clas , manager ) : <EOL> return clas ( manager ) <EOL> def frontier_start ( self ) : <EOL> pass <EOL> def frontier_stop ( self ) : <EOL> self . spider_log_producer . flush ( ) <EOL> def add_seeds ( self , seeds ) : <EOL> self . spider_log_producer . send ( seeds [ <NUM_LIT:0> ] . meta [ '<STR_LIT>' ] , self . _encoder . encode_add_seeds ( seeds ) ) <EOL> def page_crawled ( self , response , links ) : <EOL> self . spider_log_producer . send ( response . meta [ '<STR_LIT>' ] , self . _encoder . encode_page_crawled ( response , links ) ) <EOL> def request_error ( self , page , error ) : <EOL> self . spider_log_producer . send ( page . meta [ '<STR_LIT>' ] , self . _encoder . encode_request_error ( page , error ) ) <EOL> def _get_next_requests ( self , max_n_requests , ** kwargs ) : <EOL> requests = [ ] <EOL> for encoded in self . consumer . get_messages ( count = max_n_requests , timeout = self . _get_timeout ) : <EOL> try : <EOL> request = self . _decoder . decode_request ( encoded ) <EOL> except Exception , exc : <EOL> self . _manager . logger . backend . warning ( "<STR_LIT>" . format ( encoded , <EOL> str ( exc ) ) ) <EOL> else : <EOL> requests . append ( request ) <EOL> self . spider_log_producer . send ( '<STR_LIT>' , <EOL> self . _encoder . encode_offset ( self . partition_id , self . consumer . get_offset ( ) ) ) <EOL> return requests <EOL> def get_next_requests ( self , max_n_requests , ** kwargs ) : <EOL> return self . _buffer . get_next_requests ( max_n_requests , ** kwargs ) <EOL> def finished ( self ) : <EOL> return False <EOL> @ property <EOL> def metadata ( self ) : <EOL> return None <EOL> @ property <EOL> def queue ( self ) : <EOL> return None <EOL> @ property <EOL> def states ( self ) : <EOL> return None </s>
