Method socket_handler

dummyserver/testcase.py:94–118

(listener: socket.socket)

        quit_event = threading.Event()

        def socket_handler(listener: socket.socket) -> None:
            for _ in range(num):
                ready_event.set()

                listener.settimeout(LONG_TIMEOUT)
                while True:
                    if quit_event.is_set():
                        return
                    try:
                        sock = listener.accept()[0]
                        break
                    except TimeoutError:
                        continue
                consume_socket(sock, quit_event=quit_event)
                if quit_event.is_set():
                    sock.close()
                    return
                if block_send:
                    while not block_send.wait(LONG_TIMEOUT):
                        if quit_event.is_set():
                            sock.close()
                            return
                    block_send.clear()
                sock.send(response)
                sock.close()

        cls._start_server(socket_handler, quit_event=quit_event)
        return ready_event