o2common/service/messagebus.py

   1 # Copyright (C) 2021 Wind River Systems, Inc.
   2 #
   3 #  Licensed under the Apache License, Version 2.0 (the "License");
   4 #  you may not use this file except in compliance with the License.
   5 #  You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #  Unless required by applicable law or agreed to in writing, software
  10 #  distributed under the License is distributed on an "AS IS" BASIS,
  11 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #  See the License for the specific language governing permissions and
  13 #  limitations under the License.
  14
  15 # pylint: disable=broad-except, attribute-defined-outside-init
  16 from __future__ import annotations
  17 from typing import Callable, Dict, List, Union, Type, TYPE_CHECKING
  18 from o2common.domain import commands, events
  19
  20 if TYPE_CHECKING:
  21     from . import unit_of_work
  22
  23 from o2common.helper import o2logging
  24 logger = o2logging.get_logger(__name__)
  25
  26 Message = Union[commands.Command, events.Event]
  27
  28
  29 class MessageBus:
  30     def __init__(
  31         self,
  32         uow: unit_of_work.AbstractUnitOfWork,
  33         event_handlers: Dict[Type[events.Event], List[Callable]],
  34         command_handlers: Dict[Type[commands.Command], Callable],
  35     ):
  36         self.uow = uow
  37         self.event_handlers = event_handlers
  38         self.command_handlers = command_handlers
  39
  40     def handle(self, message: Message):
  41         self.queue = [message]
  42         while self.queue:
  43             message = self.queue.pop(0)
  44             if not message:
  45                 continue
  46             elif isinstance(message, events.Event):
  47                 self.handle_event(message)
  48             elif isinstance(message, commands.Command):
  49                 self.handle_command(message)
  50             else:
  51                 raise Exception(f"{message} was not an Event or Command")
  52
  53     def handle_event(self, event: events.Event):
  54         for handler in self.event_handlers[type(event)]:
  55             try:
  56                 logger.debug("handling event %s with handler %s",
  57                              event, handler)
  58                 handler(event)
  59                 self.queue.extend(self.uow.collect_new_events())
  60             except Exception:
  61                 logger.exception("Exception handling event %s", event)
  62                 continue
  63
  64     def handle_command(self, command: commands.Command):
  65         logger.debug("handling command %s", command)
  66         try:
  67             handler = self.command_handlers[type(command)]
  68             handler(command)
  69             self.queue.extend(self.uow.collect_new_events())
  70         except Exception as ex:
  71             logger.exception("Exception handling command %s", command)
  72             raise ex