import logging from typing import Iterator, Set from .backend._Backend import _Backend from .Constants import CONSUME_TIMEOUT from .Message import Message LOGGER = logging.getLogger(__name__) class MessageBroker: def __init__(self, backend : _Backend): if not isinstance(backend, _Backend): str_class_path = '{}.{}'.format(_Backend.__module__, _Backend.__name__) raise AttributeError('backend must inherit from {}'.format(str_class_path)) self._backend = backend @property def backend(self) -> _Backend: return self._backend def publish(self, message : Message) -> None: self._backend.publish(message.topic, message.content) def consume(self, topic_names : Set[str], consume_timeout : float = CONSUME_TIMEOUT) -> Iterator[Message]: for pair in self._backend.consume(topic_names, consume_timeout=consume_timeout): yield Message(*pair) def terminate(self): self._backend.terminate()