<s> import logging <EOL> import time <EOL> import threading <EOL> import kombu <EOL> import socket <EOL> from umbra . browser import BrowserPool , BrowsingException <EOL> class AmqpBrowserController : <EOL> """<STR_LIT>""" <EOL> logger = logging . getLogger ( __module__ + "<STR_LIT:.>" + __qualname__ ) <EOL> def __init__ ( self , amqp_url = '<STR_LIT>' , <EOL> chrome_exe = '<STR_LIT>' , max_active_browsers = <NUM_LIT:1> , <EOL> queue_name = '<STR_LIT>' , exchange_name = '<STR_LIT>' , routing_key = '<STR_LIT>' ) : <EOL> self . amqp_url = amqp_url <EOL> self . queue_name = queue_name <EOL> self . exchange_name = exchange_name <EOL> self . routing_key = routing_key <EOL> self . max_active_browsers = max_active_browsers <EOL> self . _browser_pool = BrowserPool ( size = max_active_browsers , chrome_exe = chrome_exe ) <EOL> def start ( self ) : <EOL> self . _browsing_threads = set ( ) <EOL> self . _browsing_threads_lock = threading . Lock ( ) <EOL> self . _exchange = kombu . Exchange ( name = self . exchange_name , type = '<STR_LIT>' , <EOL> durable = True ) <EOL> self . _reconnect_requested = False <EOL> self . _producer = None <EOL> self . _producer_lock = threading . Lock ( ) <EOL> with self . _producer_lock : <EOL> self . _producer_conn = kombu . Connection ( self . amqp_url ) <EOL> self . _producer = self . _producer_conn . Producer ( serializer = '<STR_LIT>' ) <EOL> self . _consumer_thread = threading . Thread ( target = self . _consume_amqp , name = '<STR_LIT>' ) <EOL> self . _consumer_stop = threading . Event ( ) <EOL> self . _consumer_thread . start ( ) <EOL> def shutdown ( self ) : <EOL> self . logger . info ( "<STR_LIT>" . format ( self . amqp_url ) ) <EOL> self . _consumer_stop . set ( ) <EOL> self . _consumer_thread . join ( ) <EOL> def shutdown_now ( self ) : <EOL> self . _consumer_stop . set ( ) <EOL> self . _browser_pool . shutdown_now ( ) <EOL> self . _consumer_thread . join ( ) <EOL> def reconnect ( self , * args , ** kwargs ) : <EOL> self . _reconnect_requested = True <EOL> self . _browser_pool . shutdown_now ( ) <EOL> def _wait_for_and_browse_urls ( self , conn , consumer , timeout ) : <EOL> start = time . time ( ) <EOL> browser = None <EOL> consumer . qos ( prefetch_count = self . max_active_browsers ) <EOL> while not self . _consumer_stop . is_set ( ) and time . time ( ) - start < timeout and not self . _reconnect_requested : <EOL> try : <EOL> browser = self . _browser_pool . acquire ( ) <EOL> browser . start ( ) <EOL> def callback ( body , message ) : <EOL> try : <EOL> client_id = body . get ( '<STR_LIT>' ) <EOL> url = body [ '<STR_LIT:url>' ] <EOL> metadata = body . get ( '<STR_LIT>' ) <EOL> behavior_parameters = body . get ( '<STR_LIT>' ) <EOL> except : <EOL> self . logger . error ( "<STR_LIT>" , <EOL> message , exc_info = True ) <EOL> self . logger . error ( "<STR_LIT>" ) <EOL> message . reject ( ) <EOL> browser . stop ( ) <EOL> self . _browser_pool . release ( browser ) <EOL> return <EOL> self . _start_browsing_page ( browser , message , client_id , url , <EOL> metadata , behavior_parameters ) <EOL> consumer . callbacks = [ callback ] <EOL> while True : <EOL> try : <EOL> conn . drain_events ( timeout = <NUM_LIT:0.5> ) <EOL> break <EOL> except socket . timeout : <EOL> pass <EOL> except socket . error : <EOL> self . logger . error ( "<STR_LIT>" , exc_info = True ) <EOL> self . _reconnect_requested = True <EOL> if self . _consumer_stop . is_set ( ) or time . time ( ) - start >= timeout or self . _reconnect_requested : <EOL> browser . stop ( ) <EOL> self . _browser_pool . release ( browser ) <EOL> break <EOL> except KeyError : <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> except : <EOL> self . logger . critical ( "<STR_LIT>" , exc_info = True ) <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> finally : <EOL> consumer . callbacks = None <EOL> def _wait_for_active_browsers ( self ) : <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> while True : <EOL> with self . _browsing_threads_lock : <EOL> if len ( self . _browsing_threads ) == <NUM_LIT:0> : <EOL> break <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> def _consume_amqp ( self ) : <EOL> RECONNECT_AFTER_SECONDS = <NUM_LIT> * <NUM_LIT> <EOL> url_queue = kombu . Queue ( self . queue_name , exchange = self . _exchange , routing_key = self . routing_key ) <EOL> while not self . _consumer_stop . is_set ( ) : <EOL> try : <EOL> self . logger . info ( "<STR_LIT>" . format ( self . _exchange . name , self . amqp_url ) ) <EOL> self . _reconnect_requested = False <EOL> with kombu . Connection ( self . amqp_url ) as conn : <EOL> with conn . Consumer ( url_queue ) as consumer : <EOL> self . _wait_for_and_browse_urls ( conn , consumer , timeout = RECONNECT_AFTER_SECONDS ) <EOL> self . _wait_for_active_browsers ( ) <EOL> except BaseException as e : <EOL> self . logger . error ( "<STR_LIT>" . format ( e ) , exc_info = True ) <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> self . logger . error ( "<STR_LIT>" ) <EOL> def _start_browsing_page ( self , browser , message , client_id , url , parent_url_metadata , behavior_parameters ) : <EOL> def on_response ( chrome_msg ) : <EOL> if ( chrome_msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:url>' ] . lower ( ) . startswith ( '<STR_LIT>' ) <EOL> or chrome_msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> or not '<STR_LIT>' in chrome_msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) : <EOL> return <EOL> payload = { <EOL> '<STR_LIT:url>' : chrome_msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:url>' ] , <EOL> '<STR_LIT>' : chrome_msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : url , <EOL> '<STR_LIT>' : parent_url_metadata , <EOL> } <EOL> if '<STR_LIT>' in chrome_msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] : <EOL> payload [ '<STR_LIT>' ] = chrome_msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> elif '<STR_LIT>' in chrome_msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] : <EOL> req = chrome_msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> payload [ '<STR_LIT>' ] = req [ : req . index ( '<STR_LIT:U+0020>' ) ] <EOL> else : <EOL> self . logger . warn ( '<STR_LIT>' , <EOL> chrome_msg ) <EOL> payload [ '<STR_LIT>' ] = '<STR_LIT:GET>' <EOL> self . logger . debug ( <EOL> '<STR_LIT>' , <EOL> self . exchange_name , client_id , payload ) <EOL> with self . _producer_lock : <EOL> publish = self . _producer_conn . ensure ( self . _producer , <EOL> self . _producer . publish ) <EOL> publish ( payload , exchange = self . _exchange , routing_key = client_id ) <EOL> def browse_page_sync ( ) : <EOL> self . logger . info ( <EOL> '<STR_LIT>' , <EOL> browser , client_id , url , behavior_parameters ) <EOL> try : <EOL> browser . browse_page ( url , on_response = on_response , <EOL> behavior_parameters = behavior_parameters ) <EOL> message . ack ( ) <EOL> except BrowsingException as e : <EOL> self . logger . warn ( "<STR_LIT>" . format ( url , e ) ) <EOL> message . requeue ( ) <EOL> except : <EOL> self . logger . critical ( "<STR_LIT>" . format ( url ) , exc_info = True ) <EOL> message . requeue ( ) <EOL> finally : <EOL> browser . stop ( ) <EOL> self . _browser_pool . release ( browser ) <EOL> def browse_thread_run_then_cleanup ( ) : <EOL> browse_page_sync ( ) <EOL> with self . _browsing_threads_lock : <EOL> self . _browsing_threads . remove ( threading . current_thread ( ) ) <EOL> import random <EOL> thread_name = "<STR_LIT>" . format ( browser . chrome_port , <EOL> '<STR_LIT>' . join ( ( random . choice ( '<STR_LIT>' ) for _ in range ( <NUM_LIT:6> ) ) ) ) <EOL> th = threading . Thread ( target = browse_thread_run_then_cleanup , name = thread_name ) <EOL> with self . _browsing_threads_lock : <EOL> self . _browsing_threads . add ( th ) <EOL> th . start ( ) </s>
