<s> from __future__ import absolute_import <EOL> import logging <EOL> import time <EOL> import functools <EOL> import urllib <EOL> import random <EOL> import urlparse <EOL> import cgi <EOL> import warnings <EOL> import inspect <EOL> try : <EOL> import simplejson as json <EOL> except ImportError : <EOL> import json <EOL> from tornado . ioloop import PeriodicCallback <EOL> import tornado . httpclient <EOL> from . backoff_timer import BackoffTimer <EOL> from . client import Client <EOL> from . import protocol <EOL> from . import async <EOL> logger = logging . getLogger ( __name__ ) <EOL> class Reader ( Client ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( <EOL> self , <EOL> topic , <EOL> channel , <EOL> message_handler = None , <EOL> name = None , <EOL> nsqd_tcp_addresses = None , <EOL> lookupd_http_addresses = None , <EOL> max_tries = <NUM_LIT:5> , <EOL> max_in_flight = <NUM_LIT:1> , <EOL> lookupd_poll_interval = <NUM_LIT> , <EOL> low_rdy_idle_timeout = <NUM_LIT:10> , <EOL> max_backoff_duration = <NUM_LIT> , <EOL> lookupd_poll_jitter = <NUM_LIT> , <EOL> lookupd_connect_timeout = <NUM_LIT:1> , <EOL> lookupd_request_timeout = <NUM_LIT:2> , <EOL> ** kwargs ) : <EOL> super ( Reader , self ) . __init__ ( ** kwargs ) <EOL> assert isinstance ( topic , ( str , unicode ) ) and len ( topic ) > <NUM_LIT:0> <EOL> assert isinstance ( channel , ( str , unicode ) ) and len ( channel ) > <NUM_LIT:0> <EOL> assert isinstance ( max_in_flight , int ) and max_in_flight > <NUM_LIT:0> <EOL> assert isinstance ( max_backoff_duration , ( int , float ) ) and max_backoff_duration > <NUM_LIT:0> <EOL> assert isinstance ( name , ( str , unicode , None . __class__ ) ) <EOL> assert isinstance ( lookupd_poll_interval , int ) <EOL> assert isinstance ( lookupd_poll_jitter , float ) <EOL> assert isinstance ( lookupd_connect_timeout , int ) <EOL> assert isinstance ( lookupd_request_timeout , int ) <EOL> assert lookupd_poll_jitter >= <NUM_LIT:0> and lookupd_poll_jitter <= <NUM_LIT:1> <EOL> if nsqd_tcp_addresses : <EOL> if not isinstance ( nsqd_tcp_addresses , ( list , set , tuple ) ) : <EOL> assert isinstance ( nsqd_tcp_addresses , ( str , unicode ) ) <EOL> nsqd_tcp_addresses = [ nsqd_tcp_addresses ] <EOL> else : <EOL> nsqd_tcp_addresses = [ ] <EOL> if lookupd_http_addresses : <EOL> if not isinstance ( lookupd_http_addresses , ( list , set , tuple ) ) : <EOL> assert isinstance ( lookupd_http_addresses , ( str , unicode ) ) <EOL> lookupd_http_addresses = [ lookupd_http_addresses ] <EOL> random . shuffle ( lookupd_http_addresses ) <EOL> else : <EOL> lookupd_http_addresses = [ ] <EOL> assert nsqd_tcp_addresses or lookupd_http_addresses <EOL> self . name = name or ( topic + '<STR_LIT::>' + channel ) <EOL> self . message_handler = None <EOL> if message_handler : <EOL> self . set_message_handler ( message_handler ) <EOL> self . topic = topic <EOL> self . channel = channel <EOL> self . nsqd_tcp_addresses = nsqd_tcp_addresses <EOL> self . lookupd_http_addresses = lookupd_http_addresses <EOL> self . lookupd_query_index = <NUM_LIT:0> <EOL> self . max_tries = max_tries <EOL> self . max_in_flight = max_in_flight <EOL> self . low_rdy_idle_timeout = low_rdy_idle_timeout <EOL> self . total_rdy = <NUM_LIT:0> <EOL> self . need_rdy_redistributed = False <EOL> self . lookupd_poll_interval = lookupd_poll_interval <EOL> self . lookupd_poll_jitter = lookupd_poll_jitter <EOL> self . lookupd_connect_timeout = lookupd_connect_timeout <EOL> self . lookupd_request_timeout = lookupd_request_timeout <EOL> self . random_rdy_ts = time . time ( ) <EOL> valid_args = inspect . getargspec ( async . AsyncConn . __init__ ) [ <NUM_LIT:0> ] <EOL> diff = set ( kwargs ) - set ( valid_args ) <EOL> assert len ( diff ) == <NUM_LIT:0> , '<STR_LIT>' % list ( diff ) <EOL> self . conn_kwargs = kwargs <EOL> self . backoff_timer = BackoffTimer ( <NUM_LIT:0> , max_backoff_duration ) <EOL> self . backoff_timeout = None <EOL> self . backoff_block = False <EOL> self . backoff_block_completed = True <EOL> self . conns = { } <EOL> self . connection_attempts = { } <EOL> self . http_client = tornado . httpclient . AsyncHTTPClient ( io_loop = self . io_loop ) <EOL> self . io_loop . add_callback ( self . _run ) <EOL> self . redist_periodic = None <EOL> self . query_periodic = None <EOL> def _run ( self ) : <EOL> assert self . message_handler , "<STR_LIT>" <EOL> logger . info ( '<STR_LIT>' , self . name , self . topic , self . channel ) <EOL> for addr in self . nsqd_tcp_addresses : <EOL> address , port = addr . split ( '<STR_LIT::>' ) <EOL> self . connect_to_nsqd ( address , int ( port ) ) <EOL> self . redist_periodic = PeriodicCallback ( <EOL> self . _redistribute_rdy_state , <EOL> <NUM_LIT:5> * <NUM_LIT:1000> , <EOL> io_loop = self . io_loop , <EOL> ) <EOL> self . redist_periodic . start ( ) <EOL> if not self . lookupd_http_addresses : <EOL> return <EOL> self . query_lookupd ( ) <EOL> self . query_periodic = PeriodicCallback ( <EOL> self . query_lookupd , <EOL> self . lookupd_poll_interval * <NUM_LIT:1000> , <EOL> io_loop = self . io_loop , <EOL> ) <EOL> delay = random . random ( ) * self . lookupd_poll_interval * self . lookupd_poll_jitter <EOL> self . io_loop . add_timeout ( time . time ( ) + delay , self . query_periodic . start ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> for conn in self . conns . values ( ) : <EOL> conn . close ( ) <EOL> self . redist_periodic . stop ( ) <EOL> if self . query_periodic is not None : <EOL> self . query_periodic . stop ( ) <EOL> def set_message_handler ( self , message_handler ) : <EOL> """<STR_LIT>""" <EOL> assert callable ( message_handler ) , '<STR_LIT>' <EOL> self . message_handler = message_handler <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> def is_starved ( self ) : <EOL> """<STR_LIT>""" <EOL> for conn in self . conns . itervalues ( ) : <EOL> if conn . in_flight > <NUM_LIT:0> and conn . in_flight >= ( conn . last_rdy * <NUM_LIT> ) : <EOL> return True <EOL> return False <EOL> def _on_message ( self , conn , message , ** kwargs ) : <EOL> try : <EOL> self . _handle_message ( conn , message ) <EOL> except Exception : <EOL> logger . exception ( '<STR_LIT>' , conn . id , self . name , message ) <EOL> def _handle_message ( self , conn , message ) : <EOL> self . total_rdy = max ( self . total_rdy - <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> rdy_conn = conn <EOL> if len ( self . conns ) > self . max_in_flight and time . time ( ) - self . random_rdy_ts > <NUM_LIT:30> : <EOL> self . random_rdy_ts = time . time ( ) <EOL> conns_with_no_rdy = [ c for c in self . conns . itervalues ( ) if not c . rdy ] <EOL> if conns_with_no_rdy : <EOL> rdy_conn = random . choice ( conns_with_no_rdy ) <EOL> if rdy_conn is not conn : <EOL> logger . info ( '<STR_LIT>' , <EOL> conn . id , self . name , rdy_conn . id ) <EOL> self . _maybe_update_rdy ( rdy_conn ) <EOL> success = False <EOL> try : <EOL> if <NUM_LIT:0> < self . max_tries < message . attempts : <EOL> self . giving_up ( message ) <EOL> return message . finish ( ) <EOL> pre_processed_message = self . preprocess_message ( message ) <EOL> if not self . validate_message ( pre_processed_message ) : <EOL> return message . finish ( ) <EOL> success = self . process_message ( message ) <EOL> except Exception : <EOL> logger . exception ( '<STR_LIT>' , <EOL> conn . id , self . name , message . id , message . body ) <EOL> if not message . has_responded ( ) : <EOL> return message . requeue ( ) <EOL> if not message . is_async ( ) and not message . has_responded ( ) : <EOL> assert success is not None , '<STR_LIT>' <EOL> if success : <EOL> return message . finish ( ) <EOL> return message . requeue ( ) <EOL> def _maybe_update_rdy ( self , conn ) : <EOL> if self . backoff_timer . get_interval ( ) or self . max_in_flight == <NUM_LIT:0> : <EOL> return <EOL> if conn . rdy <= <NUM_LIT:1> or conn . rdy < int ( conn . last_rdy * <NUM_LIT> ) : <EOL> self . _send_rdy ( conn , self . _connection_max_in_flight ( ) ) <EOL> def _finish_backoff_block ( self ) : <EOL> self . backoff_timeout = None <EOL> self . backoff_block = False <EOL> if not self . backoff_timer . get_interval ( ) : <EOL> self . _complete_backoff_block ( ) <EOL> return <EOL> if not self . conns or self . max_in_flight == <NUM_LIT:0> : <EOL> return <EOL> conn = random . choice ( self . conns . values ( ) ) <EOL> logger . info ( '<STR_LIT>' , conn . id , self . name ) <EOL> self . _send_rdy ( conn , <NUM_LIT:1> ) <EOL> return conn <EOL> def _on_backoff_resume ( self , success , ** kwargs ) : <EOL> if success : <EOL> self . backoff_timer . success ( ) <EOL> elif success is False and not self . backoff_block : <EOL> self . backoff_timer . failure ( ) <EOL> self . _enter_continue_or_exit_backoff ( ) <EOL> def _complete_backoff_block ( self ) : <EOL> self . backoff_block_completed = True <EOL> rdy = self . _connection_max_in_flight ( ) <EOL> logger . info ( '<STR_LIT>' , <EOL> self . name , len ( self . conns ) ) <EOL> for c in self . conns . values ( ) : <EOL> self . _send_rdy ( c , rdy ) <EOL> def _enter_continue_or_exit_backoff ( self ) : <EOL> current_backoff_interval = self . backoff_timer . get_interval ( ) <EOL> if self . backoff_block : <EOL> return <EOL> if not self . backoff_block_completed and not current_backoff_interval : <EOL> self . _complete_backoff_block ( ) <EOL> return <EOL> if current_backoff_interval : <EOL> self . _start_backoff_block ( ) <EOL> def _start_backoff_block ( self ) : <EOL> self . backoff_block = True <EOL> self . backoff_block_completed = False <EOL> backoff_interval = self . backoff_timer . get_interval ( ) <EOL> logger . info ( '<STR_LIT>' , <EOL> self . name , backoff_interval , len ( self . conns ) ) <EOL> for c in self . conns . values ( ) : <EOL> self . _send_rdy ( c , <NUM_LIT:0> ) <EOL> self . backoff_timeout = self . io_loop . add_timeout ( time . time ( ) + backoff_interval , <EOL> self . _finish_backoff_block ) <EOL> def _rdy_retry ( self , conn , value ) : <EOL> conn . rdy_timeout = None <EOL> self . _send_rdy ( conn , value ) <EOL> def _send_rdy ( self , conn , value ) : <EOL> if conn . rdy_timeout : <EOL> self . io_loop . remove_timeout ( conn . rdy_timeout ) <EOL> conn . rdy_timeout = None <EOL> if value and ( self . disabled ( ) or self . max_in_flight == <NUM_LIT:0> ) : <EOL> logger . info ( '<STR_LIT>' , conn . id , self . name ) <EOL> rdy_retry_callback = functools . partial ( self . _rdy_retry , conn , value ) <EOL> conn . rdy_timeout = self . io_loop . add_timeout ( time . time ( ) + <NUM_LIT:15> , rdy_retry_callback ) <EOL> return <EOL> if value > conn . max_rdy_count : <EOL> value = conn . max_rdy_count <EOL> if ( self . total_rdy + value ) > self . max_in_flight : <EOL> if not conn . rdy : <EOL> rdy_retry_callback = functools . partial ( self . _rdy_retry , conn , value ) <EOL> conn . rdy_timeout = self . io_loop . add_timeout ( time . time ( ) + <NUM_LIT:5> , rdy_retry_callback ) <EOL> return <EOL> if conn . send_rdy ( value ) : <EOL> self . total_rdy = max ( self . total_rdy - conn . rdy + value , <NUM_LIT:0> ) <EOL> def connect_to_nsqd ( self , host , port ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( host , ( str , unicode ) ) <EOL> assert isinstance ( port , int ) <EOL> conn = async . AsyncConn ( host , port , ** self . conn_kwargs ) <EOL> conn . on ( '<STR_LIT>' , self . _on_connection_identify ) <EOL> conn . on ( '<STR_LIT>' , self . _on_connection_identify_response ) <EOL> conn . on ( '<STR_LIT>' , self . _on_connection_auth ) <EOL> conn . on ( '<STR_LIT>' , self . _on_connection_auth_response ) <EOL> conn . on ( '<STR_LIT:error>' , self . _on_connection_error ) <EOL> conn . on ( '<STR_LIT>' , self . _on_connection_close ) <EOL> conn . on ( '<STR_LIT>' , self . _on_connection_ready ) <EOL> conn . on ( '<STR_LIT:message>' , self . _on_message ) <EOL> conn . on ( '<STR_LIT>' , self . _on_heartbeat ) <EOL> conn . on ( '<STR_LIT>' , functools . partial ( self . _on_backoff_resume , success = False ) ) <EOL> conn . on ( '<STR_LIT>' , functools . partial ( self . _on_backoff_resume , success = True ) ) <EOL> conn . on ( '<STR_LIT>' , functools . partial ( self . _on_backoff_resume , success = None ) ) <EOL> if conn . id in self . conns : <EOL> return <EOL> now = time . time ( ) <EOL> last_connect_attempt = self . connection_attempts . get ( conn . id ) <EOL> if last_connect_attempt and last_connect_attempt > now - <NUM_LIT:10> : <EOL> return <EOL> self . connection_attempts [ conn . id ] = now <EOL> logger . info ( '<STR_LIT>' , conn . id , self . name ) <EOL> conn . connect ( ) <EOL> return conn <EOL> def _on_connection_ready ( self , conn , ** kwargs ) : <EOL> conn . send ( protocol . subscribe ( self . topic , self . channel ) ) <EOL> if conn . id in self . conns : <EOL> logger . warning ( <EOL> '<STR_LIT>' , <EOL> conn . id , self . name ) <EOL> conn . close ( ) <EOL> return <EOL> if conn . max_rdy_count < self . max_in_flight : <EOL> logger . warning ( <EOL> '<STR_LIT>' , <EOL> conn . id , self . name , conn . max_rdy_count , self . max_in_flight ) <EOL> self . conns [ conn . id ] = conn <EOL> if not self . backoff_timer . get_interval ( ) or len ( self . conns ) == <NUM_LIT:1> : <EOL> self . _send_rdy ( conn , <NUM_LIT:1> ) <EOL> def _on_connection_close ( self , conn , ** kwargs ) : <EOL> if conn . id in self . conns : <EOL> del self . conns [ conn . id ] <EOL> self . total_rdy = max ( self . total_rdy - conn . rdy , <NUM_LIT:0> ) <EOL> logger . warning ( '<STR_LIT>' , conn . id , self . name ) <EOL> if ( conn . rdy_timeout or conn . rdy ) and ( len ( self . conns ) == self . max_in_flight or self . backoff_timer . get_interval ( ) ) : <EOL> self . need_rdy_redistributed = True <EOL> if conn . rdy_timeout : <EOL> self . io_loop . remove_timeout ( conn . rdy_timeout ) <EOL> conn . rdy_timeout = None <EOL> if not self . lookupd_http_addresses : <EOL> logger . info ( '<STR_LIT>' , conn . id , self . name ) <EOL> reconnect_callback = functools . partial ( self . connect_to_nsqd , <EOL> host = conn . host , port = conn . port ) <EOL> self . io_loop . add_timeout ( time . time ( ) + <NUM_LIT:15> , reconnect_callback ) <EOL> def query_lookupd ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = self . lookupd_http_addresses [ self . lookupd_query_index ] <EOL> self . lookupd_query_index = ( self . lookupd_query_index + <NUM_LIT:1> ) % len ( self . lookupd_http_addresses ) <EOL> if '<STR_LIT>' not in endpoint : <EOL> endpoint = '<STR_LIT>' + endpoint <EOL> scheme , netloc , path , query , fragment = urlparse . urlsplit ( endpoint ) <EOL> if not path or path == "<STR_LIT:/>" : <EOL> path = "<STR_LIT>" <EOL> params = cgi . parse_qs ( query ) <EOL> params [ '<STR_LIT>' ] = self . topic <EOL> query = urllib . urlencode ( _utf8_params ( params ) , doseq = <NUM_LIT:1> ) <EOL> lookupd_url = urlparse . urlunsplit ( ( scheme , netloc , path , query , fragment ) ) <EOL> req = tornado . httpclient . HTTPRequest ( <EOL> lookupd_url , method = '<STR_LIT:GET>' , <EOL> connect_timeout = self . lookupd_connect_timeout , <EOL> request_timeout = self . lookupd_request_timeout ) <EOL> callback = functools . partial ( self . _finish_query_lookupd , lookupd_url = lookupd_url ) <EOL> self . http_client . fetch ( req , callback = callback ) <EOL> def _finish_query_lookupd ( self , response , lookupd_url ) : <EOL> if response . error : <EOL> logger . warning ( '<STR_LIT>' , <EOL> self . name , lookupd_url , response . error ) <EOL> return <EOL> try : <EOL> lookup_data = json . loads ( response . body ) <EOL> except ValueError : <EOL> logger . warning ( '<STR_LIT>' , <EOL> self . name , lookupd_url , response . body ) <EOL> return <EOL> if lookup_data [ '<STR_LIT>' ] != <NUM_LIT:200> : <EOL> logger . warning ( '<STR_LIT>' , <EOL> self . name , lookupd_url , lookup_data [ '<STR_LIT>' ] ) <EOL> return <EOL> for producer in lookup_data [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] : <EOL> address = producer . get ( '<STR_LIT>' , producer . get ( '<STR_LIT:address>' ) ) <EOL> assert address <EOL> self . connect_to_nsqd ( address , producer [ '<STR_LIT>' ] ) <EOL> def set_max_in_flight ( self , max_in_flight ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( max_in_flight , int ) <EOL> self . max_in_flight = max_in_flight <EOL> if max_in_flight == <NUM_LIT:0> : <EOL> for conn in self . conns . itervalues ( ) : <EOL> if conn . rdy > <NUM_LIT:0> : <EOL> logger . debug ( '<STR_LIT>' , conn . id , self . name , conn . rdy ) <EOL> self . _send_rdy ( conn , <NUM_LIT:0> ) <EOL> self . total_rdy = <NUM_LIT:0> <EOL> else : <EOL> self . need_rdy_redistributed = True <EOL> self . _redistribute_rdy_state ( ) <EOL> def _redistribute_rdy_state ( self ) : <EOL> if not self . conns : <EOL> return <EOL> if self . disabled ( ) or self . backoff_block or self . max_in_flight == <NUM_LIT:0> : <EOL> return <EOL> if len ( self . conns ) > self . max_in_flight : <EOL> self . need_rdy_redistributed = True <EOL> logger . debug ( '<STR_LIT>' , <EOL> len ( self . conns ) , self . max_in_flight ) <EOL> backoff_interval = self . backoff_timer . get_interval ( ) <EOL> if backoff_interval and len ( self . conns ) > <NUM_LIT:1> : <EOL> self . need_rdy_redistributed = True <EOL> logger . debug ( '<STR_LIT>' , <EOL> backoff_interval , len ( self . conns ) ) <EOL> if self . need_rdy_redistributed : <EOL> self . need_rdy_redistributed = False <EOL> for conn_id , conn in self . conns . iteritems ( ) : <EOL> last_message_duration = time . time ( ) - conn . last_msg_timestamp <EOL> logger . debug ( '<STR_LIT>' , <EOL> conn . id , self . name , conn . rdy , last_message_duration ) <EOL> if conn . rdy > <NUM_LIT:0> and last_message_duration > self . low_rdy_idle_timeout : <EOL> logger . info ( '<STR_LIT>' , conn . id , self . name ) <EOL> self . _send_rdy ( conn , <NUM_LIT:0> ) <EOL> if backoff_interval : <EOL> max_in_flight = <NUM_LIT:1> - self . total_rdy <EOL> else : <EOL> max_in_flight = self . max_in_flight - self . total_rdy <EOL> possible_conns = self . conns . values ( ) <EOL> while possible_conns and max_in_flight : <EOL> max_in_flight -= <NUM_LIT:1> <EOL> conn = possible_conns . pop ( random . randrange ( len ( possible_conns ) ) ) <EOL> logger . info ( '<STR_LIT>' , conn . id , self . name ) <EOL> self . _send_rdy ( conn , <NUM_LIT:1> ) <EOL> return conn <EOL> def process_message ( self , message ) : <EOL> """<STR_LIT>""" <EOL> return self . message_handler ( message ) <EOL> def giving_up ( self , message ) : <EOL> """<STR_LIT>""" <EOL> logger . warning ( '<STR_LIT>' , <EOL> self . name , message . id , message . attempts , self . max_tries , message . body ) <EOL> def _on_connection_identify_response ( self , conn , data , ** kwargs ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . _disabled_notice = True <EOL> def semver ( v ) : <EOL> def cast ( x ) : <EOL> try : <EOL> return int ( x ) <EOL> except : <EOL> return x <EOL> return map ( cast , v . replace ( '<STR_LIT:->' , '<STR_LIT:.>' ) . split ( '<STR_LIT:.>' ) ) <EOL> if self . disabled . __code__ != Reader . disabled . __code__ and semver ( data [ '<STR_LIT:version>' ] ) >= semver ( '<STR_LIT>' ) : <EOL> logging . warning ( '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> warnings . warn ( '<STR_LIT>' + <EOL> '<STR_LIT>' , DeprecationWarning ) <EOL> return super ( Reader , self ) . _on_connection_identify_response ( conn , data , ** kwargs ) <EOL> @ classmethod <EOL> def disabled ( cls ) : <EOL> """<STR_LIT>""" <EOL> return False <EOL> def validate_message ( self , message ) : <EOL> return True <EOL> def preprocess_message ( self , message ) : <EOL> return message <EOL> def _utf8_params ( params ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( params , dict ) <EOL> encoded_params = [ ] <EOL> for k , v in params . items ( ) : <EOL> if v is None : <EOL> continue <EOL> if isinstance ( v , ( int , long , float ) ) : <EOL> v = str ( v ) <EOL> if isinstance ( v , ( list , tuple ) ) : <EOL> v = [ _utf8 ( x ) for x in v ] <EOL> else : <EOL> v = _utf8 ( v ) <EOL> encoded_params . append ( ( k , v ) ) <EOL> return dict ( encoded_params ) <EOL> def _utf8 ( s ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( s , unicode ) : <EOL> return s . encode ( "<STR_LIT:utf-8>" ) <EOL> assert isinstance ( s , str ) , "<STR_LIT>" % type ( s ) <EOL> return s </s>
