<s> '''<STR_LIT>''' <EOL> from __future__ import absolute_import <EOL> import os <EOL> import copy <EOL> import errno <EOL> import signal <EOL> import hashlib <EOL> import logging <EOL> import weakref <EOL> from random import randint <EOL> import salt . auth <EOL> import salt . crypt <EOL> import salt . utils <EOL> import salt . utils . verify <EOL> import salt . utils . event <EOL> import salt . payload <EOL> import salt . transport . client <EOL> import salt . transport . server <EOL> import salt . transport . mixins . auth <EOL> from salt . exceptions import SaltReqTimeoutError <EOL> import zmq <EOL> import zmq . eventloop . ioloop <EOL> if not hasattr ( zmq . eventloop . ioloop , '<STR_LIT>' ) : <EOL> zmq . eventloop . ioloop . ZMQIOLoop = zmq . eventloop . ioloop . IOLoop <EOL> import zmq . eventloop . zmqstream <EOL> try : <EOL> import zmq . utils . monitor <EOL> HAS_ZMQ_MONITOR = True <EOL> except ImportError : <EOL> HAS_ZMQ_MONITOR = False <EOL> import tornado <EOL> import tornado . gen <EOL> import tornado . concurrent <EOL> import salt . ext . six as six <EOL> from Crypto . Cipher import PKCS1_OAEP <EOL> log = logging . getLogger ( __name__ ) <EOL> class AsyncZeroMQReqChannel ( salt . transport . client . ReqChannel ) : <EOL> '''<STR_LIT>''' <EOL> instance_map = weakref . WeakKeyDictionary ( ) <EOL> def __new__ ( cls , opts , ** kwargs ) : <EOL> '''<STR_LIT>''' <EOL> io_loop = kwargs . get ( '<STR_LIT>' ) <EOL> if io_loop is None : <EOL> zmq . eventloop . ioloop . install ( ) <EOL> io_loop = tornado . ioloop . IOLoop . current ( ) <EOL> if io_loop not in cls . instance_map : <EOL> cls . instance_map [ io_loop ] = weakref . WeakValueDictionary ( ) <EOL> loop_instance_map = cls . instance_map [ io_loop ] <EOL> key = cls . __key ( opts , ** kwargs ) <EOL> if key not in loop_instance_map : <EOL> log . debug ( '<STR_LIT>' . format ( key ) ) <EOL> new_obj = object . __new__ ( cls ) <EOL> new_obj . __singleton_init__ ( opts , ** kwargs ) <EOL> loop_instance_map [ key ] = new_obj <EOL> log . trace ( '<STR_LIT>' . format ( id ( loop_instance_map ) , key , os . getpid ( ) ) ) <EOL> else : <EOL> log . debug ( '<STR_LIT>' . format ( key ) ) <EOL> try : <EOL> return loop_instance_map [ key ] <EOL> except KeyError : <EOL> log . debug ( '<STR_LIT>' . format ( key ) ) <EOL> new_obj = object . __new__ ( cls ) <EOL> new_obj . __singleton_init__ ( opts , ** kwargs ) <EOL> loop_instance_map [ key ] = new_obj <EOL> return loop_instance_map [ key ] <EOL> def __deepcopy__ ( self , memo ) : <EOL> cls = self . __class__ <EOL> result = cls . __new__ ( cls , copy . deepcopy ( self . opts , memo ) ) <EOL> memo [ id ( self ) ] = result <EOL> for key in self . __dict__ : <EOL> if key in ( '<STR_LIT>' , ) : <EOL> continue <EOL> if key == '<STR_LIT>' : <EOL> setattr ( result , key , <EOL> AsyncReqMessageClient ( result . opts , <EOL> self . master_uri , <EOL> io_loop = result . _io_loop ) ) <EOL> continue <EOL> setattr ( result , key , copy . deepcopy ( self . __dict__ [ key ] , memo ) ) <EOL> return result <EOL> @ classmethod <EOL> def __key ( cls , opts , ** kwargs ) : <EOL> return ( opts [ '<STR_LIT>' ] , <EOL> opts [ '<STR_LIT:id>' ] , <EOL> kwargs . get ( '<STR_LIT>' , opts . get ( '<STR_LIT>' ) ) , <EOL> kwargs . get ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ) <EOL> def __init__ ( self , opts , ** kwargs ) : <EOL> pass <EOL> def __singleton_init__ ( self , opts , ** kwargs ) : <EOL> self . opts = dict ( opts ) <EOL> self . ttype = '<STR_LIT>' <EOL> self . crypt = kwargs . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if '<STR_LIT>' in kwargs : <EOL> self . opts [ '<STR_LIT>' ] = kwargs [ '<STR_LIT>' ] <EOL> self . _io_loop = kwargs . get ( '<STR_LIT>' ) <EOL> if self . _io_loop is None : <EOL> zmq . eventloop . ioloop . install ( ) <EOL> self . _io_loop = tornado . ioloop . IOLoop . current ( ) <EOL> if self . crypt != '<STR_LIT>' : <EOL> self . auth = salt . crypt . AsyncAuth ( self . opts , io_loop = self . _io_loop ) <EOL> self . message_client = AsyncReqMessageClient ( self . opts , <EOL> self . master_uri , <EOL> io_loop = self . _io_loop , <EOL> ) <EOL> def __del__ ( self ) : <EOL> '''<STR_LIT>''' <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . message_client . destroy ( ) <EOL> else : <EOL> log . debug ( '<STR_LIT>' ) <EOL> @ property <EOL> def master_uri ( self ) : <EOL> return self . opts [ '<STR_LIT>' ] <EOL> def _package_load ( self , load ) : <EOL> return { <EOL> '<STR_LIT>' : self . crypt , <EOL> '<STR_LIT>' : load , <EOL> } <EOL> @ tornado . gen . coroutine <EOL> def crypted_transfer_decode_dictentry ( self , load , dictkey = None , tries = <NUM_LIT:3> , timeout = <NUM_LIT> ) : <EOL> if not self . auth . authenticated : <EOL> yield self . auth . authenticate ( ) <EOL> ret = yield self . message_client . send ( <EOL> self . _package_load ( self . auth . crypticle . dumps ( load ) ) , <EOL> timeout = timeout , <EOL> tries = tries , <EOL> ) <EOL> key = self . auth . get_keys ( ) <EOL> cipher = PKCS1_OAEP . new ( key ) <EOL> aes = cipher . decrypt ( ret [ '<STR_LIT:key>' ] ) <EOL> pcrypt = salt . crypt . Crypticle ( self . opts , aes ) <EOL> raise tornado . gen . Return ( pcrypt . loads ( ret [ dictkey ] ) ) <EOL> @ tornado . gen . coroutine <EOL> def _crypted_transfer ( self , load , tries = <NUM_LIT:3> , timeout = <NUM_LIT> ) : <EOL> '''<STR_LIT>''' <EOL> @ tornado . gen . coroutine <EOL> def _do_transfer ( ) : <EOL> data = yield self . message_client . send ( <EOL> self . _package_load ( self . auth . crypticle . dumps ( load ) ) , <EOL> timeout = timeout , <EOL> tries = tries , <EOL> ) <EOL> if data : <EOL> data = self . auth . crypticle . loads ( data ) <EOL> raise tornado . gen . Return ( data ) <EOL> if not self . auth . authenticated : <EOL> yield self . auth . authenticate ( ) <EOL> try : <EOL> ret = yield _do_transfer ( ) <EOL> except salt . crypt . AuthenticationError : <EOL> yield self . auth . authenticate ( ) <EOL> ret = yield _do_transfer ( ) <EOL> raise tornado . gen . Return ( ret ) <EOL> @ tornado . gen . coroutine <EOL> def _uncrypted_transfer ( self , load , tries = <NUM_LIT:3> , timeout = <NUM_LIT> ) : <EOL> '''<STR_LIT>''' <EOL> ret = yield self . message_client . send ( <EOL> self . _package_load ( load ) , <EOL> timeout = timeout , <EOL> tries = tries , <EOL> ) <EOL> raise tornado . gen . Return ( ret ) <EOL> @ tornado . gen . coroutine <EOL> def send ( self , load , tries = <NUM_LIT:3> , timeout = <NUM_LIT> ) : <EOL> '''<STR_LIT>''' <EOL> if self . crypt == '<STR_LIT>' : <EOL> ret = yield self . _uncrypted_transfer ( load , tries = tries , timeout = timeout ) <EOL> else : <EOL> ret = yield self . _crypted_transfer ( load , tries = tries , timeout = timeout ) <EOL> raise tornado . gen . Return ( ret ) <EOL> class AsyncZeroMQPubChannel ( salt . transport . mixins . auth . AESPubClientMixin , salt . transport . client . AsyncPubChannel ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , <EOL> opts , <EOL> ** kwargs ) : <EOL> self . opts = opts <EOL> self . ttype = '<STR_LIT>' <EOL> self . io_loop = kwargs . get ( '<STR_LIT>' ) <EOL> if self . io_loop is None : <EOL> zmq . eventloop . ioloop . install ( ) <EOL> self . io_loop = tornado . ioloop . IOLoop . current ( ) <EOL> self . hexid = hashlib . sha1 ( six . b ( self . opts [ '<STR_LIT:id>' ] ) ) . hexdigest ( ) <EOL> self . auth = salt . crypt . AsyncAuth ( self . opts , io_loop = self . io_loop ) <EOL> self . serial = salt . payload . Serial ( self . opts ) <EOL> self . context = zmq . Context ( ) <EOL> self . _socket = self . context . socket ( zmq . SUB ) <EOL> if self . opts [ '<STR_LIT>' ] : <EOL> self . _socket . setsockopt ( zmq . SUBSCRIBE , '<STR_LIT>' ) <EOL> self . _socket . setsockopt ( zmq . SUBSCRIBE , self . hexid ) <EOL> else : <EOL> self . _socket . setsockopt ( zmq . SUBSCRIBE , '<STR_LIT>' ) <EOL> self . _socket . setsockopt ( zmq . IDENTITY , self . opts [ '<STR_LIT:id>' ] ) <EOL> if hasattr ( zmq , '<STR_LIT>' ) : <EOL> self . _socket . setsockopt ( <EOL> zmq . TCP_KEEPALIVE , self . opts [ '<STR_LIT>' ] <EOL> ) <EOL> self . _socket . setsockopt ( <EOL> zmq . TCP_KEEPALIVE_IDLE , self . opts [ '<STR_LIT>' ] <EOL> ) <EOL> self . _socket . setsockopt ( <EOL> zmq . TCP_KEEPALIVE_CNT , self . opts [ '<STR_LIT>' ] <EOL> ) <EOL> self . _socket . setsockopt ( <EOL> zmq . TCP_KEEPALIVE_INTVL , self . opts [ '<STR_LIT>' ] <EOL> ) <EOL> recon_delay = self . opts [ '<STR_LIT>' ] <EOL> if self . opts [ '<STR_LIT>' ] : <EOL> recon_delay = randint ( self . opts [ '<STR_LIT>' ] , <EOL> self . opts [ '<STR_LIT>' ] + self . opts [ '<STR_LIT>' ] <EOL> ) <EOL> log . debug ( "<STR_LIT>" . format ( <EOL> self . opts [ '<STR_LIT>' ] , <EOL> self . opts [ '<STR_LIT>' ] + self . opts [ '<STR_LIT>' ] , <EOL> recon_delay ) <EOL> ) <EOL> log . debug ( "<STR_LIT>" . format ( recon_delay ) ) <EOL> self . _socket . setsockopt ( zmq . RECONNECT_IVL , recon_delay ) <EOL> if hasattr ( zmq , '<STR_LIT>' ) : <EOL> log . debug ( "<STR_LIT>" . format ( <EOL> self . opts [ '<STR_LIT>' ] + self . opts [ '<STR_LIT>' ] ) <EOL> ) <EOL> self . _socket . setsockopt ( <EOL> zmq . RECONNECT_IVL_MAX , self . opts [ '<STR_LIT>' ] <EOL> ) <EOL> if self . opts [ '<STR_LIT>' ] is True and hasattr ( zmq , '<STR_LIT>' ) : <EOL> self . _socket . setsockopt ( zmq . IPV4ONLY , <NUM_LIT:0> ) <EOL> if HAS_ZMQ_MONITOR and self . opts [ '<STR_LIT>' ] : <EOL> self . _monitor = ZeroMQSocketMonitor ( self . _socket ) <EOL> self . _monitor . start_io_loop ( self . io_loop ) <EOL> def destroy ( self ) : <EOL> if hasattr ( self , '<STR_LIT>' ) and self . _monitor is not None : <EOL> self . _monitor . stop ( ) <EOL> self . _monitor = None <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . _stream . io_loop . remove_handler ( self . _stream . socket ) <EOL> self . _stream . socket . close ( <NUM_LIT:0> ) <EOL> elif hasattr ( self , '<STR_LIT>' ) : <EOL> self . _socket . close ( <NUM_LIT:0> ) <EOL> if hasattr ( self , '<STR_LIT>' ) and self . context . closed is False : <EOL> self . context . term ( ) <EOL> def __del__ ( self ) : <EOL> self . destroy ( ) <EOL> @ tornado . gen . coroutine <EOL> def connect ( self ) : <EOL> if not self . auth . authenticated : <EOL> yield self . auth . authenticate ( ) <EOL> self . publish_port = self . auth . creds [ '<STR_LIT>' ] <EOL> self . _socket . connect ( self . master_pub ) <EOL> @ property <EOL> def master_pub ( self ) : <EOL> '''<STR_LIT>''' <EOL> return '<STR_LIT>' . format ( ip = self . opts [ '<STR_LIT>' ] , <EOL> port = self . publish_port ) <EOL> @ tornado . gen . coroutine <EOL> def _decode_messages ( self , messages ) : <EOL> '''<STR_LIT>''' <EOL> messages_len = len ( messages ) <EOL> if messages_len == <NUM_LIT:1> : <EOL> payload = self . serial . loads ( messages [ <NUM_LIT:0> ] ) <EOL> elif messages_len == <NUM_LIT:2> : <EOL> if messages [ <NUM_LIT:0> ] not in ( '<STR_LIT>' , self . hexid ) : <EOL> log . debug ( '<STR_LIT>' . format ( messages [ <NUM_LIT:0> ] ) ) <EOL> raise tornado . gen . Return ( None ) <EOL> payload = self . serial . loads ( messages [ <NUM_LIT:1> ] ) <EOL> else : <EOL> raise Exception ( ( '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( len ( messages_len ) ) ) <EOL> ret = yield self . _decode_payload ( payload ) <EOL> raise tornado . gen . Return ( ret ) <EOL> @ property <EOL> def stream ( self ) : <EOL> '''<STR_LIT>''' <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . _stream = zmq . eventloop . zmqstream . ZMQStream ( self . _socket , io_loop = self . io_loop ) <EOL> return self . _stream <EOL> def on_recv ( self , callback ) : <EOL> '''<STR_LIT>''' <EOL> if callback is None : <EOL> return self . stream . on_recv ( None ) <EOL> @ tornado . gen . coroutine <EOL> def wrap_callback ( messages ) : <EOL> payload = yield self . _decode_messages ( messages ) <EOL> if payload is not None : <EOL> callback ( payload ) <EOL> return self . stream . on_recv ( wrap_callback ) <EOL> class ZeroMQReqServerChannel ( salt . transport . mixins . auth . AESReqServerMixin , salt . transport . server . ReqServerChannel ) : <EOL> def __init__ ( self , opts ) : <EOL> salt . transport . server . ReqServerChannel . __init__ ( self , opts ) <EOL> self . _closing = False <EOL> def zmq_device ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . __setup_signals ( ) <EOL> salt . utils . appendproctitle ( '<STR_LIT>' ) <EOL> self . context = zmq . Context ( self . opts [ '<STR_LIT>' ] ) <EOL> self . uri = '<STR_LIT>' . format ( ** self . opts ) <EOL> self . clients = self . context . socket ( zmq . ROUTER ) <EOL> if self . opts [ '<STR_LIT>' ] is True and hasattr ( zmq , '<STR_LIT>' ) : <EOL> self . clients . setsockopt ( zmq . IPV4ONLY , <NUM_LIT:0> ) <EOL> if HAS_ZMQ_MONITOR and self . opts [ '<STR_LIT>' ] : <EOL> import threading <EOL> self . _monitor = ZeroMQSocketMonitor ( self . clients ) <EOL> t = threading . Thread ( target = self . _monitor . start_poll ) <EOL> t . start ( ) <EOL> self . workers = self . context . socket ( zmq . DEALER ) <EOL> if self . opts . get ( '<STR_LIT>' , '<STR_LIT>' ) == '<STR_LIT>' : <EOL> self . w_uri = '<STR_LIT>' . format ( <EOL> self . opts . get ( '<STR_LIT>' , <NUM_LIT> ) <EOL> ) <EOL> else : <EOL> self . w_uri = '<STR_LIT>' . format ( <EOL> os . path . join ( self . opts [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> ) <EOL> log . info ( '<STR_LIT>' ) <EOL> self . clients . bind ( self . uri ) <EOL> self . workers . bind ( self . w_uri ) <EOL> while True : <EOL> if self . clients . closed or self . workers . closed : <EOL> break <EOL> try : <EOL> zmq . device ( zmq . QUEUE , self . clients , self . workers ) <EOL> except zmq . ZMQError as exc : <EOL> if exc . errno == errno . EINTR : <EOL> continue <EOL> raise exc <EOL> except ( KeyboardInterrupt , SystemExit ) : <EOL> break <EOL> def close ( self ) : <EOL> '''<STR_LIT>''' <EOL> if self . _closing : <EOL> return <EOL> self . _closing = True <EOL> if hasattr ( self , '<STR_LIT>' ) and self . _monitor is not None : <EOL> self . _monitor . stop ( ) <EOL> self . _monitor = None <EOL> if hasattr ( self , '<STR_LIT>' ) and self . clients . closed is False : <EOL> self . clients . close ( ) <EOL> if hasattr ( self , '<STR_LIT>' ) and self . workers . closed is False : <EOL> self . workers . close ( ) <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . stream . close ( ) <EOL> if hasattr ( self , '<STR_LIT>' ) and self . context . closed is False : <EOL> self . context . term ( ) <EOL> def pre_fork ( self , process_manager ) : <EOL> '''<STR_LIT>''' <EOL> salt . transport . mixins . auth . AESReqServerMixin . pre_fork ( self , process_manager ) <EOL> process_manager . add_process ( self . zmq_device ) <EOL> def post_fork ( self , payload_handler , io_loop ) : <EOL> '''<STR_LIT>''' <EOL> self . payload_handler = payload_handler <EOL> self . io_loop = io_loop <EOL> self . context = zmq . Context ( <NUM_LIT:1> ) <EOL> self . _socket = self . context . socket ( zmq . REP ) <EOL> if self . opts . get ( '<STR_LIT>' , '<STR_LIT>' ) == '<STR_LIT>' : <EOL> self . w_uri = '<STR_LIT>' . format ( <EOL> self . opts . get ( '<STR_LIT>' , <NUM_LIT> ) <EOL> ) <EOL> else : <EOL> self . w_uri = '<STR_LIT>' . format ( <EOL> os . path . join ( self . opts [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> ) <EOL> log . info ( '<STR_LIT>' . format ( self . w_uri ) ) <EOL> self . _socket . connect ( self . w_uri ) <EOL> salt . transport . mixins . auth . AESReqServerMixin . post_fork ( self , payload_handler , io_loop ) <EOL> self . stream = zmq . eventloop . zmqstream . ZMQStream ( self . _socket , io_loop = self . io_loop ) <EOL> self . stream . on_recv_stream ( self . handle_message ) <EOL> @ tornado . gen . coroutine <EOL> def handle_message ( self , stream , payload ) : <EOL> '''<STR_LIT>''' <EOL> try : <EOL> payload = self . serial . loads ( payload [ <NUM_LIT:0> ] ) <EOL> payload = self . _decode_payload ( payload ) <EOL> except Exception as exc : <EOL> log . error ( '<STR_LIT>' , type ( exc ) . __name__ , exc ) <EOL> stream . send ( self . serial . dumps ( '<STR_LIT>' ) ) <EOL> raise tornado . gen . Return ( ) <EOL> if not isinstance ( payload , dict ) or not isinstance ( payload . get ( '<STR_LIT>' ) , dict ) : <EOL> log . error ( '<STR_LIT>' . format ( payload , payload . get ( '<STR_LIT>' ) ) ) <EOL> stream . send ( self . serial . dumps ( '<STR_LIT>' ) ) <EOL> raise tornado . gen . Return ( ) <EOL> if payload [ '<STR_LIT>' ] == '<STR_LIT>' and payload . get ( '<STR_LIT>' , { } ) . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> stream . send ( self . serial . dumps ( self . _auth ( payload [ '<STR_LIT>' ] ) ) ) <EOL> raise tornado . gen . Return ( ) <EOL> try : <EOL> ret , req_opts = yield self . payload_handler ( payload ) <EOL> except Exception as e : <EOL> stream . send ( '<STR_LIT>' ) <EOL> log . error ( '<STR_LIT>' , exc_info = True ) <EOL> raise tornado . gen . Return ( ) <EOL> req_fun = req_opts . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if req_fun == '<STR_LIT>' : <EOL> stream . send ( self . serial . dumps ( ret ) ) <EOL> elif req_fun == '<STR_LIT>' : <EOL> stream . send ( self . serial . dumps ( self . crypticle . dumps ( ret ) ) ) <EOL> elif req_fun == '<STR_LIT>' : <EOL> stream . send ( self . serial . dumps ( self . _encrypt_private ( ret , <EOL> req_opts [ '<STR_LIT:key>' ] , <EOL> req_opts [ '<STR_LIT>' ] , <EOL> ) ) ) <EOL> else : <EOL> log . error ( '<STR_LIT>' . format ( req_fun ) ) <EOL> stream . send ( '<STR_LIT>' ) <EOL> raise tornado . gen . Return ( ) <EOL> def __setup_signals ( self ) : <EOL> signal . signal ( signal . SIGINT , self . _handle_signals ) <EOL> signal . signal ( signal . SIGTERM , self . _handle_signals ) <EOL> def _handle_signals ( self , signum , sigframe ) : <EOL> msg = '<STR_LIT>' . format ( self . __class__ . __name__ ) <EOL> if signum == signal . SIGINT : <EOL> msg += '<STR_LIT>' <EOL> elif signum == signal . SIGTERM : <EOL> msg += '<STR_LIT>' <EOL> msg += '<STR_LIT>' <EOL> log . debug ( msg ) <EOL> exit ( salt . defaults . exitcodes . EX_OK ) <EOL> class ZeroMQPubServerChannel ( salt . transport . server . PubServerChannel ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , opts ) : <EOL> self . opts = opts <EOL> self . serial = salt . payload . Serial ( self . opts ) <EOL> self . ckminions = salt . utils . minions . CkMinions ( self . opts ) <EOL> def connect ( self ) : <EOL> return tornado . gen . sleep ( <NUM_LIT:5> ) <EOL> def _publish_daemon ( self ) : <EOL> '''<STR_LIT>''' <EOL> salt . utils . appendproctitle ( self . __class__ . __name__ ) <EOL> context = zmq . Context ( <NUM_LIT:1> ) <EOL> pub_sock = context . socket ( zmq . PUB ) <EOL> try : <EOL> pub_sock . setsockopt ( zmq . HWM , self . opts . get ( '<STR_LIT>' , <NUM_LIT:1000> ) ) <EOL> except AttributeError : <EOL> pub_sock . setsockopt ( zmq . SNDHWM , self . opts . get ( '<STR_LIT>' , <NUM_LIT:1000> ) ) <EOL> pub_sock . setsockopt ( zmq . RCVHWM , self . opts . get ( '<STR_LIT>' , <NUM_LIT:1000> ) ) <EOL> if self . opts [ '<STR_LIT>' ] is True and hasattr ( zmq , '<STR_LIT>' ) : <EOL> pub_sock . setsockopt ( zmq . IPV4ONLY , <NUM_LIT:0> ) <EOL> pub_uri = '<STR_LIT>' . format ( ** self . opts ) <EOL> pull_sock = context . socket ( zmq . PULL ) <EOL> if self . opts . get ( '<STR_LIT>' , '<STR_LIT>' ) == '<STR_LIT>' : <EOL> pull_uri = '<STR_LIT>' . format ( <EOL> self . opts . get ( '<STR_LIT>' , <NUM_LIT> ) <EOL> ) <EOL> else : <EOL> pull_uri = '<STR_LIT>' . format ( <EOL> os . path . join ( self . opts [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> ) <EOL> salt . utils . zeromq . check_ipc_path_max_len ( pull_uri ) <EOL> log . info ( '<STR_LIT>' . format ( pub_uri ) ) <EOL> pub_sock . bind ( pub_uri ) <EOL> log . info ( '<STR_LIT>' . format ( pull_uri ) ) <EOL> old_umask = os . umask ( <NUM_LIT> ) <EOL> try : <EOL> pull_sock . bind ( pull_uri ) <EOL> finally : <EOL> os . umask ( old_umask ) <EOL> try : <EOL> while True : <EOL> try : <EOL> package = pull_sock . recv ( ) <EOL> unpacked_package = salt . payload . unpackage ( package ) <EOL> payload = unpacked_package [ '<STR_LIT>' ] <EOL> if self . opts [ '<STR_LIT>' ] : <EOL> if '<STR_LIT>' in unpacked_package : <EOL> for topic in unpacked_package [ '<STR_LIT>' ] : <EOL> htopic = hashlib . sha1 ( topic ) . hexdigest ( ) <EOL> pub_sock . send ( htopic , flags = zmq . SNDMORE ) <EOL> pub_sock . send ( payload ) <EOL> else : <EOL> pub_sock . send ( '<STR_LIT>' , flags = zmq . SNDMORE ) <EOL> pub_sock . send ( payload ) <EOL> else : <EOL> pub_sock . send ( payload ) <EOL> except zmq . ZMQError as exc : <EOL> if exc . errno == errno . EINTR : <EOL> continue <EOL> raise exc <EOL> except KeyboardInterrupt : <EOL> if pub_sock . closed is False : <EOL> pub_sock . setsockopt ( zmq . LINGER , <NUM_LIT:1> ) <EOL> pub_sock . close ( ) <EOL> if pull_sock . closed is False : <EOL> pull_sock . setsockopt ( zmq . LINGER , <NUM_LIT:1> ) <EOL> pull_sock . close ( ) <EOL> if context . closed is False : <EOL> context . term ( ) <EOL> def pre_fork ( self , process_manager ) : <EOL> '''<STR_LIT>''' <EOL> process_manager . add_process ( self . _publish_daemon ) <EOL> def publish ( self , load ) : <EOL> '''<STR_LIT>''' <EOL> payload = { '<STR_LIT>' : '<STR_LIT>' } <EOL> crypticle = salt . crypt . Crypticle ( self . opts , salt . master . SMaster . secrets [ '<STR_LIT>' ] [ '<STR_LIT>' ] . value ) <EOL> payload [ '<STR_LIT>' ] = crypticle . dumps ( load ) <EOL> if self . opts [ '<STR_LIT>' ] : <EOL> master_pem_path = os . path . join ( self . opts [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> log . debug ( "<STR_LIT>" ) <EOL> payload [ '<STR_LIT>' ] = salt . crypt . sign_message ( master_pem_path , payload [ '<STR_LIT>' ] ) <EOL> context = zmq . Context ( <NUM_LIT:1> ) <EOL> pub_sock = context . socket ( zmq . PUSH ) <EOL> if self . opts . get ( '<STR_LIT>' , '<STR_LIT>' ) == '<STR_LIT>' : <EOL> pull_uri = '<STR_LIT>' . format ( <EOL> self . opts . get ( '<STR_LIT>' , <NUM_LIT> ) <EOL> ) <EOL> else : <EOL> pull_uri = '<STR_LIT>' . format ( <EOL> os . path . join ( self . opts [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> ) <EOL> pub_sock . connect ( pull_uri ) <EOL> int_payload = { '<STR_LIT>' : self . serial . dumps ( payload ) } <EOL> if load [ '<STR_LIT>' ] == '<STR_LIT:list>' : <EOL> int_payload [ '<STR_LIT>' ] = load [ '<STR_LIT>' ] <EOL> match_targets = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:list>" ] <EOL> if self . opts [ '<STR_LIT>' ] and load [ '<STR_LIT>' ] in match_targets : <EOL> match_ids = self . ckminions . check_minions ( load [ '<STR_LIT>' ] , <EOL> expr_form = load [ '<STR_LIT>' ] <EOL> ) <EOL> log . debug ( "<STR_LIT>" . format ( match_ids ) ) <EOL> int_payload [ '<STR_LIT>' ] = match_ids <EOL> pub_sock . send ( self . serial . dumps ( int_payload ) ) <EOL> pub_sock . close ( ) <EOL> context . term ( ) <EOL> class AsyncReqMessageClient ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , opts , addr , linger = <NUM_LIT:0> , io_loop = None ) : <EOL> '''<STR_LIT>''' <EOL> self . opts = opts <EOL> self . addr = addr <EOL> self . linger = linger <EOL> if io_loop is None : <EOL> zmq . eventloop . ioloop . install ( ) <EOL> tornado . ioloop . IOLoop . current ( ) <EOL> else : <EOL> self . io_loop = io_loop <EOL> self . serial = salt . payload . Serial ( self . opts ) <EOL> self . context = zmq . Context ( ) <EOL> self . _init_socket ( ) <EOL> self . send_queue = [ ] <EOL> self . send_future_map = { } <EOL> self . send_timeout_map = { } <EOL> def destroy ( self ) : <EOL> if hasattr ( self , '<STR_LIT>' ) and self . stream is not None : <EOL> if self . stream . socket : <EOL> self . stream . socket . close ( ) <EOL> self . stream . io_loop . remove_handler ( self . stream . socket ) <EOL> self . stream . socket = None <EOL> self . stream = None <EOL> self . socket . close ( ) <EOL> if self . context . closed is False : <EOL> self . context . term ( ) <EOL> def __del__ ( self ) : <EOL> self . destroy ( ) <EOL> def _init_socket ( self ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . stream . close ( ) <EOL> self . socket . close ( ) <EOL> del self . stream <EOL> del self . socket <EOL> self . socket = self . context . socket ( zmq . REQ ) <EOL> if hasattr ( zmq , '<STR_LIT>' ) : <EOL> self . socket . setsockopt ( <EOL> zmq . RECONNECT_IVL_MAX , <NUM_LIT> <EOL> ) <EOL> self . _set_tcp_keepalive ( ) <EOL> if self . addr . startswith ( '<STR_LIT>' ) : <EOL> if hasattr ( zmq , '<STR_LIT>' ) : <EOL> self . socket . setsockopt ( zmq . IPV6 , <NUM_LIT:1> ) <EOL> elif hasattr ( zmq , '<STR_LIT>' ) : <EOL> self . socket . setsockopt ( zmq . IPV4ONLY , <NUM_LIT:0> ) <EOL> self . socket . linger = self . linger <EOL> self . socket . connect ( self . addr ) <EOL> self . stream = zmq . eventloop . zmqstream . ZMQStream ( self . socket , io_loop = self . io_loop ) <EOL> def _set_tcp_keepalive ( self ) : <EOL> '''<STR_LIT>''' <EOL> if hasattr ( zmq , '<STR_LIT>' ) and self . opts : <EOL> if '<STR_LIT>' in self . opts : <EOL> self . socket . setsockopt ( <EOL> zmq . TCP_KEEPALIVE , self . opts [ '<STR_LIT>' ] <EOL> ) <EOL> if '<STR_LIT>' in self . opts : <EOL> self . socket . setsockopt ( <EOL> zmq . TCP_KEEPALIVE_IDLE , self . opts [ '<STR_LIT>' ] <EOL> ) <EOL> if '<STR_LIT>' in self . opts : <EOL> self . socket . setsockopt ( <EOL> zmq . TCP_KEEPALIVE_CNT , self . opts [ '<STR_LIT>' ] <EOL> ) <EOL> if '<STR_LIT>' in self . opts : <EOL> self . socket . setsockopt ( <EOL> zmq . TCP_KEEPALIVE_INTVL , self . opts [ '<STR_LIT>' ] <EOL> ) <EOL> @ tornado . gen . coroutine <EOL> def _internal_send_recv ( self ) : <EOL> while len ( self . send_queue ) > <NUM_LIT:0> : <EOL> message = self . send_queue [ <NUM_LIT:0> ] <EOL> future = self . send_future_map . get ( message , None ) <EOL> if future is None : <EOL> del self . send_queue [ <NUM_LIT:0> ] <EOL> continue <EOL> def mark_future ( msg ) : <EOL> if not future . done ( ) : <EOL> future . set_result ( self . serial . loads ( msg [ <NUM_LIT:0> ] ) ) <EOL> self . stream . on_recv ( mark_future ) <EOL> self . stream . send ( message ) <EOL> try : <EOL> ret = yield future <EOL> except : <EOL> self . _init_socket ( ) <EOL> del self . send_queue [ <NUM_LIT:0> ] <EOL> continue <EOL> del self . send_queue [ <NUM_LIT:0> ] <EOL> self . send_future_map . pop ( message , None ) <EOL> self . remove_message_timeout ( message ) <EOL> def remove_message_timeout ( self , message ) : <EOL> if message not in self . send_timeout_map : <EOL> return <EOL> timeout = self . send_timeout_map . pop ( message , None ) <EOL> if timeout is not None : <EOL> self . io_loop . remove_timeout ( timeout ) <EOL> def timeout_message ( self , message ) : <EOL> '''<STR_LIT>''' <EOL> future = self . send_future_map . pop ( message , None ) <EOL> if future is not None : <EOL> del self . send_timeout_map [ message ] <EOL> if future . attempts < future . tries : <EOL> future . attempts += <NUM_LIT:1> <EOL> log . debug ( '<STR_LIT>' . format ( future . attempts , future . tries ) ) <EOL> self . send ( <EOL> message , <EOL> timeout = future . timeout , <EOL> tries = future . tries , <EOL> future = future , <EOL> ) <EOL> else : <EOL> future . set_exception ( SaltReqTimeoutError ( '<STR_LIT>' ) ) <EOL> def send ( self , message , timeout = None , tries = <NUM_LIT:3> , future = None , callback = None ) : <EOL> '''<STR_LIT>''' <EOL> if future is None : <EOL> future = tornado . concurrent . Future ( ) <EOL> future . tries = tries <EOL> future . attempts = <NUM_LIT:0> <EOL> future . timeout = timeout <EOL> message = self . serial . dumps ( message ) <EOL> if callback is not None : <EOL> def handle_future ( future ) : <EOL> response = future . result ( ) <EOL> self . io_loop . add_callback ( callback , response ) <EOL> future . add_done_callback ( handle_future ) <EOL> self . send_future_map [ message ] = future <EOL> if timeout is not None : <EOL> send_timeout = self . io_loop . call_later ( timeout , self . timeout_message , message ) <EOL> self . send_timeout_map [ message ] = send_timeout <EOL> if len ( self . send_queue ) == <NUM_LIT:0> : <EOL> self . io_loop . spawn_callback ( self . _internal_send_recv ) <EOL> self . send_queue . append ( message ) <EOL> return future <EOL> class ZeroMQSocketMonitor ( object ) : <EOL> __EVENT_MAP = None <EOL> def __init__ ( self , socket ) : <EOL> '''<STR_LIT>''' <EOL> self . _socket = socket <EOL> self . _monitor_socket = self . _socket . get_monitor_socket ( ) <EOL> self . _monitor_stream = None <EOL> def start_io_loop ( self , io_loop ) : <EOL> log . trace ( "<STR_LIT>" ) <EOL> self . _monitor_stream = zmq . eventloop . zmqstream . ZMQStream ( self . _monitor_socket , io_loop = io_loop ) <EOL> self . _monitor_stream . on_recv ( self . monitor_callback ) <EOL> def start_poll ( self ) : <EOL> log . trace ( "<STR_LIT>" ) <EOL> while self . _monitor_socket is not None and self . _monitor_socket . poll ( ) : <EOL> msg = self . _monitor_socket . recv_multipart ( ) <EOL> self . monitor_callback ( msg ) <EOL> @ property <EOL> def event_map ( self ) : <EOL> if ZeroMQSocketMonitor . __EVENT_MAP is None : <EOL> event_map = { } <EOL> for name in dir ( zmq ) : <EOL> if name . startswith ( '<STR_LIT>' ) : <EOL> value = getattr ( zmq , name ) <EOL> event_map [ value ] = name <EOL> ZeroMQSocketMonitor . __EVENT_MAP = event_map <EOL> return ZeroMQSocketMonitor . __EVENT_MAP <EOL> def monitor_callback ( self , msg ) : <EOL> evt = zmq . utils . monitor . parse_monitor_message ( msg ) <EOL> evt [ '<STR_LIT:description>' ] = self . event_map [ evt [ '<STR_LIT>' ] ] <EOL> log . debug ( "<STR_LIT>" . format ( evt ) ) <EOL> if evt [ '<STR_LIT>' ] == zmq . EVENT_MONITOR_STOPPED : <EOL> self . stop ( ) <EOL> def stop ( self ) : <EOL> if self . _socket is None : <EOL> return <EOL> self . _socket . disable_monitor ( ) <EOL> self . _socket = None <EOL> self . _monitor_socket = None <EOL> if self . _monitor_stream is not None : <EOL> self . _monitor_stream . close ( ) <EOL> self . _monitor_stream = None <EOL> log . trace ( "<STR_LIT>" ) </s>
