Samreay commented on issue #184:
URL:
https://github.com/apache/pulsar-client-python/issues/184#issuecomment-1869953268
Can this be documented at the very least then? Having a hidden "do this and
your app will terminate" with no warning isn't ideal for an end user :)
Additionally, doco on `pulsar.Result` types would be great, I struggled to
handle this.
If you mean in terms of what should you see in the stack if the callback
raises, I'd expect the `send_async` to be the root cause with the callback
underneath it. If you mean in what execution context should a python async
callback raise exceptions, that would be the event loop in general.
The goal, IMHO, is to get the async producer send as close to pythons
standard library async as possible. Pythonically, this means you'd have
something like:
```python
try:
result = await producer.send_async(b"hello")
except:
catch issues here
```
For a more complete example, this is my current workaround, which uses janus
Queues (ie queues with sync and async interfaces) to convert the callback
function into actual python async:
```python
import asyncio
from collections.abc import Awaitable
import janus
import pulsar
async def better_send(producer: pulsar.Producer, data: bytes) ->
Awaitable[pulsar.MessageId]:
queue = janus.Queue()
def callback(result: pulsar.Result, message_id: pulsar.MessageId):
queue.sync_q.put((result, message_id))
producer.send_async(data, callback)
result, message_id = await queue.async_q.get()
if result == pulsar.Result.Timeout:
raise TimeoutError()
elif result != pulsar.Result.Ok:
raise ValueError(str(result))
return message_id
async def main():
client = pulsar.Client("pulsar://localhost:6650")
producer = client.create_producer("topic-example", send_timeout_millis=1)
await better_send(producer, b"hello")
if __name__ == "__main__":
asyncio.run(main())
```
Running this gives you a pythonic exception which doesn't crash the
application, and has proper asyncio support.
```bash
Traceback (most recent call last):
File "/home/sam/arenko/service-utils/tmp.py", line 30, in <module>
asyncio.run(main())
File "/home/sam/.pyenv/versions/3.11.4/lib/python3.11/asyncio/runners.py",
line 190, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/home/sam/.pyenv/versions/3.11.4/lib/python3.11/asyncio/runners.py",
line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File
"/home/sam/.pyenv/versions/3.11.4/lib/python3.11/asyncio/base_events.py", line
653, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^
File "/home/sam/arenko/service-utils/tmp.py", line 26, in main
await better_send(producer, b"hello")
File "/home/sam/arenko/service-utils/tmp.py", line 17, in better_send
raise TimeoutError()
TimeoutError
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
