greenio_test.py 文件源码

python
阅读 24 收藏 0 点赞 0 评论 0

项目:deb-python-eventlet 作者: openstack 项目源码 文件源码
def test_pipe(self):
        r, w = os.pipe()
        rf = greenio.GreenPipe(r, 'rb')
        wf = greenio.GreenPipe(w, 'wb', 0)

        def sender(f, content):
            for ch in map(six.int2byte, six.iterbytes(content)):
                eventlet.sleep(0.0001)
                f.write(ch)
            f.close()

        one_line = b"12345\n"
        eventlet.spawn(sender, wf, one_line * 5)
        for i in range(5):
            line = rf.readline()
            eventlet.sleep(0.01)
            self.assertEqual(line, one_line)
        self.assertEqual(rf.readline(), b'')
评论列表
文章目录


问题


面经


文章

微信
公众号

扫码关注公众号