test_agent_serialization.py 文件源码

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

项目:osbrain 作者: opensistemas-hub 项目源码 文件源码
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()
评论列表
文章目录


问题


面经


文章

微信
公众号

扫码关注公众号