1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-23 19:44:20 +00:00

Rename event to HeraldEvent

This commit is contained in:
Steffo 2020-07-26 19:29:42 +02:00
parent f2ff31d155
commit 51a50a3b2a
5 changed files with 10 additions and 10 deletions

View file

@ -1,7 +1,7 @@
from royalnet.commands import *
class ExceptionEvent(Event):
class ExceptionEvent(HeraldEvent):
name = "exception"
def run(self, **kwargs):

View file

@ -3,7 +3,7 @@
from .command import Command
from .commanddata import CommandData
from .commandargs import CommandArgs
from .event import Event
from .heraldevent import HeraldEvent
from .errors import \
CommandError, InvalidInputError, UnsupportedError, ConfigurationError, ExternalError, UserError, ProgramError
from .keyboardkey import KeyboardKey
@ -20,7 +20,7 @@ __all__ = [
"ExternalError",
"UserError",
"ProgramError",
"Event",
"HeraldEvent",
"KeyboardKey",
"ConfigDict",
]

View file

@ -5,7 +5,7 @@ if TYPE_CHECKING:
from ..serf import Serf
class Event:
class HeraldEvent:
"""A remote procedure call triggered by a :mod:`royalnet.herald` request."""
name = NotImplemented

View file

@ -95,7 +95,7 @@ class Constellation:
self.Interface: Type[rc.CommandInterface] = self.interface_factory()
"""The :class:`~rc.CommandInterface` class of this :class:`Constellation`."""
self.events: Dict[str, rc.Event] = {}
self.events: Dict[str, rc.HeraldEvent] = {}
"""A dictionary containing all :class:`~rc.Event` that can be handled by this :class:`Constellation`."""
self.starlette = starlette.applications.Starlette(debug=__debug__)
@ -208,7 +208,7 @@ class Constellation:
async def network_handler(self, message: Union[rh.Request, rh.Broadcast]) -> rh.Response:
try:
event: rc.Event = self.events[message.handler]
event: rc.HeraldEvent = self.events[message.handler]
except KeyError:
log.warning(f"No event for '{message.handler}'")
return rh.ResponseFailure("no_event", f"This serf does not have any event for {message.handler}.")
@ -228,7 +228,7 @@ class Constellation:
elif isinstance(message, rh.Broadcast):
await event.run(**message.data)
def register_events(self, events: List[Type[rc.Event]], pack_cfg: Dict[str, Any]):
def register_events(self, events: List[Type[rc.HeraldEvent]], pack_cfg: Dict[str, Any]):
for SelectedEvent in events:
# Create a new interface
interface = self.Interface(config=pack_cfg)

View file

@ -89,7 +89,7 @@ class Serf(abc.ABC):
self.herald_task: Optional[aio.Task] = None
"""A reference to the :class:`asyncio.Task` that runs the :class:`Link`."""
self.events: Dict[str, Event] = {}
self.events: Dict[str, HeraldEvent] = {}
"""A dictionary containing all :class:`Event` that can be handled by this :class:`Serf`."""
self.commands: Dict[str, Command] = {}
@ -211,7 +211,7 @@ class Serf(abc.ABC):
herald_cfg["name"] = self.interface_name
self.herald: rh.Link = rh.Link(rh.Config.from_config(**herald_cfg), self.network_handler)
def register_events(self, events: List[Type[Event]], pack_cfg: Dict[str, Any]):
def register_events(self, events: List[Type[HeraldEvent]], pack_cfg: Dict[str, Any]):
for SelectedEvent in events:
# Initialize the event
try:
@ -230,7 +230,7 @@ class Serf(abc.ABC):
async def network_handler(self, message: Union[rh.Request, rh.Broadcast]) -> rh.Response:
try:
event: Event = self.events[message.handler]
event: HeraldEvent = self.events[message.handler]
except KeyError:
log.warning(f"No event for '{message.handler}'")
return rh.ResponseFailure("no_event", f"This serf does not have any event for {message.handler}.")