aioemit allows you to manage events asynchonosly and notify subscribers when those events occur. It provides a simple way to implement event bus pattern in an event driven architecture application.
pip install aioemitThe Event class represents an event with a specified event type and optional data. You can create an event by initializing an instance of the Event class with the event type and data (if any).
event = Event("example_event", "example_data")To use the event emitter, create an instance of the Emitter class.
emitter = Emitter()To subscribe to events, use the subscribe method of the Emitter class. Pass the event type and an observer function that will be called when the event is emitted.
def event_observer(event):
# Handle the event
print("Received event:", event)
emitter.subscribe("example_event", event_observer)If you no longer want to receive notifications for a specific event, you can unsubscribe from it using the unsubscribe method. Provide the event type and the observer function that you want to remove.
emitter.unsubscribe("example_event", event_observer)To emit an event and notify all subscribers, use the emit method of the Emitter class. Pass the event you want to emit.
event = Event("example_event", "example_data")
await emitter.emit(event)The emit method will asynchronously call all the subscribed observer functions that are associated with the event type.
This project is licensed under the MIT License. Feel free to use, modify, and distribute the code as per the terms of the license.