X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=src%2Fo2ims%2Fservice%2Fmessagebus.py;fp=src%2Fo2ims%2Fservice%2Fmessagebus.py;h=0000000000000000000000000000000000000000;hb=81e3575a77366f30c2049f98c48a3087db0ea992;hp=bff45ab893c6eea41c83138bee7be5b7963ccb92;hpb=8339c9a882a586578b37f44a504e21c5208611c0;p=pti%2Fo2.git diff --git a/src/o2ims/service/messagebus.py b/src/o2ims/service/messagebus.py deleted file mode 100644 index bff45ab..0000000 --- a/src/o2ims/service/messagebus.py +++ /dev/null @@ -1,69 +0,0 @@ -# Copyright (C) 2021 Wind River Systems, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -# pylint: disable=broad-except, attribute-defined-outside-init -from __future__ import annotations -import logging -from typing import Callable, Dict, List, Union, Type, TYPE_CHECKING -from o2ims.domain import commands, events - -if TYPE_CHECKING: - from . import unit_of_work - -logger = logging.getLogger(__name__) - -Message = Union[commands.Command, events.Event] - - -class MessageBus: - def __init__( - self, - uow: unit_of_work.AbstractUnitOfWork, - event_handlers: Dict[Type[events.Event], List[Callable]], - command_handlers: Dict[Type[commands.Command], Callable], - ): - self.uow = uow - self.event_handlers = event_handlers - self.command_handlers = command_handlers - - def handle(self, message: Message): - self.queue = [message] - while self.queue: - message = self.queue.pop(0) - if isinstance(message, events.Event): - self.handle_event(message) - elif isinstance(message, commands.Command): - self.handle_command(message) - else: - raise Exception(f"{message} was not an Event or Command") - - def handle_event(self, event: events.Event): - for handler in self.event_handlers[type(event)]: - try: - logger.debug("handling event %s with handler %s", event, handler) - handler(event) - self.queue.extend(self.uow.collect_new_events()) - except Exception: - logger.exception("Exception handling event %s", event) - continue - - def handle_command(self, command: commands.Command): - logger.debug("handling command %s", command) - try: - handler = self.command_handlers[type(command)] - handler(command) - self.queue.extend(self.uow.collect_new_events()) - except Exception: - logger.exception("Exception handling command %s", command) - raise