def __call__(self):
poller = zmq.Poller()<EOL>for socket in self.sockets:<EOL><INDENT>poller.register(socket, zmq.POLLIN)<EOL><DEDENT>group = self.greenlet_group<EOL>msgs = []<EOL>capture = msgs.extend<EOL>def accept(socket, call, args, kwargs, topics):<EOL><INDENT>group.spawn(self.work, socket, call, args, kwargs, topics)<EOL>group.join(<NUM_LIT:0>)<EOL><DEDENT>def reject(socket, call, topics):<EOL><INDENT>__, call_id, reply_to, __, __ = call<EOL>reply_socket, topics = self.replier(socket, topics, reply_to)<EOL>self.reject(reply_socket, call_id, topics)<EOL><DEDENT>def reject_if(call, topics):<EOL><INDENT>if self.reject_if(call, topics):<EOL><INDENT>return True<EOL><DEDENT>try:<EOL><INDENT>__, rpc_spec = self.find_call_target(call)<EOL><DEDENT>except KeyError:<EOL><INDENT>return True<EOL><DEDENT>return rpc_spec.reject_if.__get__(self.app)(call, topics)<EOL><DEDENT>try:<EOL><INDENT>while True:<EOL><INDENT>for socket, event in safe(poller.poll):<EOL><INDENT>assert event & zmq.POLLIN<EOL>del msgs[:]<EOL>try:<EOL><INDENT>header, payload, topics = recv(socket, capture=capture)<EOL>call = parse_call(header)<EOL>if group.full() or reject_if(call, topics):<EOL><INDENT>reject(socket, call, topics)<EOL>continue<EOL><DEDENT>args, kwargs = self.unpack(payload)<EOL><DEDENT>except:<EOL><INDENT>handle = self.malformed_message_handler<EOL>if handle is not None:<EOL><INDENT>exc_info = sys.exc_info()<EOL>handle(self, exc_info, msgs[:])<EOL><DEDENT>del handle<EOL>continue<EOL><DEDENT>accept(socket, call, args, kwargs, topics)<EOL><DEDENT>try:<EOL><INDENT>del header, payload, topics<EOL>del call<EOL>del args, kwargs<EOL><DEDENT>except UnboundLocalError:<EOL><INDENT>pass<EOL><DEDENT><DEDENT><DEDENT>finally:<EOL><INDENT>group.kill()<EOL><DEDENT>
Runs the worker.  While running, an RPC service is online.