def test_sockopts(self):
"""setting socket options with ctx attributes"""
ctx = self.Context()
ctx.linger = 5
self.assertEqual(ctx.linger, 5)
s = ctx.socket(zmq.REQ)
self.assertEqual(s.linger, 5)
self.assertEqual(s.getsockopt(zmq.LINGER), 5)
s.close()
# check that subscribe doesn't get set on sockets that don't subscribe:
ctx.subscribe = b''
s = ctx.socket(zmq.REQ)
s.close()
ctx.term()
评论列表
文章目录