greenio_test.py 文件源码

def test_send_timeout(self):
        self.reset_timeout(2)
        listener = bufsized(eventlet.listen(('', 0)))

        evt = event.Event()

        def server():
            # accept the connection in another greenlet
            sock, addr = listener.accept()
            sock = bufsized(sock)
            evt.wait()

        gt = eventlet.spawn(server)

        addr = listener.getsockname()

        client = bufsized(greenio.GreenSocket(socket.socket()))
        client.connect(addr)

        client.settimeout(0.00001)
        msg = b"A" * 100000  # large enough number to overwhelm most buffers

        # want to exceed the size of the OS buffer so it'll block in a
        # single send
        def send():
            for x in range(10):
                client.send(msg)

        expect_socket_timeout(send)

        evt.send()
        gt.wait()