test_asyncore.py 文件源码

def test_quick_connect(self):
        # see: http://bugs.python.org/issue10340
        if self.family in (socket.AF_INET, getattr(socket, "AF_INET6", object())):
            server = BaseServer(self.family, self.addr)
            t = threading.Thread(target=lambda: asyncore.loop(timeout=0.1,
                                                              count=500))
            t.start()
            self.addCleanup(t.join)

            s = socket.socket(self.family, socket.SOCK_STREAM)
            s.settimeout(.2)
            s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
                         struct.pack('ii', 1, 0))
            try:
                s.connect(server.address)
            except socket.error:
                pass
            finally:
                s.close()