"""Event Manager."""fromcollectionsimportdequefromgeneric.eventimportEvent,Handlerfromgeneric.eventimportManageras_Managerfromgaphor.abcimportServicedefevent_handler(*event_types):"""Mark a function/method as an event handler for a particular type of event."""defwrapper(func):func.__event_types__=event_typesreturnfuncreturnwrapper
[docs]defsubscribe(self,handler:Handler)->None:"""Register a handler. Handlers are triggered (executed) when specific events are emitted through the handle() method. """self._subscribe(handler,self._events)
[docs]defpriority_subscribe(self,handler:Handler)->None:"""Register a handler. Priority handlers are executed directly. They should not raise other events, cause that can cause a problem in the exection order. It's basically to make sure that all events are recorded by the undo manager. """self._subscribe(handler,self._priority)
def_subscribe(self,handler:Handler,manager:_Manager)->None:event_types=getattr(handler,"__event_types__",None)ifnotevent_types:raiseException(f"No event types provided for function {handler}")foretinevent_types:manager.subscribe(handler,et)
[docs]defunsubscribe(self,handler:Handler)->None:"""Unregister a previously registered handler."""event_types=getattr(handler,"__event_types__",None)ifnotevent_types:raiseException(f"No event types provided for function {handler}")foretinevent_types:self._priority.unsubscribe(handler,et)self._events.unsubscribe(handler,et)
[docs]defhandle(self,*events:Event)->None:"""Send event notifications to registered handlers."""queue=self._queuequeue.extendleft(events)foreventinevents:self._priority.handle(event)ifnotself._handling:self._handling=Truetry:whilequeue:self._events.handle(queue.pop())finally:self._handling=False