Callback function is not asynchronous

[litnimax]

I have the following simple test script:

import asyncio
import nats
from aiorun import run

class Client:
    async def nats_loop(self):
        self.nc = await nats.connect()
        await self.nc.subscribe('test', cb=self.sleep)
        while True:
            await asyncio.sleep(1)

    async def sleep(self, msg):
        print('Test start')
        await asyncio.sleep(5)
        print('Test end')

async def main():
    client = Client()
    asyncio.gather(client.nats_loop())

if __name__ == '__main__':    
    run(main())

I expect that when I publish 3 messages one by one the output will be:

Test start
Test start
Test start
Test end
Test end
Test end

But the actual output is:

Test start
Test end
Test start
Test end
Test start
Test end

That means the sleep callback is called in a sync way not asyc.

I can get the desired result if I modify the script as follows:

import asyncio
import nats
from aiorun import run

class Client:
    async def nats_loop(self):
        self.nc = await nats.connect()
        await self.nc.subscribe('test', cb=self.test)
        while True:
            await asyncio.sleep(1)

    async def test(self, msg):
        asyncio.ensure_future(self.sleep(msg))

    async def sleep(self, msg):
        print('Test start')
        await asyncio.sleep(5)
        print('Test end')

async def main():
    client = Client()
    asyncio.gather(client.nats_loop())

if __name__ == '__main__':    
    run(main())

So now I get

Test start
Test start
Test start
Test end
Test end
Test end

The question is if it's made by design or it's a bug?

Thanks.

[stankudrow]

@wallyqs , @caspervonb , could you give an answer, please?

[wallyqs]

yes each callback is executed serially as in the other nats clients, so in order to process a message without causing head of line blocking it is necessary to handle manually for example doing asyncio.ensure_future(self.sleep(msg)) without waiting for the completion of the task

[stankudrow]

yes each callback is executed serially as in the other nats clients, so in order to process a message without causing head of line blocking it is necessary to handle manually for example doing asyncio.ensure_future(self.sleep(msg)) without waiting for the completion of the task

So this is not a bug and it is by design, right? No measures must be taken to fix this situation, correct?

[wallyqs]

right this would be working as designed.