azure-storage queue async error

[hholst80]

Python 3.9.7
Fedora 34 (Workstation Edition)
azure-core          1.16.0
azure-storage-blob  12.8.1
azure-storage-queue 12.1.6

I was expecting the async code below to output the exact same result as the one below when executed against an empty storage queue.

The serial code executes just fine and emits, nothing, as expected on an empty queue.
The async code throws this at me:

Fatal error on SSL transport
protocol: <asyncio.sslproto.SSLProtocol object at 0x7f7fe8b4f4f0>
transport: <_SelectorSocketTransport closing fd=6>
Traceback (most recent call last):
  File "/usr/lib64/python3.9/asyncio/selector_events.py", line 918, in write
    n = self._sock.send(data)
OSError: [Errno 9] Bad file descriptor

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib64/python3.9/asyncio/sslproto.py", line 684, in _process_write_backlog
    self._transport.write(chunk)
  File "/usr/lib64/python3.9/asyncio/selector_events.py", line 924, in write
    self._fatal_error(exc, 'Fatal write error on socket transport')
  File "/usr/lib64/python3.9/asyncio/selector_events.py", line 719, in _fatal_error
    self._force_close(exc)
  File "/usr/lib64/python3.9/asyncio/selector_events.py", line 731, in _force_close
    self._loop.call_soon(self._call_connection_lost, exc)
  File "/usr/lib64/python3.9/asyncio/base_events.py", line 746, in call_soon
    self._check_closed()
  File "/usr/lib64/python3.9/asyncio/base_events.py", line 510, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7f7fe7dda5e0>
Unclosed connector
connections: ['[(<aiohttp.client_proto.ResponseHandler object at 0x7f7fe7dc8b80>, 19043.264955058)]']
connector: <aiohttp.connector.TCPConnector object at 0x7f7fe7dda5b0>

import asyncio
import os, uuid

async def amain():
    from azure.storage.queue.aio import QueueClient
    connect_str = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
    queue_name = os.getenv("QUEUE_NAME", "events")
    queue_client = QueueClient.from_connection_string(connect_str, queue_name)
    messages = queue_client.receive_messages()
    async for msg in messages:
        print(msg.content)
        await queue_client.delete_message(msg)

def main():
    from azure.storage.queue import QueueClient
    connect_str = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
    queue_name = os.getenv("QUEUE_NAME", "events")
    queue_client = QueueClient.from_connection_string(connect_str, queue_name)
    messages = queue_client.receive_messages()
    for msg in messages:
        print(msg.content)
        queue_client.delete_message(msg)

#asyncio.run(amain())

main()

[kristapratico]

@hholst80 thanks for the feedback and the repro instructions, @xiafu-msft please take a look.

[HarshitaSingh-MSFT]

@hholst80, I was able to reproduce your scenario and it looks like you might have missed out a small code piece in your asynchronous method. Please find the working code below and try running it instead:

import asyncio
import os, uuid

async def amain():

    from azure.storage.queue.aio import QueueClient
    connect_str = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
    queue_name = os.getenv("QUEUE_NAME", "events")
    queue_client = QueueClient.from_connection_string(connect_str, queue_name)

    async with queue_client:
        messages = queue_client.receive_messages()
        async for msg in messages:
            print(msg.content)
            await queue_client.delete_message(msg)

asyncio.run(amain())

Let me know if this works for you.

[hholst80]

I will try it. I might have missed it but I cannot recall seeing this construct in the documentation. I will try and find the exact link.

[HarshitaSingh-MSFT]

@hholst80 , here is the sample code's link.

[hholst80]

@HarshitaSingh-MSFT works perfect thank you!

I based my code on this snippet here:

https://docs.microsoft.com/en-us/python/api/overview/azure/storage-queue-readme?view=azure-python

Search for Receive and process messages asynchronously

I think this can be closed, maybe a small doc update to make it clear that the queue_client needs to be wrapped as above async with ...? Or is there some implicit context here that I am not thinking of?

[HarshitaSingh-MSFT]

@hholst80, I created a PR and it is being reviewed. You can expect to see the changes soon.

[HarshitaSingh-MSFT]

@hholst80, we will now close this issue, since your query is answered and corresponding action is taken. If you have further questions, please feel free to reach out over this thread by tagging me.