<s> import cPickle <EOL> import json <EOL> import logging <EOL> import pika <EOL> import requests <EOL> from rackclient import exceptions <EOL> from rackclient . client import Client <EOL> LOG = logging . getLogger ( __name__ ) <EOL> LOG . setLevel ( logging . DEBUG ) <EOL> META_URL = '<STR_LIT>' <EOL> def get_rack_context ( proxy_port = <NUM_LIT> , client_version = "<STR_LIT:1>" , api_version = "<STR_LIT>" ) : <EOL> try : <EOL> resp = requests . get ( META_URL ) <EOL> metadata = json . loads ( resp . text ) [ "<STR_LIT>" ] <EOL> proxy_url = '<STR_LIT>' % ( <EOL> metadata [ "<STR_LIT>" ] , proxy_port , api_version ) <EOL> client = Client ( client_version , rack_url = proxy_url ) <EOL> proxy_info = client . proxy . get ( metadata [ "<STR_LIT>" ] ) <EOL> rack_ctx = type ( '<STR_LIT>' , ( object , ) , metadata ) <EOL> rack_ctx . ppid = getattr ( rack_ctx , "<STR_LIT>" , "<STR_LIT>" ) <EOL> rack_ctx . client = client <EOL> rack_ctx . fs_endpoint = proxy_info . fs_endpoint <EOL> rack_ctx . ipc_endpoint = proxy_info . ipc_endpoint <EOL> rack_ctx . shm_endpoint = proxy_info . shm_endpoint <EOL> try : <EOL> rack_ctx . client . processes . get ( rack_ctx . gid , rack_ctx . pid ) <EOL> except exceptions . NotFound : <EOL> msg = "<STR_LIT>" <EOL> raise exceptions . InvalidProcessError ( msg ) <EOL> if rack_ctx . ppid : <EOL> LOG . debug ( "<STR_LIT>" , rack_ctx . ppid ) <EOL> msg = _Messaging ( rack_ctx ) <EOL> msg . send_msg ( rack_ctx . ppid ) <EOL> while True : <EOL> receive_msg = msg . receive_msg ( <EOL> getattr ( rack_ctx , "<STR_LIT>" , <NUM_LIT> ) ) <EOL> if receive_msg and rack_ctx . ppid == receive_msg . get ( "<STR_LIT>" ) : <EOL> LOG . debug ( <EOL> "<STR_LIT>" , <EOL> receive_msg . get ( "<STR_LIT>" ) ) <EOL> break <EOL> except Exception as e : <EOL> msg = "<STR_LIT>" % e . message <EOL> raise exceptions . ProcessInitError ( msg ) <EOL> return rack_ctx <EOL> class _Messaging ( object ) : <EOL> def __init__ ( self , ctx ) : <EOL> self . ctx = ctx <EOL> connection_param = pika . ConnectionParameters ( self . ctx . proxy_ip ) <EOL> if self . ctx . ipc_endpoint : <EOL> connection_param = pika . ConnectionParameters ( self . ctx . ipc_endpoint ) <EOL> self . connection = pika . BlockingConnection ( connection_param ) <EOL> self . channel = self . connection . channel ( ) <EOL> self . channel . exchange_declare ( exchange = ctx . gid , type = '<STR_LIT>' ) <EOL> self . channel . queue_declare ( queue = ctx . pid ) <EOL> self . channel . queue_bind ( <EOL> exchange = ctx . gid , <EOL> queue = ctx . pid , <EOL> routing_key = ctx . gid + '<STR_LIT:.>' + ctx . pid ) <EOL> def receive_msg ( self , timeout_limit = <NUM_LIT> ) : <EOL> queue_name = self . ctx . pid <EOL> self . channel = self . connection . channel ( ) <EOL> receive = self . Receive ( ) <EOL> self . connection . add_timeout ( <EOL> deadline = int ( timeout_limit ) , <EOL> callback_method = receive . time_out ) <EOL> self . channel . basic_consume ( <EOL> receive . get_msg , <EOL> queue = queue_name , <EOL> no_ack = False ) <EOL> receive . channel = self . channel <EOL> self . channel . start_consuming ( ) <EOL> return receive . message <EOL> def send_msg ( self , target ) : <EOL> routing_key = self . ctx . gid + '<STR_LIT:.>' + target <EOL> send_dict = { '<STR_LIT>' : self . ctx . pid } <EOL> send_msg = cPickle . dumps ( send_dict ) <EOL> self . channel . basic_publish ( <EOL> exchange = self . ctx . gid , <EOL> routing_key = routing_key , <EOL> body = send_msg ) <EOL> class Receive ( object ) : <EOL> def __init__ ( self ) : <EOL> self . channel = None <EOL> self . message = None <EOL> def get_msg ( self , ch , method , properties , body ) : <EOL> self . message = cPickle . loads ( body ) <EOL> ch . basic_ack ( delivery_tag = method . delivery_tag ) <EOL> ch . stop_consuming ( ) <EOL> def time_out ( self ) : <EOL> self . channel . stop_consuming ( ) </s>
