<s> """<STR_LIT>""" <EOL> import gevent <EOL> from gevent import event , coros <EOL> from pika . credentials import PlainCredentials <EOL> from pika . connection import ConnectionParameters <EOL> from pika . adapters import SelectConnection <EOL> from pika import channel as pikachannel <EOL> from pika . exceptions import NoFreeChannels <EOL> from pyon . core . bootstrap import CFG , get_sys_name <EOL> from pyon . net import channel <EOL> from pyon . util . async import blocking_cb <EOL> from pyon . util . containers import for_name <EOL> from pyon . util . log import log <EOL> from pyon . util . pool import IDPool <EOL> from pyon . net . transport import LocalTransport , LocalRouter , AMQPTransport , ComposableTransport <EOL> from collections import defaultdict <EOL> import traceback <EOL> class BaseNode ( object ) : <EOL> """<STR_LIT:U+0020>""" <EOL> def __init__ ( self ) : <EOL> self . client = None <EOL> self . running = False <EOL> self . ready = event . Event ( ) <EOL> self . _lock = coros . RLock ( ) <EOL> self . interceptors = { } <EOL> def on_connection_open ( self , client ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" ) <EOL> log . debug ( "<STR_LIT>" % str ( client ) ) <EOL> client . add_on_close_callback ( self . on_connection_close ) <EOL> self . client = client <EOL> self . start_node ( ) <EOL> def on_connection_close ( self , * a ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" ) <EOL> def start_node ( self ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . running = True <EOL> self . ready . set ( ) <EOL> def stop_node ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . running = False <EOL> def channel ( self , ch_type ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" ) <EOL> def _new_channel ( self , ch_type , ch_number = None , transport = None ) : <EOL> """<STR_LIT>""" <EOL> chan = ch_type ( ) <EOL> new_transport = self . _new_transport ( ch_number = ch_number ) <EOL> if transport is not None : <EOL> new_transport = ComposableTransport ( transport , new_transport , * ComposableTransport . common_methods ) <EOL> chan . on_channel_open ( new_transport ) <EOL> return chan <EOL> def _new_transport ( self , ch_number = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def setup_interceptors ( self , interceptor_cfg ) : <EOL> stack = interceptor_cfg [ "<STR_LIT>" ] <EOL> defs = interceptor_cfg [ "<STR_LIT>" ] <EOL> interceptors = defaultdict ( list ) <EOL> by_name_dict = { } <EOL> for type_and_direction in stack : <EOL> interceptor_names = stack [ type_and_direction ] <EOL> for name in interceptor_names : <EOL> if name in by_name_dict : <EOL> classinst = by_name_dict [ name ] <EOL> else : <EOL> interceptor_def = defs [ name ] <EOL> modpath , classname = interceptor_def [ '<STR_LIT:class>' ] . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> classinst = for_name ( modpath , classname ) <EOL> classinst . configure ( config = interceptor_def [ "<STR_LIT>" ] if "<STR_LIT>" in interceptor_def else None ) <EOL> by_name_dict [ name ] = classinst <EOL> interceptors [ type_and_direction ] . append ( classinst ) <EOL> self . interceptors = dict ( interceptors ) <EOL> class NodeB ( BaseNode ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> BaseNode . __init__ ( self ) <EOL> self . _pool = IDPool ( ) <EOL> self . _bidir_pool = { } <EOL> self . _pool_map = { } <EOL> self . _dead_pool = [ ] <EOL> BaseNode . __init__ ( self ) <EOL> def stop_node ( self ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" , self . running ) <EOL> if self . running : <EOL> self . _destroy_pool ( ) <EOL> self . client . close ( ) <EOL> BaseNode . stop_node ( self ) <EOL> def _destroy_pool ( self ) : <EOL> """<STR_LIT>""" <EOL> for chan in self . _bidir_pool . itervalues ( ) : <EOL> if chan . _recv_name : <EOL> chan . _destroy_queue ( ) <EOL> def _new_transport ( self , ch_number = None ) : <EOL> """<STR_LIT>""" <EOL> amq_chan = blocking_cb ( self . client . channel , '<STR_LIT>' , channel_number = ch_number ) <EOL> if amq_chan is None : <EOL> log . error ( "<STR_LIT>" , ch_number ) <EOL> from pyon . container . cc import Container <EOL> if Container . instance is not None : <EOL> Container . instance . fail_fast ( "<STR_LIT>" , True ) <EOL> raise StandardError ( "<STR_LIT>" % ch_number ) <EOL> transport = AMQPTransport ( amq_chan ) <EOL> self . client . _pending . remove ( transport . channel_number ) <EOL> transport . qos_impl ( prefetch_count = CFG . get_safe ( '<STR_LIT>' , <NUM_LIT:1> ) ) <EOL> return transport <EOL> def _check_pooled_channel_health ( self , ch ) : <EOL> """<STR_LIT>""" <EOL> cbs = self . client . callbacks . _callbacks <EOL> if "<STR_LIT>" not in cbs [ ch . get_channel_id ( ) ] . iterkeys ( ) : <EOL> return False <EOL> return True <EOL> def channel ( self , ch_type , transport = None ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> if ch_type == channel . BidirClientChannel and not ch_type . _queue_auto_delete : <EOL> attempts = <NUM_LIT:5> <EOL> while attempts > <NUM_LIT:0> : <EOL> attempts -= <NUM_LIT:1> <EOL> chid = self . _pool . get_id ( ) <EOL> if chid in self . _bidir_pool : <EOL> log . debug ( "<STR_LIT>" , chid ) <EOL> assert not chid in self . _pool_map . values ( ) <EOL> ch = self . _bidir_pool [ chid ] <EOL> if not self . _check_pooled_channel_health ( ch ) : <EOL> log . warning ( "<STR_LIT>" , ch . get_channel_id ( ) ) <EOL> self . _pool . release_id ( chid ) <EOL> self . _dead_pool . append ( ch ) <EOL> del self . _bidir_pool [ chid ] <EOL> assert ch . _close_callback == self . on_channel_request_close <EOL> ch . _close_callback = None <EOL> ch . close ( ) <EOL> continue <EOL> self . _pool_map [ ch . get_channel_id ( ) ] = chid <EOL> else : <EOL> log . debug ( "<STR_LIT>" , chid ) <EOL> ch = self . _new_channel ( ch_type , transport = transport ) <EOL> ch . set_close_callback ( self . on_channel_request_close ) <EOL> self . _bidir_pool [ chid ] = ch <EOL> self . _pool_map [ ch . get_channel_id ( ) ] = chid <EOL> break <EOL> else : <EOL> raise StandardError ( "<STR_LIT>" ) <EOL> else : <EOL> ch = self . _new_channel ( ch_type , transport = transport ) <EOL> assert ch <EOL> return ch <EOL> def on_channel_request_close ( self , ch ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" , ch . __class__ , ch . get_channel_id ( ) ) <EOL> assert ch . get_channel_id ( ) in self . _pool_map <EOL> with self . _lock : <EOL> chid = self . _pool_map . pop ( ch . get_channel_id ( ) ) <EOL> log . debug ( "<STR_LIT>" , ch . get_channel_id ( ) , chid ) <EOL> self . _pool . release_id ( chid ) <EOL> ch . reset ( ) <EOL> if ch . _queue_auto_delete : <EOL> log . warn ( "<STR_LIT>" ) <EOL> self . _bidir_pool . pop ( chid ) <EOL> self . _pool . _ids_free . remove ( chid ) <EOL> def ioloop ( connection , name = None ) : <EOL> if name : <EOL> name = "<STR_LIT>" % name <EOL> import threading <EOL> threading . current_thread ( ) . name = name or "<STR_LIT>" <EOL> try : <EOL> log . debug ( "<STR_LIT>" ) <EOL> connection . ioloop . start ( ) <EOL> log . debug ( "<STR_LIT>" ) <EOL> except KeyboardInterrupt : <EOL> log . debug ( "<STR_LIT>" ) <EOL> connection . close ( ) <EOL> connection . ioloop . start ( ) <EOL> class PyonSelectConnection ( SelectConnection ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , parameters = None , on_open_callback = None , <EOL> reconnection_strategy = None ) : <EOL> SelectConnection . __init__ ( self , parameters = parameters , on_open_callback = on_open_callback , reconnection_strategy = reconnection_strategy ) <EOL> self . _bad_channel_numbers = set ( ) <EOL> self . _pending = set ( ) <EOL> def _next_channel_number ( self ) : <EOL> """<STR_LIT>""" <EOL> limit = self . parameters . channel_max or pikachannel . MAX_CHANNELS <EOL> available = set ( xrange ( <NUM_LIT:1> , limit + <NUM_LIT:1> ) ) <EOL> available . difference_update ( self . _channels . keys ( ) ) <EOL> available . difference_update ( self . _bad_channel_numbers ) <EOL> available . difference_update ( self . _pending ) <EOL> if len ( available ) == <NUM_LIT:0> : <EOL> raise NoFreeChannels ( ) <EOL> ch_num = min ( available ) <EOL> self . _pending . add ( ch_num ) <EOL> return ch_num <EOL> def mark_bad_channel ( self , ch_number ) : <EOL> log . debug ( "<STR_LIT>" , ch_number ) <EOL> self . _bad_channel_numbers . add ( ch_number ) <EOL> def make_node ( connection_params = None , name = None , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" ) <EOL> node = NodeB ( ) <EOL> connection_params = connection_params or CFG . server . amqp <EOL> credentials = PlainCredentials ( connection_params [ "<STR_LIT:username>" ] , connection_params [ "<STR_LIT:password>" ] ) <EOL> conn_parameters = ConnectionParameters ( host = connection_params [ "<STR_LIT:host>" ] , virtual_host = connection_params [ "<STR_LIT>" ] , port = connection_params [ "<STR_LIT:port>" ] , credentials = credentials ) <EOL> connection = PyonSelectConnection ( conn_parameters , node . on_connection_open ) <EOL> ioloop_process = gevent . spawn ( ioloop , connection , name = name ) <EOL> ioloop_process . _glname = "<STR_LIT>" <EOL> node . ready . wait ( timeout = timeout ) <EOL> return node , ioloop_process <EOL> class LocalNode ( BaseNode ) : <EOL> def __init__ ( self , router = None ) : <EOL> BaseNode . __init__ ( self ) <EOL> self . _own_router = True <EOL> if router is not None : <EOL> self . _local_router = router <EOL> self . _own_router = False <EOL> else : <EOL> self . _local_router = LocalRouter ( get_sys_name ( ) ) <EOL> self . _channel_id_pool = IDPool ( ) <EOL> def start_node ( self ) : <EOL> BaseNode . start_node ( self ) <EOL> if self . _own_router : <EOL> self . _local_router . start ( ) <EOL> def stop_node ( self ) : <EOL> if self . running : <EOL> if self . _own_router : <EOL> self . _local_router . stop ( ) <EOL> self . running = False <EOL> def _new_transport ( self , ch_number = None ) : <EOL> trans = LocalTransport ( self . _local_router , ch_number ) <EOL> return trans <EOL> def channel ( self , ch_type , transport = None ) : <EOL> ch = self . _new_channel ( ch_type , ch_number = self . _channel_id_pool . get_id ( ) , transport = transport ) <EOL> ch . _transport . add_on_close_callback ( self . _on_channel_close ) <EOL> return ch <EOL> def _on_channel_close ( self , ch , code , text ) : <EOL> log . debug ( "<STR_LIT>" , ch . channel_number ) <EOL> self . _channel_id_pool . release_id ( ch . channel_number ) <EOL> class LocalClient ( object ) : <EOL> def __init__ ( self ) : <EOL> pass <EOL> def add_on_close_callback ( self , cb ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> def make_local_node ( timeout = None , router = None ) : <EOL> if router is not None : <EOL> node = LocalNode ( router = router ) <EOL> else : <EOL> node = LocalNode ( ) <EOL> node . on_connection_open ( LocalClient ( ) ) <EOL> node . ready . wait ( timeout = timeout ) <EOL> return node , node . _local_router . gl_ioloop </s>
