_test_asyncio.py 文件源码

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

项目:zanph 作者: zanph 项目源码 文件源码
def test_recv(self):
        @asyncio.coroutine
        def test():
            a, b = self.create_bound_pair(zmq.PUSH, zmq.PULL)
            f1 = b.recv()
            f2 = b.recv()
            assert not f1.done()
            assert not f2.done()
            yield from  a.send_multipart([b'hi', b'there'])
            recvd = yield from f2
            assert f1.done()
            self.assertEqual(f1.result(), b'hi')
            self.assertEqual(recvd, b'there')
        self.loop.run_until_complete(test())
评论列表
文章目录


问题


面经


文章

微信
公众号

扫码关注公众号