test_asyncore.py 文件源码

def capture_server(evt, buf, serv):
    try:
        serv.listen(5)
        conn, addr = serv.accept()
    except socket.timeout:
        pass
    else:
        n = 200
        start = time.time()
        while n > 0 and time.time() - start < 3.0:
            r, w, e = select.select([conn], [], [], 0.1)
            if r:
                n -= 1
                data = conn.recv(10)
                # keep everything except for the newline terminator
                buf.write(data.replace(b'\n', b''))
                if b'\n' in data:
                    break
            time.sleep(0.01)

        conn.close()
    finally:
        serv.close()
        evt.set()