fix: protocol publisher fails to report error

[eliteprox]

This pull request introduces significant improvements to error handling, background task management, and stream lifecycle events in the pytrickle library and its example usage. The changes focus on making error callbacks consistently asynchronous, improving background task cleanup, and ensuring that stream stop events are handled gracefully. Additionally, the protocol and client classes now propagate errors and shutdown events more reliably, and global exception handling for asyncio is improved to suppress expected errors during shutdown.

Error handling and callback improvements:

• All error callbacks are now required to be asynchronous functions, simplifying error propagation and handling throughout the codebase. (pytrickle/__init__.py [1] pytrickle/client.py [2] [3] [4] [5]
• The protocol and client now propagate protocol shutdown and subscription end events to the client via the error callback, and the client distinguishes between clean shutdown and error conditions. (pytrickle/client.py [1] [2] [3]

Background task management and cleanup:

• Added unified tracking and cleanup of background tasks in TrickleComponent, with automatic removal and exception handling on task completion. (pytrickle/base.py [1] [2]
• The example video processor now starts background tasks only when the event loop is running, and cleans them up when the stream stops, using the new on_stream_stop callback. (examples/process_video_example.py [1] [2] [3]

Asyncio and shutdown robustness:

• Introduced a global asyncio exception handler to suppress expected aiohttp connection reset errors during shutdown, reducing noise in logs. (pytrickle/base.py [1] pytrickle/protocol.py [2] [3]
• Improved the data sending loop in the client to respond to both stop and error events, ensuring reliable shutdown and error handling. (pytrickle/client.py [1] [2]

Protocol task error handling:

• Protocol subscribe and publish tasks are now wrapped with a generic error-handling wrapper, ensuring that any exceptions are logged and propagated via error callbacks. (pytrickle/protocol.py [1] [2] [3]

These changes make the library more robust, easier to debug, and safer to use in production environments by improving error visibility and resource cleanup.

[ad-astra-video]

frame_processor does not have on_stream_stop right?

[eliteprox]

Correct, it's not an abstract method of frame_processor. It's appended by StreamProcessor on initialization like other callbacks registered to StreamProcessor.

StreamProcessor accepts an on_stream_stop callback as a parameter, similar to param_updater. _InternalFrameProcessor extends the abstract FrameProcessor class and stores the on_stream_stop callback as an attribute

[ad-astra-video]

If we want to include it I think we should add it to FrameProcessor similar to the error_callback. It seems strange to call something from TrickleClient that is setup in a higher level abstraction.

Let me know if I am missing something here tho.

[eliteprox]

In this case, on_stream_stop is more of an event triggered by the client, rather than a method called by the FrameProcessor. I added it as an abstract class now so it is available to implement in frame processors f347e41

Client still needs to call on_stream_stop at this point in the protocol shutdown sequence. There is no other coordination with FrameProcessors currently afaik.

pytrickle/pytrickle/client.py

Lines 117 to 124 in f347e41

	if self.frame_processor.on_stream_stop:
	try:
	await self.frame_processor.on_stream_stop()
	logger.info("Stream stop callback executed successfully")
	except Exception as e:
	logger.error(f"Error in stream stop callback: {e}")
	await self.protocol.stop()

[eliteprox]

It seems strange to call something from TrickleClient that is setup in a higher level abstraction.

Yeah, it sounds off because TrickleClient was originally intended as a class for interacting with trickle protocol directly in a multimedia context. In a consumer context it's an internal component so the term client is a bit misleading.

I think once we reorganize classes into package namespaces this will be more clear

[ad-astra-video]

WDYT about just waiting for stop_event here?

Will error_event cause channels to close down before this could execute one last time?

[eliteprox]

Good catch!

[eliteprox]

fixed in 2703cff

[ad-astra-video]

I moved the waiting for stop/error to a function. Can you test to confirm still same behavior?

[eliteprox]

I tested and it works great, no issues found in stopping the publisher. Ran a high-rate publisher test and got ~120 msg/s which was expected. One message did fail, but did not catch why, likely due to message size limit from batching

[eliteprox]

We could possibly improve this by allowing _wait_for_interval to raise it's own error