I was digging through the zmq docs for a friend and started wondering how does one begin to approach the problem of subscribing to push data to some other actor? I.e. what is effectively the way to accomplish reverse pub-sub or represent push-pull zmq sockets using async generator semantics? The reactive x paradigm supports this using...

One idea I had was to always use a pull approach such that a client actor can tell another to async for from it - a kind of call-back and iterate my async generator why don't you.

Something maybe like:

async def push_data():
    for i in range(10**6):
        yield i
        await trio.sleep(0.3)

async with tractor.connect('ml_core') as portal:
    async with portal.push(push_data, 'remote_func_name', mod=6) as portal:
        # other app code that can run alongside push task
    # at close of block push stream is cancelled

And then on the server side the remote_func_name must be defined:

async def remote_func_name(pushed, mod):
    async for item in pushed:
        # stash for downstream processing
        if item % mod == 0:
            await tractor.current_actor().statespace['queue'].put(item)