pulsar updates...

[daveads]

Refactor PulsarBackend for multi-channel support and improved error handling

• Implement separate producers and consumers for each channel
• Add channel-specific subscribe and unsubscribe functionality
• Improve logging with a dedicated logger
• Enhance error handling and timeout management in next_published method
• Optimize connection and disconnection processes

[gemanor]

@daveads please address @danyi1212's comments

[daveads]

@daveads please address @danyi1212's comments

Sure

[gemanor]

Hey @daveads any news on this?

[daveads]

Hey @daveads any news on this?

@gemanor, I’ve been busy with OpenFGA integration and other tasks. I’ll send a commit tomorrow with the required changes.

[daveads]

@gemanor @danyi1212 Done

[gemanor]

@danyi1212 it's LGTM, please run a last sanity review and merge.

[daveads]

will work on the review this week...

[daveads]

we were suppose to drop test for 3.8 and add 3.12..

will get back to this... need to attend to something via opal...

@danyi1212

Done

[daveads]

Replaced anyio.to_thread.run_sync() with asyncio.to_thread()

[danyi1212]

Looks very good, left few small comments

[danyi1212]

Notice the difference between

await asyncio.gather(*coros, return_exceptions=True)  

Compared to

try:
    await asyncio.gather(*coros)
except Exception as e:
    logger.exception("Failed ...")

In the first option it will supress any exception raised inside the coros and include them in the list returned by the gather, retaining the order of the received coros.

In the second example, when a single coro inside the gather fails, it will immediately cancel all running coros, and drop the returned values of completed coros.

If you want all coros to complete regardless of when some of them fail, you should use return_exceptions=True. Through you should remember to actually use the returned values, and if exception where raised, handle them (in our case, at least log them)