test_server.py 文件源码

def start_listener(host='127.0.0.1', port=8642, shutdown_port=8643):
    server = socket.socket()
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(1)
    shutdown_server = socket.socket()
    shutdown_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    shutdown_server.bind((host, shutdown_port))
    shutdown_server.listen(1)

    done = False
    filenos = {s.fileno(): s for s in (server, shutdown_server)}

    while not done:
        r, _, _ = select.select(filenos.keys(), [], [], 0)

        for sock in [filenos[fd] for fd in r]:
            if sock is server:
                new_conn(server.accept()[0])
            else:
                done = True