queue_test.py 文件源码

def test_zero_max_size(self):
        q = eventlet.Queue(0)

        def sender(evt, q):
            q.put('hi')
            evt.send('done')

        def receiver(q):
            x = q.get()
            return x

        evt = event.Event()
        gt = eventlet.spawn(sender, evt, q)
        eventlet.sleep(0)
        assert not evt.ready()
        gt2 = eventlet.spawn(receiver, q)
        self.assertEqual(gt2.wait(), 'hi')
        self.assertEqual(evt.wait(), 'done')
        gt.wait()