<s> import os <EOL> import random <EOL> import struct <EOL> import errno <EOL> import time <EOL> import pickle <EOL> from functools import partial <EOL> import zmq <EOL> from zmq import * <EOL> from . import core , channel <EOL> from . core import gethub <EOL> try : <EOL> REP_SOCKET = zmq . XREP <EOL> except AttributeError : <EOL> REP_SOCKET = zmq . ROUTER <EOL> try : <EOL> REQ_SOCKET = zmq . XREQ <EOL> except AttributeError : <EOL> REQ_SOCKET = zmq . DEALER <EOL> DEFAULT_IO_THREADS = <NUM_LIT:1> <EOL> class Socket ( zmq . Socket ) : <EOL> def dict_configure ( self , data ) : <EOL> if '<STR_LIT>' in data : <EOL> if isinstance ( data [ '<STR_LIT>' ] , ( tuple , list ) ) : <EOL> for addr in data [ '<STR_LIT>' ] : <EOL> self . bind ( addr ) <EOL> else : <EOL> self . bind ( data [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in data : <EOL> if isinstance ( data [ '<STR_LIT>' ] , ( tuple , list ) ) : <EOL> for addr in data [ '<STR_LIT>' ] : <EOL> self . connect ( addr ) <EOL> else : <EOL> self . connect ( data [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in data : <EOL> self . setsockopt ( zmq . HWM , int ( data [ '<STR_LIT>' ] ) ) <EOL> if '<STR_LIT>' in data : <EOL> self . setsockopt ( zmq . AFFINITY , int ( data [ '<STR_LIT>' ] ) ) <EOL> if '<STR_LIT>' in data : <EOL> self . setsockopt ( zmq . BACKLOG , int ( data [ '<STR_LIT>' ] ) ) <EOL> if '<STR_LIT>' in data : <EOL> self . setsockopt ( zmq . LINGER , int ( data [ '<STR_LIT>' ] ) ) <EOL> if '<STR_LIT>' in data : <EOL> self . setsockopt ( zmq . SNDBUF , int ( data [ '<STR_LIT>' ] ) ) <EOL> if '<STR_LIT>' in data : <EOL> self . setsockopt ( zmq . RCVBUF , int ( data [ '<STR_LIT>' ] ) ) <EOL> def send_data ( sock , data , address = None ) : <EOL> if address is None : <EOL> _rep = [ ] <EOL> else : <EOL> _rep = list ( address ) <EOL> for a in data : <EOL> if isinstance ( a , str ) : <EOL> _rep . append ( a . encode ( '<STR_LIT:utf-8>' ) ) <EOL> else : <EOL> _rep . append ( a ) <EOL> while True : <EOL> try : <EOL> sock . send_multipart ( _rep , zmq . NOBLOCK ) <EOL> except zmq . ZMQError as e : <EOL> if e . errno == errno . EAGAIN : <EOL> gethub ( ) . do_write ( sock ) <EOL> continue <EOL> elif e . errno == errno . EINTR : <EOL> continue <EOL> else : <EOL> raise <EOL> else : <EOL> break <EOL> def rep_responder ( sock , address , callback , data ) : <EOL> hub = core . gethub ( ) <EOL> reply = callback ( * data ) <EOL> if isinstance ( reply , bytes ) : <EOL> send_data ( sock , ( reply , ) , address = address ) <EOL> elif isinstance ( reply , str ) : <EOL> send_data ( sock , ( reply . encode ( '<STR_LIT:utf-8>' ) , ) , address = address ) <EOL> elif reply is None : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> else : <EOL> send_data ( sock , reply , address = address ) <EOL> def rep_listener ( sock , callback ) : <EOL> hub = core . gethub ( ) <EOL> wait_slot = getattr ( callback , '<STR_LIT>' , None ) <EOL> while True : <EOL> hub . do_read ( sock ) <EOL> while True : <EOL> if wait_slot is not None : <EOL> wait_slot ( ) <EOL> try : <EOL> data = sock . recv_multipart ( zmq . NOBLOCK ) <EOL> except zmq . ZMQError as e : <EOL> if e . errno == errno . EAGAIN or e . errno == errno . EINTR : <EOL> break <EOL> else : <EOL> raise <EOL> i = data . index ( b'<STR_LIT>' ) <EOL> addr = data [ : i + <NUM_LIT:1> ] <EOL> data = data [ i + <NUM_LIT:1> : ] <EOL> hub . do_spawnswitch ( <EOL> partial ( rep_responder , sock , addr , callback , data ) ) <EOL> def rep_socket ( callback ) : <EOL> sock = Socket ( context ( ) , REP_SOCKET ) <EOL> core . gethub ( ) . do_spawnservice ( partial ( rep_listener , sock , callback ) ) <EOL> return sock <EOL> def sub_listener ( sock , callback ) : <EOL> hub = core . gethub ( ) <EOL> wait_slot = getattr ( callback , '<STR_LIT>' , None ) <EOL> while True : <EOL> hub . do_read ( sock ) <EOL> while True : <EOL> if wait_slot is not None : <EOL> wait_slot ( ) <EOL> try : <EOL> data = sock . recv_multipart ( zmq . NOBLOCK ) <EOL> except zmq . ZMQError as e : <EOL> if e . errno == errno . EAGAIN or e . errno == errno . EINTR : <EOL> break <EOL> else : <EOL> raise <EOL> hub . do_spawnswitch ( partial ( callback , * data ) ) <EOL> def sub_socket ( callback ) : <EOL> sock = Socket ( context ( ) , zmq . SUB ) <EOL> core . gethub ( ) . do_spawnservice ( partial ( sub_listener , sock , callback ) ) <EOL> return sock <EOL> def pub_socket ( ) : <EOL> return PubChannel ( ) <EOL> def pull_socket ( callback ) : <EOL> sock = Socket ( context ( ) , zmq . PULL ) <EOL> core . gethub ( ) . do_spawnservice ( partial ( sub_listener , sock , callback ) ) <EOL> return sock <EOL> def push_socket ( ) : <EOL> return PushChannel ( ) <EOL> def req_socket ( ) : <EOL> return ReqChannel ( ) <EOL> class ReqChannel ( channel . MuxReqChannel ) : <EOL> def __init__ ( self ) : <EOL> super ( ) . __init__ ( ) <EOL> self . init_id ( ) <EOL> self . _sock = Socket ( context ( ) , REQ_SOCKET ) <EOL> self . _start ( ) <EOL> def init_id ( self ) : <EOL> self . prefix = struct . pack ( '<STR_LIT>' , <EOL> os . getpid ( ) % <NUM_LIT> , random . randrange ( <NUM_LIT> ) , int ( time . time ( ) ) ) <EOL> self . counter = <NUM_LIT:0> <EOL> def new_id ( self ) : <EOL> self . counter += <NUM_LIT:1> <EOL> if self . counter >= <NUM_LIT> : <EOL> self . init_id ( ) <EOL> return self . prefix + struct . pack ( '<STR_LIT:L>' , self . counter ) <EOL> def bind ( self , value ) : <EOL> self . _sock . bind ( value ) <EOL> def connect ( self , value ) : <EOL> self . _sock . connect ( value ) <EOL> def dict_configure ( self , dic ) : <EOL> self . _sock . dict_configure ( dic ) <EOL> def sender ( self ) : <EOL> wait_write = core . gethub ( ) . do_write <EOL> while True : <EOL> wait_write ( self . _sock ) <EOL> id , data = self . peek_request ( ) <EOL> self . _sock . send ( id , zmq . SNDMORE ) <EOL> self . _sock . send ( b"<STR_LIT>" , zmq . SNDMORE ) <EOL> self . _sock . send_multipart ( data ) <EOL> self . pop_request ( ) <EOL> def receiver ( self ) : <EOL> wait_read = core . gethub ( ) . do_read <EOL> while True : <EOL> wait_read ( self . _sock ) <EOL> data = self . _sock . recv_multipart ( ) <EOL> assert data [ <NUM_LIT:1> ] == b'<STR_LIT>' <EOL> self . produce ( data [ <NUM_LIT:0> ] , data [ <NUM_LIT:2> : ] ) <EOL> class OutputChannel ( object ) : <EOL> zmq_kind = None <EOL> def __init__ ( self ) : <EOL> super ( ) . __init__ ( ) <EOL> self . _sock = Socket ( context ( ) , self . zmq_kind ) <EOL> def bind ( self , value ) : <EOL> self . _sock . bind ( value ) <EOL> def connect ( self , value ) : <EOL> self . _sock . connect ( value ) <EOL> def dict_configure ( self , dic ) : <EOL> self . _sock . dict_configure ( dic ) <EOL> class PubChannel ( OutputChannel ) : <EOL> zmq_kind = zmq . PUB <EOL> def publish ( self , * args ) : <EOL> send_data ( self . _sock , args ) <EOL> class PushChannel ( OutputChannel ) : <EOL> zmq_kind = zmq . PUSH <EOL> def push ( self , * args ) : <EOL> send_data ( self . _sock , args ) <EOL> def _get_fd ( value ) : <EOL> if isinstance ( value , zmq . Socket ) : <EOL> return value <EOL> else : <EOL> return value . fileno ( ) <EOL> def plug ( hub , io_threads = DEFAULT_IO_THREADS ) : <EOL> assert not hasattr ( hub , '<STR_LIT>' ) <EOL> ctx = hub . zmq_context = zmq . Context ( io_threads ) <EOL> hub . change_poller ( zmq . Poller , filedes = _get_fd , <EOL> POLLIN = zmq . POLLIN , POLLOUT = zmq . POLLOUT , POLLERR = zmq . POLLERR , POLLHUP = <NUM_LIT:0> ) <EOL> core . os_errors += ( ZMQError , ) <EOL> hub . log_plugged ( ctx , name = '<STR_LIT>' ) <EOL> return ctx <EOL> def context ( ) : <EOL> hub = core . gethub ( ) <EOL> ctx = getattr ( hub , '<STR_LIT>' , None ) <EOL> if ctx is None : <EOL> return plug ( hub ) <EOL> return ctx <EOL> class MethodCallError ( Exception ) : <EOL> pass <EOL> class MethodException ( Exception ) : <EOL> pass <EOL> class Method ( object ) : <EOL> __slots__ = ( '<STR_LIT>' , '<STR_LIT:name>' ) <EOL> def __init__ ( self , owner , name ) : <EOL> self . owner = owner <EOL> self . name = name <EOL> def __call__ ( self , * args ) : <EOL> o = self . owner <EOL> lst = [ self . name ] <EOL> lst . extend ( map ( o . dumps , args ) ) <EOL> res = o . channel . request ( lst ) . get ( ) <EOL> if res [ <NUM_LIT:0> ] == b'<STR_LIT>' : <EOL> return self . owner . loads ( res [ <NUM_LIT:1> ] ) <EOL> elif res [ <NUM_LIT:0> ] == b'<STR_LIT>' : <EOL> raise MethodCallError ( res [ <NUM_LIT:1> ] . decode ( '<STR_LIT:ascii>' ) ) <EOL> elif res [ <NUM_LIT:0> ] == b'<STR_LIT>' : <EOL> raise MethodException ( res [ <NUM_LIT:1> ] . decode ( '<STR_LIT:utf-8>' ) ) <EOL> else : <EOL> raise MethodException ( "<STR_LIT>" ) <EOL> class Requester ( object ) : <EOL> loads = pickle . loads <EOL> dumps = pickle . dumps <EOL> def __init__ ( self , channel , prefix = '<STR_LIT>' ) : <EOL> self . channel = channel <EOL> if isinstance ( prefix , str ) : <EOL> self . prefix = prefix . encode ( '<STR_LIT:ascii>' ) <EOL> def __getattr__ ( self , name ) : <EOL> if name . startswith ( '<STR_LIT:_>' ) : <EOL> raise AttributeError ( "<STR_LIT>" ) <EOL> fullname = self . prefix + name . encode ( '<STR_LIT:ascii>' ) <EOL> return Method ( self , fullname ) <EOL> class Responder ( object ) : <EOL> loads = pickle . loads <EOL> dumps = pickle . dumps <EOL> def __call__ ( self , name , * args ) : <EOL> name = name . decode ( '<STR_LIT:ascii>' ) <EOL> if name . startswith ( '<STR_LIT:_>' ) : <EOL> return ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> meth = getattr ( self , name , None ) <EOL> if meth is None or not callable ( meth ) : <EOL> return ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> try : <EOL> args = tuple ( map ( self . loads , args ) ) <EOL> except Exception as e : <EOL> return ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> try : <EOL> result = meth ( * args ) <EOL> except Exception as e : <EOL> return ( b'<STR_LIT>' , repr ( e ) ) <EOL> try : <EOL> result = self . dumps ( result ) <EOL> except Exception as e : <EOL> return ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> return ( b'<STR_LIT>' , result ) <EOL> class Dispatcher ( dict ) : <EOL> def __init__ ( self , default , ** kw ) : <EOL> if isinstance ( default , dict ) : <EOL> super ( ) . __init__ ( default , ** kw ) <EOL> else : <EOL> self [ None ] = default <EOL> super ( ) . __init__ ( ** kw ) <EOL> def __call__ ( self , name , * args ) : <EOL> parts = name . rsplit ( b'<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> if len ( parts ) > <NUM_LIT:1> : <EOL> oname , name = parts <EOL> oname = oname . decode ( '<STR_LIT:ascii>' ) <EOL> else : <EOL> oname = None <EOL> name = parts [ <NUM_LIT:0> ] <EOL> obj = self . get ( oname ) <EOL> if obj is None : <EOL> return ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> else : <EOL> return obj ( name , * args ) </s>
