<s> import socket <EOL> import uuid <EOL> import zmq <EOL> from ipyparallel . util import disambiguate_url <EOL> class EngineCommunicator ( object ) : <EOL> def __init__ ( self , interface = '<STR_LIT>' , identity = None ) : <EOL> self . _ctx = zmq . Context ( ) <EOL> self . socket = self . _ctx . socket ( zmq . XREP ) <EOL> self . pub = self . _ctx . socket ( zmq . PUB ) <EOL> self . sub = self . _ctx . socket ( zmq . SUB ) <EOL> self . identity = identity or bytes ( uuid . uuid4 ( ) ) <EOL> self . socket . IDENTITY = self . identity <EOL> self . sub . SUBSCRIBE = b'<STR_LIT>' <EOL> port = self . socket . bind_to_random_port ( interface ) <EOL> pub_port = self . pub . bind_to_random_port ( interface ) <EOL> self . url = interface + "<STR_LIT>" % port <EOL> self . pub_url = interface + "<STR_LIT>" % pub_port <EOL> self . location = socket . gethostbyname_ex ( socket . gethostname ( ) ) [ - <NUM_LIT:1> ] [ <NUM_LIT:0> ] <EOL> self . peers = { } <EOL> def __del__ ( self ) : <EOL> self . socket . close ( ) <EOL> self . pub . close ( ) <EOL> self . sub . close ( ) <EOL> self . _ctx . term ( ) <EOL> @ property <EOL> def info ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( self . identity , self . url , self . pub_url , self . location ) <EOL> def connect ( self , peers ) : <EOL> """<STR_LIT>""" <EOL> for peer , ( ident , url , pub_url , location ) in peers . items ( ) : <EOL> self . peers [ peer ] = ident <EOL> if ident != self . identity : <EOL> self . sub . connect ( disambiguate_url ( pub_url , location ) ) <EOL> if ident > self . identity : <EOL> self . socket . connect ( disambiguate_url ( url , location ) ) <EOL> def send ( self , peers , msg , flags = <NUM_LIT:0> , copy = True ) : <EOL> if not isinstance ( peers , list ) : <EOL> peers = [ peers ] <EOL> if not isinstance ( msg , list ) : <EOL> msg = [ msg ] <EOL> for p in peers : <EOL> ident = self . peers [ p ] <EOL> self . socket . send_multipart ( [ ident ] + msg , flags = flags , copy = copy ) <EOL> def recv ( self , flags = <NUM_LIT:0> , copy = True ) : <EOL> return self . socket . recv_multipart ( flags = flags , copy = copy ) [ <NUM_LIT:1> : ] <EOL> def publish ( self , msg , flags = <NUM_LIT:0> , copy = True ) : <EOL> if not isinstance ( msg , list ) : <EOL> msg = [ msg ] <EOL> self . pub . send_multipart ( msg , copy = copy ) <EOL> def consume ( self , flags = <NUM_LIT:0> , copy = True ) : <EOL> return self . sub . recv_multipart ( flags = flags , copy = copy ) </s>
