Method build_send_waiter

Lib/test/test__interpchannels.py:834–855

(self, obj, *, buffer=False)

    # send with waiting

    def build_send_waiter(self, obj, *, buffer=False):
        # We want a long enough sleep that send() actually has to wait.

        if buffer:
            send = _channels.send_buffer
        else:
            send = _channels.send

        cid = _channels.create(REPLACE)
        try:
            started = time.monotonic()
            send(cid, obj, blocking=False)
            stopped = time.monotonic()
            recv_nowait(cid)
        finally:
            _channels.destroy(cid)
        delay = stopped - started  # seconds
        delay *= 3

        def wait():
            time.sleep(delay)
        return wait

    def test_send_blocking_waiting(self):
        received = None