Session push hangs on reconnect

[oleynikandrey]

After asynctnt reconnected to Tarantool, PushIterator is blocked forever on waiting asyncio.Event.

The only solution I've come up with is constantly check id of connection._transport and once it's changed recreate PushIterator, eg:

init.lua

box.cfg{listen=3301}
push = require('push')

push.lua

local fiber = require('fiber')

local x = 0

local listen = function()
    while true do
        fiber.sleep(2)
        x = x + 1
        box.session.push(x)
    end
end

return {
    listen = listen
}

asynctnt client

import asyncio
import asynctnt


def get_push_iterator(connection):
    fut = connection.call("push.listen", push_subscribe=True)
    return asynctnt.PushIterator(fut)


async def main():
    async with asynctnt.Connection(port=3301) as conn:

        it = get_push_iterator(conn)
        transport_id = id(conn._transport)

        while True:
            if (current_transport_id := id(conn._transport)) != transport_id:
                transport_id = current_transport_id
                it = get_push_iterator(conn)

            try:
                result = await asyncio.wait_for(it.__anext__(), timeout=10)
            except asyncio.exceptions.TimeoutError:
                pass


if __name__ == "__main__":
    asyncio.run(main())

Any thoughts on this?

[igorcoding]

Hi!
I understand the issue, I think I got the fix in the branch infinite_push_bug. So it'll be great if you would try it for yourself (just install from the branch).
And btw I think I want to maybe rework the iterator thing. So, If you have any suggestions - I will be happy to listen.

[igorcoding]

Forgot to mention: I decided to throw an exception if this situation occurs, hope that is ok, because I think it's the most viable thing.

[oleynikandrey]

I've tested and it's work as expected. With an error thrown on connection lost I can handle the error and resubscribe to pushes.

What about moving PushIterator logic to Response class and creating wrapper in Connection which will return async iterator? It will looks like that:

async with asynctnt.Connection(port=3301) as conn:
    async for msg in conn.subscribe("infinite_push_loop"):
        print(msg)

So that there is no need to create iterator from a result of call method.

[igorcoding]

I'll release soon v1.2 with the fix.
About your suggestions. I like the idea to merge Response object with PusIterator functionality but it requires some refactoring which is a good thing. But making a .subscribe() method is incorrect because in theory any tarantool request may yield stuff back to the client (right now there are call and eval and maybe in future it will be select and others), so it is impractical to make this method.
I guess it'll be fine if every method would return not a plain Future object, which results into a Response object, but a Response itself, which you can await on and also iterate using __aiter__ and __anext__.