test_aiozmq_client.py

import asyncio
from aiozmq import rpc


class EventHandler(rpc.AttrHandler):

    @rpc.method
    def remote_func(self, msg):
        print('EventHandler: {}'.format(msg))


async def run_client():

    print('start rpc client...')
    client = await rpc.connect_rpc(connect='tcp://127.0.0.1:5503')

    ret = await client.call.myfunc1(1, 2)
    print('myfunc1 ret: {}'.format(ret))

    ret = await client.call.myfunc2(1, 2)
    print('myfunc2 ret: {}'.format(ret))

    subscribe_port = await client.call.register_event('topic')
    subscriber = await rpc.serve_pubsub(EventHandler(), subscribe='topic', bind='tcp://127.0.0.1:{}'.format(subscribe_port))
    subscriber_addr = list(subscriber.transport.bindings())[0]
    print('subscriber_addr: {}'.format(subscriber_addr))

    await asyncio.sleep(10)


asyncio.get_event_loop().run_until_complete(run_client())

test_aiozmq_server.py

import asyncio
from aiozmq import rpc
from itertools import count


class MyRPCHandler(rpc.AttrHandler):

    @rpc.method
    def myfunc1(self, arg1, arg2):
        print('myfunc1: arg1 = {}, arg2 = {}'.format(arg1, arg2))
        return arg1 + arg2

    @rpc.method
    def myfunc2(self, arg1, arg2):
        print('myfunc2: arg1 = {}, arg2 = {}'.format(arg1, arg2))
        return arg1 * arg2

    @rpc.method
    def register_event(self, topic):
        print('register_event: topic = {}'.format(topic))
        return 10501


async def run_server():
    print('start rpc server...')
    rpc_server = await rpc.serve_rpc(MyRPCHandler(), bind='tcp://127.0.0.1:5503')

    # setup the publisher for sending event message
    publisher = await rpc.connect_pubsub(connect='tcp://127.0.0.1:10501')

    for i in range(10):
        await publisher.publish('topic').remote_func('test123: {}'.format(i))
        await asyncio.sleep(1)

    await rpc_server.wait_closed()
    print('rpc server close')

asyncio.get_event_loop().run_until_complete(run_server())