<s> from __future__ import absolute_import , division <EOL> import logging <EOL> import random <EOL> import gevent <EOL> import blinker <EOL> import time <EOL> import six <EOL> from itertools import cycle <EOL> from collections import defaultdict <EOL> from multiprocessing import cpu_count <EOL> from gevent . queue import Queue <EOL> from six . moves import range <EOL> from . lookupd import Lookupd <EOL> from . nsqd import Nsqd <EOL> from . backofftimer import BackoffTimer <EOL> from . states import INIT , RUNNING , BACKOFF , THROTTLED , CLOSED <EOL> from . decorators import cached_property <EOL> from . errors import ( <EOL> NSQException , <EOL> NSQNoConnections , <EOL> NSQRequeueMessage , <EOL> NSQSocketError , <EOL> ) <EOL> class Reader ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( <EOL> self , <EOL> topic , <EOL> channel , <EOL> nsqd_tcp_addresses = [ ] , <EOL> lookupd_http_addresses = [ ] , <EOL> name = None , <EOL> message_handler = None , <EOL> async = False , <EOL> max_tries = <NUM_LIT:5> , <EOL> max_in_flight = <NUM_LIT:1> , <EOL> max_concurrency = <NUM_LIT:0> , <EOL> requeue_delay = <NUM_LIT:0> , <EOL> lookupd_poll_interval = <NUM_LIT> , <EOL> lookupd_poll_jitter = <NUM_LIT> , <EOL> low_ready_idle_timeout = <NUM_LIT:10> , <EOL> max_backoff_duration = <NUM_LIT> , <EOL> backoff_on_requeue = True , <EOL> ** kwargs <EOL> ) : <EOL> if not nsqd_tcp_addresses and not lookupd_http_addresses : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> nsqd_tcp_addresses = self . _get_nsqds ( nsqd_tcp_addresses ) <EOL> lookupd_http_addresses = self . _get_lookupds ( lookupd_http_addresses ) <EOL> random . shuffle ( lookupd_http_addresses ) <EOL> self . nsqd_tcp_addresses = nsqd_tcp_addresses <EOL> self . lookupds = [ Lookupd ( a ) for a in lookupd_http_addresses ] <EOL> self . iterlookupds = cycle ( self . lookupds ) <EOL> self . topic = topic <EOL> self . channel = channel <EOL> self . async = async <EOL> self . max_tries = max_tries <EOL> self . max_in_flight = max_in_flight <EOL> self . requeue_delay = requeue_delay <EOL> self . lookupd_poll_interval = lookupd_poll_interval <EOL> self . lookupd_poll_jitter = lookupd_poll_jitter <EOL> self . low_ready_idle_timeout = low_ready_idle_timeout <EOL> self . backoff_on_requeue = backoff_on_requeue <EOL> self . max_backoff_duration = max_backoff_duration <EOL> self . conn_kwargs = kwargs <EOL> if name : <EOL> self . name = name <EOL> else : <EOL> self . name = '<STR_LIT>' % ( __name__ , self . topic , self . channel ) <EOL> if message_handler is not None : <EOL> self . on_message . connect ( message_handler , weak = False ) <EOL> if max_concurrency < <NUM_LIT:0> : <EOL> self . max_concurrency = cpu_count ( ) <EOL> else : <EOL> self . max_concurrency = max_concurrency <EOL> if max_concurrency : <EOL> self . queue = Queue ( ) <EOL> else : <EOL> self . queue = None <EOL> self . _need_ready_redistributed = False <EOL> self . last_random_ready = time . time ( ) <EOL> self . state = INIT <EOL> self . logger = logging . getLogger ( self . name ) <EOL> self . conn_backoffs = defaultdict ( self . create_backoff ) <EOL> self . backoff = self . create_backoff ( ) <EOL> self . conns = set ( ) <EOL> self . pending = set ( ) <EOL> self . workers = [ ] <EOL> self . conn_workers = { } <EOL> @ cached_property <EOL> def on_message ( self ) : <EOL> """<STR_LIT>""" <EOL> return blinker . Signal ( doc = '<STR_LIT>' ) <EOL> @ cached_property <EOL> def on_response ( self ) : <EOL> """<STR_LIT>""" <EOL> return blinker . Signal ( doc = '<STR_LIT>' ) <EOL> @ cached_property <EOL> def on_error ( self ) : <EOL> """<STR_LIT>""" <EOL> return blinker . Signal ( doc = '<STR_LIT>' ) <EOL> @ cached_property <EOL> def on_finish ( self ) : <EOL> """<STR_LIT>""" <EOL> return blinker . Signal ( doc = '<STR_LIT>' ) <EOL> @ cached_property <EOL> def on_requeue ( self ) : <EOL> """<STR_LIT>""" <EOL> return blinker . Signal ( doc = '<STR_LIT>' ) <EOL> @ cached_property <EOL> def on_giving_up ( self ) : <EOL> """<STR_LIT>""" <EOL> return blinker . Signal ( doc = '<STR_LIT>' ) <EOL> @ cached_property <EOL> def on_auth ( self ) : <EOL> """<STR_LIT>""" <EOL> return blinker . Signal ( doc = '<STR_LIT>' ) <EOL> @ cached_property <EOL> def on_exception ( self ) : <EOL> """<STR_LIT>""" <EOL> return blinker . Signal ( doc = '<STR_LIT>' ) <EOL> @ cached_property <EOL> def on_close ( self ) : <EOL> """<STR_LIT>""" <EOL> return blinker . Signal ( doc = '<STR_LIT>' ) <EOL> def _get_nsqds ( self , nsqd_tcp_addresses ) : <EOL> if isinstance ( nsqd_tcp_addresses , six . string_types ) : <EOL> return set ( [ nsqd_tcp_addresses ] ) <EOL> elif isinstance ( nsqd_tcp_addresses , ( list , tuple , set ) ) : <EOL> return set ( nsqd_tcp_addresses ) <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> def _get_lookupds ( self , lookupd_http_addresses ) : <EOL> if isinstance ( lookupd_http_addresses , six . string_types ) : <EOL> return [ lookupd_http_addresses ] <EOL> elif isinstance ( lookupd_http_addresses , ( list , tuple ) ) : <EOL> lookupd_http_addresses = list ( lookupd_http_addresses ) <EOL> return lookupd_http_addresses <EOL> msg = '<STR_LIT>' <EOL> raise TypeError ( msg ) <EOL> def start ( self , block = True ) : <EOL> """<STR_LIT>""" <EOL> if self . state != INIT : <EOL> self . logger . warn ( '<STR_LIT>' % self . name ) <EOL> if block : <EOL> self . join ( ) <EOL> return <EOL> if not any ( self . on_message . receivers_for ( blinker . ANY ) ) : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> self . logger . debug ( '<STR_LIT>' % self . name ) <EOL> self . state = RUNNING <EOL> self . query_nsqd ( ) <EOL> if self . lookupds : <EOL> self . query_lookupd ( ) <EOL> self . workers . append ( gevent . spawn ( self . _poll_lookupd ) ) <EOL> self . workers . append ( gevent . spawn ( self . _poll_ready ) ) <EOL> for _ in range ( self . max_concurrency ) : <EOL> self . workers . append ( gevent . spawn ( self . _run ) ) <EOL> if block : <EOL> self . join ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_running : <EOL> return <EOL> self . state = CLOSED <EOL> self . logger . debug ( '<STR_LIT>' % len ( self . workers ) ) <EOL> gevent . killall ( self . workers , block = False ) <EOL> self . logger . debug ( '<STR_LIT>' % len ( self . conns ) ) <EOL> for conn in self . conns : <EOL> conn . close_stream ( ) <EOL> self . on_close . send ( self ) <EOL> def join ( self , timeout = None , raise_error = False ) : <EOL> """<STR_LIT>""" <EOL> gevent . joinall ( self . workers , timeout , raise_error ) <EOL> gevent . joinall ( list ( self . conn_workers . values ( ) ) , timeout , raise_error ) <EOL> @ property <EOL> def is_running ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . state in ( RUNNING , BACKOFF , THROTTLED ) <EOL> @ property <EOL> def is_starved ( self ) : <EOL> """<STR_LIT>""" <EOL> return any ( conn . is_starved for conn in self . conns ) <EOL> @ property <EOL> def connection_max_in_flight ( self ) : <EOL> return max ( <NUM_LIT:1> , self . max_in_flight // max ( <NUM_LIT:1> , len ( self . conns ) ) ) <EOL> @ property <EOL> def total_ready_count ( self ) : <EOL> return sum ( c . ready_count for c in self . conns ) <EOL> @ property <EOL> def total_in_flight ( self ) : <EOL> return sum ( c . in_flight for c in self . conns ) <EOL> @ property <EOL> def total_in_flight_or_ready ( self ) : <EOL> return self . total_in_flight + self . total_ready_count <EOL> def send_ready ( self , conn , count ) : <EOL> if self . state == CLOSED : <EOL> self . logger . debug ( '<STR_LIT>' % conn ) <EOL> return <EOL> if self . state == BACKOFF : <EOL> self . logger . debug ( '<STR_LIT>' % conn ) <EOL> return <EOL> if self . state == THROTTLED and self . total_in_flight_or_ready : <EOL> msg = '<STR_LIT>' <EOL> self . logger . debug ( msg % ( conn , self . total_in_flight_or_ready ) ) <EOL> return <EOL> if not conn . is_connected : <EOL> self . logger . debug ( '<STR_LIT>' % conn ) <EOL> return <EOL> total = self . total_ready_count - conn . ready_count + count <EOL> if total > self . max_in_flight : <EOL> if not ( conn . ready_count or conn . in_flight ) : <EOL> self . logger . debug ( '<STR_LIT>' % conn ) <EOL> gevent . spawn_later ( <NUM_LIT:5> , self . send_ready , conn , count ) <EOL> return <EOL> self . logger . debug ( '<STR_LIT>' % ( conn , count ) ) <EOL> try : <EOL> conn . ready ( count ) <EOL> except NSQSocketError as error : <EOL> self . logger . warn ( '<STR_LIT>' % ( conn , count , error ) ) <EOL> def query_nsqd ( self ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> for address in self . nsqd_tcp_addresses : <EOL> address , port = address . split ( '<STR_LIT::>' ) <EOL> conn = Nsqd ( address , int ( port ) , ** self . conn_kwargs ) <EOL> self . connect_to_nsqd ( conn ) <EOL> def query_lookupd ( self ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> lookupd = next ( self . iterlookupds ) <EOL> try : <EOL> producers = lookupd . lookup ( self . topic ) [ '<STR_LIT>' ] <EOL> self . logger . debug ( '<STR_LIT>' % len ( producers ) ) <EOL> except Exception as error : <EOL> msg = '<STR_LIT>' <EOL> self . logger . warn ( msg % ( self . topic , lookupd . address , error ) ) <EOL> return <EOL> for producer in producers : <EOL> conn = Nsqd ( <EOL> producer . get ( '<STR_LIT>' ) or producer [ '<STR_LIT:address>' ] , <EOL> producer [ '<STR_LIT>' ] , <EOL> producer [ '<STR_LIT>' ] , <EOL> ** self . conn_kwargs <EOL> ) <EOL> self . connect_to_nsqd ( conn ) <EOL> def create_backoff ( self ) : <EOL> return BackoffTimer ( max_interval = self . max_backoff_duration ) <EOL> def start_backoff ( self ) : <EOL> self . state = BACKOFF <EOL> for conn in self . conns : <EOL> try : <EOL> conn . ready ( <NUM_LIT:0> ) <EOL> except NSQSocketError as error : <EOL> self . logger . warn ( '<STR_LIT>' % ( conn , error ) ) <EOL> interval = self . backoff . get_interval ( ) <EOL> self . logger . info ( '<STR_LIT>' % interval ) <EOL> gevent . sleep ( interval ) <EOL> self . state = THROTTLED <EOL> conn = self . random_connection ( ) <EOL> if not conn : <EOL> return <EOL> self . logger . info ( '<STR_LIT>' % conn ) <EOL> self . send_ready ( conn , <NUM_LIT:1> ) <EOL> def complete_backoff ( self ) : <EOL> self . state = RUNNING <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> count = self . connection_max_in_flight <EOL> for conn in self . conns : <EOL> self . send_ready ( conn , count ) <EOL> def _poll_lookupd ( self ) : <EOL> delay = self . lookupd_poll_interval * self . lookupd_poll_jitter <EOL> gevent . sleep ( random . random ( ) * delay ) <EOL> while True : <EOL> gevent . sleep ( self . lookupd_poll_interval ) <EOL> self . query_lookupd ( ) <EOL> def _poll_ready ( self ) : <EOL> while True : <EOL> gevent . sleep ( <NUM_LIT:5> ) <EOL> if not self . need_ready_redistributed : <EOL> continue <EOL> self . redistribute_ready_state ( ) <EOL> @ property <EOL> def need_ready_redistributed ( self ) : <EOL> if self . state == BACKOFF : <EOL> return False <EOL> if self . _need_ready_redistributed : <EOL> return True <EOL> if len ( self . conns ) > self . max_in_flight : <EOL> return True <EOL> if self . state == THROTTLED and len ( self . conns ) > <NUM_LIT:1> : <EOL> return True <EOL> @ need_ready_redistributed . setter <EOL> def need_ready_redistributed ( self , value ) : <EOL> self . _need_ready_redistributed = value <EOL> def redistribute_ready_state ( self ) : <EOL> self . need_ready_redistributed = False <EOL> for conn in self . conns : <EOL> if conn . ready_count == <NUM_LIT:0> : <EOL> continue <EOL> if ( time . time ( ) - conn . last_message ) < self . low_ready_idle_timeout : <EOL> continue <EOL> self . logger . info ( '<STR_LIT>' % conn ) <EOL> try : <EOL> conn . ready ( <NUM_LIT:0> ) <EOL> except NSQSocketError as error : <EOL> self . logger . warn ( '<STR_LIT>' % ( conn , error ) ) <EOL> if self . state == THROTTLED : <EOL> max_in_flight = <NUM_LIT:1> - self . total_in_flight_or_ready <EOL> else : <EOL> max_in_flight = self . max_in_flight - self . total_ready_count <EOL> if max_in_flight <= <NUM_LIT:0> : <EOL> return <EOL> conns = list ( self . conns ) <EOL> conns = random . sample ( conns , min ( max_in_flight , len ( self . conns ) ) ) <EOL> for conn in conns : <EOL> self . logger . info ( '<STR_LIT>' % conn ) <EOL> self . send_ready ( conn , <NUM_LIT:1> ) <EOL> def random_ready_conn ( self , conn ) : <EOL> if len ( self . conns ) <= self . max_in_flight : <EOL> return conn <EOL> if ( time . time ( ) - self . last_random_ready ) < <NUM_LIT:30> : <EOL> return conn <EOL> ready_conns = [ c for c in self . conns if not c . ready_count ] <EOL> if not ready_conns : <EOL> return conn <EOL> self . last_random_ready = time . time ( ) <EOL> return random . choice ( ready_conns ) <EOL> def update_ready ( self , conn ) : <EOL> if self . state in ( BACKOFF , THROTTLED ) : <EOL> return <EOL> conn = self . random_ready_conn ( conn ) <EOL> if conn . ready_count >= max ( conn . last_ready * <NUM_LIT> , <NUM_LIT:2> ) : <EOL> return <EOL> self . send_ready ( conn , self . connection_max_in_flight ) <EOL> def random_connection ( self ) : <EOL> if not self . conns : <EOL> return None <EOL> return random . choice ( list ( self . conns ) ) <EOL> def publish ( self , topic , message ) : <EOL> """<STR_LIT>""" <EOL> conn = self . random_connection ( ) <EOL> if conn is None : <EOL> raise NSQNoConnections ( ) <EOL> conn . publish ( topic , message ) <EOL> def connect_to_nsqd ( self , conn ) : <EOL> if not self . is_running : <EOL> return <EOL> if conn in self . conns : <EOL> self . logger . debug ( '<STR_LIT>' % conn ) <EOL> return <EOL> if conn in self . pending : <EOL> self . logger . debug ( '<STR_LIT>' % conn ) <EOL> return <EOL> self . logger . debug ( '<STR_LIT>' % conn ) <EOL> conn . on_response . connect ( self . handle_response ) <EOL> conn . on_error . connect ( self . handle_error ) <EOL> conn . on_finish . connect ( self . handle_finish ) <EOL> conn . on_requeue . connect ( self . handle_requeue ) <EOL> conn . on_auth . connect ( self . handle_auth ) <EOL> if self . max_concurrency : <EOL> conn . on_message . connect ( self . queue_message ) <EOL> else : <EOL> conn . on_message . connect ( self . handle_message ) <EOL> self . pending . add ( conn ) <EOL> try : <EOL> conn . connect ( ) <EOL> conn . identify ( ) <EOL> if conn . max_ready_count < self . max_in_flight : <EOL> msg = '<STR_LIT:U+0020>' . join ( [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] ) <EOL> self . logger . warning ( msg % ( <EOL> conn , <EOL> conn . max_ready_count , <EOL> self . max_in_flight <EOL> ) ) <EOL> conn . subscribe ( self . topic , self . channel ) <EOL> self . send_ready ( conn , <NUM_LIT:1> ) <EOL> except NSQException as error : <EOL> self . logger . warn ( '<STR_LIT>' % ( conn , error ) ) <EOL> self . handle_connection_failure ( conn ) <EOL> return <EOL> finally : <EOL> self . pending . remove ( conn ) <EOL> if not self . is_running : <EOL> conn . close_stream ( ) <EOL> return <EOL> self . logger . info ( '<STR_LIT>' % conn ) <EOL> self . handle_connection_success ( conn ) <EOL> def _listen ( self , conn ) : <EOL> try : <EOL> conn . listen ( ) <EOL> except NSQException as error : <EOL> self . logger . warning ( '<STR_LIT>' % ( conn , error ) ) <EOL> self . handle_connection_failure ( conn ) <EOL> def _run ( self ) : <EOL> for conn , message in self . queue : <EOL> if not conn . is_connected : <EOL> continue <EOL> self . handle_message ( conn , message ) <EOL> def queue_message ( self , conn , message ) : <EOL> self . logger . debug ( '<STR_LIT>' % ( conn , message . id ) ) <EOL> self . queue . put ( ( conn , message ) ) <EOL> def handle_connection_success ( self , conn ) : <EOL> self . conns . add ( conn ) <EOL> self . conn_workers [ conn ] = gevent . spawn ( self . _listen , conn ) <EOL> if str ( conn ) not in self . nsqd_tcp_addresses : <EOL> return <EOL> self . conn_backoffs [ conn ] . success ( ) <EOL> def handle_connection_failure ( self , conn ) : <EOL> self . conns . discard ( conn ) <EOL> self . conn_workers . pop ( conn , None ) <EOL> conn . close_stream ( ) <EOL> if self . state == CLOSED : <EOL> return <EOL> if conn . ready_count : <EOL> self . need_ready_redistributed = True <EOL> if str ( conn ) not in self . nsqd_tcp_addresses : <EOL> return <EOL> seconds = self . conn_backoffs [ conn ] . failure ( ) . get_interval ( ) <EOL> self . logger . debug ( '<STR_LIT>' % ( conn , seconds ) ) <EOL> gevent . spawn_later ( seconds , self . connect_to_nsqd , conn ) <EOL> def handle_response ( self , conn , response ) : <EOL> self . logger . debug ( '<STR_LIT>' % ( conn , response ) ) <EOL> self . on_response . send ( self , response = response ) <EOL> def handle_error ( self , conn , error ) : <EOL> self . logger . debug ( '<STR_LIT>' % ( conn , error ) ) <EOL> self . on_error . send ( self , error = error ) <EOL> def _handle_message ( self , message ) : <EOL> if self . max_tries and message . attempts > self . max_tries : <EOL> msg = "<STR_LIT>" <EOL> self . logger . warning ( msg % ( message . id , self . max_tries ) ) <EOL> self . on_giving_up . send ( self , message = message ) <EOL> return message . finish ( ) <EOL> self . on_message . send ( self , message = message ) <EOL> if not self . is_running : <EOL> return <EOL> if self . async : <EOL> return <EOL> if message . has_responded ( ) : <EOL> return <EOL> message . finish ( ) <EOL> def handle_message ( self , conn , message ) : <EOL> self . logger . debug ( '<STR_LIT>' % ( conn , message . id ) ) <EOL> self . update_ready ( conn ) <EOL> try : <EOL> return self . _handle_message ( message ) <EOL> except NSQRequeueMessage as error : <EOL> if error . backoff is None : <EOL> backoff = self . backoff_on_requeue <EOL> else : <EOL> backoff = error . backoff <EOL> except Exception as error : <EOL> backoff = True <EOL> msg = '<STR_LIT>' % conn <EOL> self . logger . exception ( msg ) <EOL> self . on_exception . send ( self , message = message , error = error ) <EOL> if not self . is_running : <EOL> return <EOL> if message . has_responded ( ) : <EOL> return <EOL> try : <EOL> message . requeue ( self . requeue_delay , backoff ) <EOL> except NSQException as error : <EOL> msg = '<STR_LIT>' % ( conn , error ) <EOL> self . logger . warning ( msg ) <EOL> def handle_finish ( self , conn , message_id ) : <EOL> self . logger . debug ( '<STR_LIT>' % ( conn , message_id ) ) <EOL> self . handle_backoff ( False ) <EOL> self . update_ready ( conn ) <EOL> self . on_finish . send ( self , message_id = message_id ) <EOL> def handle_requeue ( self , conn , message_id , timeout , backoff ) : <EOL> msg = '<STR_LIT>' <EOL> self . logger . debug ( msg % ( conn , message_id , timeout ) ) <EOL> self . handle_backoff ( backoff ) <EOL> self . update_ready ( conn ) <EOL> self . on_requeue . send ( self , message_id = message_id , timeout = timeout ) <EOL> def handle_auth ( self , conn , response ) : <EOL> metadata = [ ] <EOL> if response . get ( '<STR_LIT>' ) : <EOL> metadata . append ( "<STR_LIT>" % response [ '<STR_LIT>' ] ) <EOL> if response . get ( '<STR_LIT>' ) : <EOL> metadata . append ( "<STR_LIT>" % response [ '<STR_LIT>' ] ) <EOL> if response . get ( '<STR_LIT>' ) : <EOL> metadata . append ( response [ '<STR_LIT>' ] ) <EOL> self . logger . info ( '<STR_LIT>' % ( conn , '<STR_LIT:U+0020>' . join ( metadata ) ) ) <EOL> self . on_auth . send ( self , conn = conn , response = response ) <EOL> def handle_backoff ( self , backoff ) : <EOL> if not self . max_backoff_duration : <EOL> return <EOL> if self . state in ( BACKOFF , CLOSED ) : <EOL> return <EOL> if backoff : <EOL> self . backoff . failure ( ) <EOL> else : <EOL> self . backoff . success ( ) <EOL> if self . state == THROTTLED and self . backoff . is_reset ( ) : <EOL> return self . complete_backoff ( ) <EOL> if not self . backoff . is_reset ( ) : <EOL> return self . start_backoff ( ) </s>
