process.py 文件源码

def __init__(self, host, address, log_address):

        object.__init__(self)

        if log_address is None:
            raise NotImplementedError()
            # TODO remove
        self.logger = get_log(log_address, name=__name__)

        # TODO find proper space to define following class
        class Encoder(json.JSONEncoder):

            def default(self_, obj):
                if obj is None:
                    obj = json.JSONEncoder.default(obj)
                else:
                    if isinstance(obj, Proxy):
                        obj = obj.encode()
                    else:
                        obj = self.wrap_proxy(obj)
                        obj = obj.encode()
                return obj

        self.encoder = Encoder

        self.context = zmq.Context()
        # TODO connect tmp socket
        self.logger.debug("connect tmp socket at {a}".format(a=address))
        socket = self.context.socket(zmq.PAIR)
        socket.connect(address)
        # TODO bind rpc socket
        transport = 'tcp'
        port = '*'
        endpoint = '{h}:{p}'.format(h=host, p=port)
        address = '{t}://{e}'.format(t=transport, e=endpoint)
        self.logger.debug("bind rpc socket at {a}".format(a=address))
        self.socket = self.context.socket(zmq.PAIR)
        # self.socket.setsockopt(zmq.RCVTIMEO, 10000)
        self.socket.bind(address)
        self.address = self.socket.getsockopt(zmq.LAST_ENDPOINT)
        self.logger.debug("rpc socket binded at {a}".format(a=self.address))
        # TODO send rpc address
        self.logger.debug("send back rpc address")
        message = {
            'address': self.address,
        }
        socket.send_json(message)

        self.last_obj_id = -1
        self.objs = {}
        self.poller = zmq.Poller()
        self.poller.register(self.socket, zmq.POLLIN)