test_logging.py 文件源码

def __init__(self, addr, handler, poll_interval, sockmap):
            self._localaddr = addr
            self._remoteaddr = None
            self.data_size_limit = None
            self.sockmap = sockmap
            asyncore.dispatcher.__init__(self, map=sockmap)
            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.setblocking(0)
                self.set_socket(sock, map=sockmap)
                # try to re-use a server port if possible
                self.set_reuse_addr()
                self.bind(addr)
                self.port = sock.getsockname()[1]
                self.listen(5)
            except:
                self.close()
                raise
            self._handler = handler
            self._thread = None
            self.poll_interval = poll_interval