<s> import logging <EOL> import time <EOL> import gevent <EOL> from gevent import GreenletExit <EOL> from . constants import ( ChannelState , SinkProperties ) <EOL> from . message import ( FailedFastError , MethodReturnMessage ) <EOL> from . sink import ( ClientMessageSink , SinkProvider ) <EOL> from . varz import ( AggregateTimer , Counter , Source , VarzBase ) <EOL> ROOT_LOG = logging . getLogger ( '<STR_LIT>' ) <EOL> class ResurrectorSink ( ClientMessageSink ) : <EOL> """<STR_LIT>""" <EOL> class Varz ( VarzBase ) : <EOL> _VARZ_BASE_NAME = '<STR_LIT>' <EOL> _VARZ = { <EOL> '<STR_LIT>' : AggregateTimer , <EOL> '<STR_LIT>' : Counter , <EOL> } <EOL> def __init__ ( self , next_factory , sink_properties , global_properties ) : <EOL> endpoint = global_properties [ SinkProperties . Endpoint ] <EOL> service = global_properties [ SinkProperties . Label ] <EOL> endpoint_source = '<STR_LIT>' % ( endpoint . host , endpoint . port ) <EOL> self . endpoint = endpoint_source <EOL> self . _log = ROOT_LOG . getChild ( '<STR_LIT>' % ( service , endpoint_source ) ) <EOL> self . _properties = global_properties <EOL> self . _next_factory = next_factory <EOL> self . _resurrector = None <EOL> self . _down_on = None <EOL> self . _initial_wait_interval = sink_properties . initial_wait_interval <EOL> self . _max_wait_interval = sink_properties . max_wait_interval <EOL> self . _backoff_exponent = sink_properties . backoff_exponent <EOL> self . _varz = self . Varz ( Source ( service = service , <EOL> endpoint = endpoint_source ) ) <EOL> super ( ResurrectorSink , self ) . __init__ ( ) <EOL> def AsyncProcessRequest ( self , sink_stack , msg , stream , headers ) : <EOL> if self . next_sink is None : <EOL> gevent . sleep ( <NUM_LIT:0> ) <EOL> sink_stack . AsyncProcessResponseMessage ( MethodReturnMessage ( error = FailedFastError ( ) ) ) <EOL> else : <EOL> self . next_sink . AsyncProcessRequest ( sink_stack , msg , stream , headers ) <EOL> def AsyncProcessResponse ( self , sink_stack , context , stream , msg ) : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def _OnSinkFaulted ( self , val ) : <EOL> if not self . _down_on : <EOL> self . _down_on = time . time ( ) <EOL> sink , self . next_sink = self . next_sink , None <EOL> sink . Close ( ) <EOL> sink . on_faulted . Unsubscribe ( self . _OnSinkFaulted ) <EOL> self . _resurrector = gevent . spawn ( self . _TryResurrect ) <EOL> self . on_faulted . Set ( val ) <EOL> def _TryResurrect ( self ) : <EOL> if not self . _down_on : <EOL> return <EOL> last_attempt = self . _down_on <EOL> wait_interval = self . _initial_wait_interval <EOL> self . _log . info ( '<STR_LIT>' ) <EOL> while True : <EOL> gevent . sleep ( wait_interval ) <EOL> now = time . time ( ) <EOL> down_time , last_attempt = now - last_attempt , now <EOL> self . _varz . time_failed ( down_time ) <EOL> sink = self . _next_factory . CreateSink ( self . _properties ) <EOL> try : <EOL> self . _varz . reconnect_attempts ( ) <EOL> sink . Open ( ) . get ( ) <EOL> sink . on_faulted . Subscribe ( self . _OnSinkFaulted ) <EOL> self . next_sink = sink <EOL> self . _down_on = None <EOL> self . _log . info ( '<STR_LIT>' ) <EOL> return <EOL> except GreenletExit : <EOL> return <EOL> except : <EOL> sink . Close ( ) <EOL> wait_interval **= self . _backoff_exponent <EOL> wait_interval = min ( wait_interval , self . _max_wait_interval ) <EOL> def Open ( self ) : <EOL> if not self . next_sink : <EOL> self . next_sink = self . _next_factory . CreateSink ( self . _properties ) <EOL> self . next_sink . on_faulted . Subscribe ( self . _OnSinkFaulted ) <EOL> return self . next_sink . Open ( ) <EOL> def Close ( self ) : <EOL> if self . _resurrector : <EOL> self . _resurrector . kill ( block = False ) <EOL> self . _resurrector = None <EOL> self . _down_on = None <EOL> if self . next_sink : <EOL> self . next_sink . on_faulted . Unsubscribe ( self . _OnSinkFaulted ) <EOL> self . next_sink . Close ( ) <EOL> @ property <EOL> def state ( self ) : <EOL> if self . _down_on : <EOL> return ChannelState . Closed <EOL> elif not self . next_sink : <EOL> return ChannelState . Idle <EOL> else : <EOL> return self . next_sink . state <EOL> ResurrectorSink . Builder = SinkProvider ( ResurrectorSink , <EOL> initial_wait_interval = <NUM_LIT:5> , <EOL> max_wait_interval = <NUM_LIT> , <EOL> backoff_exponent = <NUM_LIT> ) </s>
