def test_pushpull_raw_zmq_outside(nsproxy):
"""
Simple push-pull pattern test. Channel without serialization.
The message is sent from outside osBrain, through a ZMQ PUSH socket.
"""
# Create an osBrain agent that will receive the message
a1 = run_agent('a1')
a1.set_attr(received=None)
addr = a1.bind('PULL', transport='tcp', handler=set_received,
serializer='raw')
# Create a raw ZeroMQ PUSH socket
context = zmq.Context()
socket = context.socket(zmq.PUSH)
socket.connect('tcp://%s:%s' % (addr.address.host, addr.address.port))
# Send the message
message = b'Hello world'
socket.send(message)
assert wait_agent_attr(a1, name='received', value=message)
socket.close()
context.destroy()
test_agent_serialization.py 文件源码
python
阅读 21
收藏 0
点赞 0
评论 0
评论列表
文章目录