Changes:

Common:
- Implemented Message Brokering framework supporting multiple backends
- Implemented InMemory Message Brokering Backend for testing purposes
- Implemented Redis Backend
- Implemented Message Brokering unit tests
- Minor improvements in ORM Framework
- Added Message Broker framework unit testing into automated local tests

run_local_tests.sh

@@ -16,7 +16,8 @@ rm -f $COVERAGEFILE
 
 coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
     common/metrics/tests/test_unitary.py \
-    common/orm/tests/test_unitary.py
+    common/orm/tests/test_unitary.py \
+    common/message_broker/tests/test_unitary.py
 
 #coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
 #    centralizedcybersecurity/tests/test_unitary.py

src/common/message_broker/Constants.py

+CONSUME_TIMEOUT = 0.1 # seconds

src/common/message_broker/Factory.py

+import logging, os
+from typing import Optional, Union
+from .backend._Backend import _Backend
+from .backend.BackendEnum import BackendEnum
+from .backend.inmemory.InMemoryBackend import InMemoryBackend
+from .backend.redis.RedisBackend import RedisBackend
+
+LOGGER = logging.getLogger(__name__)
+
+BACKENDS = {
+    BackendEnum.INMEMORY.value: InMemoryBackend,
+    BackendEnum.REDIS.value: RedisBackend,
+    #BackendEnum.KAFKA.value: KafkaBackend,
+    #BackendEnum.RABBITMQ.value: RabbitMQBackend,
+    #BackendEnum.ZEROMQ.value: ZeroMQBackend,
+}
+
+DEFAULT_MB_BACKEND = BackendEnum.INMEMORY
+
+def get_messagebroker_backend(backend : Optional[Union[str, BackendEnum]] = None, **settings) -> _Backend:
+    # return an instance of MessageBroker initialized with selected backend.
+    # The backend is selected using following criteria (first that is not None is selected):
+    # 1. user selected by parameter (backend=...)
+    # 2. environment variable MB_BACKEND
+    # 3. default backend: INMEMORY
+    if backend is None: backend = os.environ.get('MB_BACKEND', DEFAULT_MB_BACKEND)
+    if backend is None: raise Exception('MessageBroker Backend not specified')
+    if isinstance(backend, BackendEnum): backend = backend.value
+    backend_class = BACKENDS.get(backend)
+    if backend_class is None: raise Exception('Unsupported MessageBrokerBackend({:s})'.format(backend))
+    LOGGER.info('Selected MessageBroker Backend: {:s}'.format(backend))
+    return backend_class(**settings)

src/common/message_broker/Message.py

+from typing import NamedTuple
+
+class Message(NamedTuple):
+    topic: str
+    content: str

src/common/message_broker/MessageBroker.py

-import logging, sys, threading, time
-from queue import Queue, Empty
-from typing import Dict, Iterator, NamedTuple, Optional, Set
+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__)
 
-SUBSCRIBER_DEAD_TIMEOUT = 5.0 # seconds
-CONSUME_TIMEOUT = 0.5 # seconds
-
-class Watchdog:
-    def __init__(self, dead_timeout) -> None:
-        self._lock = threading.Lock()
-        self._dead_timeout = dead_timeout
-        self._timestamp = time.time()
-
-    def refresh(self):
-        with self._lock:
-            self._timestamp = time.time()
-
-    def is_alive(self):
-        with self._lock:
-            return (time.time() - self._timestamp) < self._dead_timeout
-
-class Message(NamedTuple):
-    topic: str
-    content: str
-
-class Subscriber(Watchdog):
-    def __init__(self, consume_timeout=CONSUME_TIMEOUT, dead_timeout=SUBSCRIBER_DEAD_TIMEOUT) -> None:
-        super().__init__(dead_timeout)
-        self._consume_timeout = consume_timeout
-        self._queue = Queue()
-
-    def publish(self, message : Message) -> None:
-        self._queue.put_nowait(message)
-
-    def consume(self) -> Optional[Message]:
-        self.refresh()
-        try:
-            return self._queue.get(block=True, timeout=self._consume_timeout)
-        except Empty:
-            return None
-
-class Topic:
-    def __init__(self) -> None:
-        self._subscribers : Set[Subscriber] = set()
-
-    def add_subscriber(self, subscriber : Subscriber) -> None:
-        self._subscribers.add(subscriber)
-
-    def remove_subscriber(self, subscriber : Subscriber) -> None:
-        self._subscribers.discard(subscriber)
-
-    def publish(self, message : Message) -> None:
-        dead_subscribers = set()
-        for subscriber in self._subscribers:
-            if subscriber.is_alive():
-                subscriber.publish(message)
-            else:
-                dead_subscribers.add(subscriber)
-        for dead_subscriber in dead_subscribers:
-            self.remove_subscriber(dead_subscriber)
-
 class MessageBroker:
-    def __init__(self):
-        self._terminate = threading.Event()
-        self._topics : Dict[str, Topic] = {}
+    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
 
-    def get_or_create_topic(self, topic_name) -> Topic:
-        return self._topics.setdefault(topic_name, Topic())
-
-    def terminate(self) -> None:
-        self._terminate.set()
+    @property
+    def backend(self) -> _Backend: return self._backend
 
     def publish(self, message : Message) -> None:
-        self.get_or_create_topic(message.topic).publish(message)
-
-    def consume(
-        self, topic_names : Set[str], consume_timeout=CONSUME_TIMEOUT, dead_timeout=SUBSCRIBER_DEAD_TIMEOUT
-        ) -> Iterator[Message]:
-
-        subscriber = Subscriber(consume_timeout=consume_timeout, dead_timeout=dead_timeout)
-        for topic_name in topic_names:
-            self.get_or_create_topic(topic_name).add_subscriber(subscriber)
-
-        while not self._terminate.is_set():
-            message = subscriber.consume()
-            if message is None: continue
-            yield message
-
-        for topic_name in topic_names:
-            self.get_or_create_topic(topic_name).remove_subscriber(subscriber)
-
-class Consumer(threading.Thread):
-    def __init__(
-        self, message_broker : MessageBroker, topic_names : Set[str], consume_timeout=CONSUME_TIMEOUT,
-        dead_timeout=SUBSCRIBER_DEAD_TIMEOUT) -> None:
-
-        super().__init__(daemon=True)
-        self._message_broker = message_broker
-        self._topic_names = topic_names
-        self._consume_timeout = consume_timeout
-        self._dead_timeout = dead_timeout
-
-    def run(self) -> None:
-        print(time.time(), self.name, 'subscribes to', self._topic_names)
-        messages = self._message_broker.consume(
-            self._topic_names, consume_timeout=self._consume_timeout, dead_timeout=self._dead_timeout)
-        for message in messages:
-            print(time.time(), self.name, 'receives', message)
-        print(time.time(), self.name, 'terminates')
-
-TOPIC_DEVICES  = 'devices'
-TOPIC_LINKS    = 'links'
-TOPIC_SERVICES = 'services'
-
-def main():
-    message_broker = MessageBroker()
-
-    consumer1 = Consumer(message_broker, {TOPIC_DEVICES, TOPIC_LINKS})
-    consumer1.start()
-
-    consumer2 = Consumer(message_broker, {TOPIC_DEVICES, TOPIC_SERVICES})
-    consumer2.start()
-
-    consumer3 = Consumer(message_broker, {TOPIC_SERVICES})
-    consumer3.start()
-
-    time.sleep(SUBSCRIBER_DEAD_TIMEOUT)
-
-    message_broker.publish(Message(topic=TOPIC_DEVICES, content='new-device-01'))
-    message_broker.publish(Message(topic=TOPIC_DEVICES, content='new-device-02'))
-    message_broker.publish(Message(topic=TOPIC_LINKS,   content='new-link-01-02'))
-
-    time.sleep(SUBSCRIBER_DEAD_TIMEOUT)
-
-    message_broker.publish(Message(topic=TOPIC_DEVICES,  content='update-device-01'))
-    message_broker.publish(Message(topic=TOPIC_DEVICES,  content='update-device-02'))
-    message_broker.publish(Message(topic=TOPIC_SERVICES, content='new-service-01-02'))
+        self._backend.publish(message.topic, message.content)
 
-    message_broker.terminate()
-    return 0
+    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)
 
-if __name__ == '__main__':
-    sys.exit(main())
+    def terminate(self):
+        self._backend.terminate()

src/common/message_broker/backend/BackendEnum.py

+from enum import Enum
+
+class BackendEnum(Enum):
+    INMEMORY = 'inmemory'
+    REDIS = 'redis'
+    #KAFKA = 'kafka'
+    #RABBITMQ = 'rabbitmq'
+    #ZEROMQ = 'zeromq'

src/common/message_broker/backend/_Backend.py

+from typing import Iterator, Set, Tuple
+
+class _Backend:
+    def __init__(self, **settings) -> None:
+        raise NotImplementedError()
+
+    def terminate(self) -> None:
+        raise NotImplementedError()
+
+    def publish(self, topic_name : str, message_content : str) -> None:
+        raise NotImplementedError()
+
+    def consume(self, topic_names : Set[str], consume_timeout : float) -> Iterator[Tuple[str, str]]:
+        raise NotImplementedError()

src/common/message_broker/backend/inmemory/InMemoryBackend.py

+# InMemeory MessageBroker Backend
+# -------------------------------
+# - WARNING: DESIGNED AND BUILT FOR UNIT TESTING AND INTEGRATION TESTING PURPOSES ONLY !!!
+#            USE ANOTHER BACKEND IN PRODUCTION ENVIRONMENTS.
+
+import logging, threading
+from queue import Queue, Empty
+from typing import Dict, Iterator, Set, Tuple
+from .._Backend import _Backend
+
+LOGGER = logging.getLogger(__name__)
+
+class InMemoryBackend(_Backend):
+    def __init__(self, **settings) -> None: # pylint: disable=super-init-not-called
+        self._lock = threading.Lock()
+        self._terminate = threading.Event()
+        self._topic__to__queues : Dict[str, Set[Queue]] = {}
+
+    def terminate(self) -> None:
+        self._terminate.set()
+
+    def publish(self, topic_name : str, message_content : str) -> None:
+        queues = self._topic__to__queues.get(topic_name, None)
+        if queues is None: return
+        for queue in queues: queue.put_nowait((topic_name, message_content))
+
+    def consume(self, topic_names : Set[str], consume_timeout : float) -> Iterator[Tuple[str, str]]:
+        queue = Queue()
+        for topic_name in topic_names:
+            self._topic__to__queues.setdefault(topic_name, set()).add(queue)
+
+        while not self._terminate.is_set():
+            try:
+                message = queue.get(block=True, timeout=consume_timeout)
+            except Empty:
+                continue
+            if message is None: continue
+            yield message
+
+        for topic_name in topic_names:
+            self._topic__to__queues.get(topic_name, set()).discard(queue)

src/common/message_broker/backend/redis/RedisBackend.py

+import os, threading
+from typing import Any, Dict, Iterator, Set, Tuple
+from redis.client import Redis
+
+from common.message_broker.Message import Message
+from .._Backend import _Backend
+
+DEFAULT_SERVICE_HOST = '127.0.0.1'
+DEFAULT_SERVICE_PORT = 6379
+DEFAULT_DATABASE_ID  = 0
+
+def get_setting(settings : Dict[str, Any], name : str, default : Any) -> Any:
+    value = settings.get(name, os.environ.get(name))
+    return default if value is None else value
+
+class RedisBackend(_Backend):
+    def __init__(self, **settings) -> None: # pylint: disable=super-init-not-called
+        host = get_setting(settings, 'REDIS_SERVICE_HOST', DEFAULT_SERVICE_HOST)
+        port = get_setting(settings, 'REDIS_SERVICE_PORT', DEFAULT_SERVICE_PORT)
+        dbid = get_setting(settings, 'REDIS_DATABASE_ID',  DEFAULT_DATABASE_ID )
+        self._client = Redis.from_url('redis://{host}:{port}/{dbid}'.format(host=host, port=port, dbid=dbid))
+        self._terminate = threading.Event()
+
+    def terminate(self) -> None:
+        self._terminate.set()
+
+    def publish(self, topic_name : str, message_content : str) -> None:
+        self._client.publish(topic_name, message_content)
+
+    def consume(self, topic_names : Set[str], consume_timeout : float) -> Iterator[Tuple[str, str]]:
+        pubsub = self._client.pubsub(ignore_subscribe_messages=True)
+        for topic_name in topic_names: pubsub.subscribe(topic_name)
+
+        while not self._terminate.is_set():
+            message = pubsub.get_message(ignore_subscribe_messages=True, timeout=consume_timeout)
+            if message is None: continue
+            if message['type'] not in {'message', 'pmessage'}: continue
+            topic = message['channel'].decode('UTF-8')
+            content = message['data'].decode('UTF-8')
+            yield Message(topic, content)
+
+        pubsub.unsubscribe()
+        while pubsub.get_message() is not None: pass
+        pubsub.close()

src/common/message_broker/tests/test_unitary.py

+import logging, pytest, threading, time
+from typing import List, Set
+from common.message_broker.Factory import get_messagebroker_backend
+from common.message_broker.Message import Message
+from common.message_broker.MessageBroker import MessageBroker
+from common.message_broker.backend.BackendEnum import BackendEnum
+from common.message_broker.backend._Backend import _Backend
+
+logging.basicConfig(level=logging.INFO)
+LOGGER = logging.getLogger(__name__)
+
+SCENARIOS = [
+    (BackendEnum.INMEMORY, {}),
+    (BackendEnum.REDIS,    {
+        'REDIS_SERVICE_HOST': '10.1.7.194',
+        'REDIS_SERVICE_PORT': 30283,
+        'REDIS_DATABASE_ID': 0,
+    }),
+]
+
+CONSUME_TIMEOUT = 0.1 # seconds
+
+TOPIC_DEVICES  = 'devices'
+TOPIC_LINKS    = 'links'
+TOPIC_SERVICES = 'services'
+
+class Consumer(threading.Thread):
+    def __init__(
+        self, message_broker : MessageBroker, # pylint: disable=redefined-outer-name
+        topic_names : Set[str], output_list : List[Message],
+        consume_timeout=CONSUME_TIMEOUT) -> None:
+
+        super().__init__(daemon=True)
+        self._message_broker = message_broker
+        self._topic_names = topic_names
+        self._output_list = output_list
+        self._consume_timeout = consume_timeout
+
+    def run(self) -> None:
+        LOGGER.info('{:s} subscribes to topics {:s}'.format(self.name, str(self._topic_names)))
+        for message in self._message_broker.consume(self._topic_names, consume_timeout=self._consume_timeout):
+            LOGGER.info('{:s} receives {:s}'.format(self.name, str(message)))
+            self._output_list.append(message)
+        LOGGER.info('{:s} terminates')
+
+@pytest.fixture(scope='session', ids=[str(scenario[0].value) for scenario in SCENARIOS], params=SCENARIOS)
+def message_broker(request):
+    backend,settings = request.param
+    LOGGER.info('Running fixture with backend={:s}, settings={:s}...'.format(str(backend), str(settings)))
+    return MessageBroker(get_messagebroker_backend(backend=backend, **settings))
+
+def test_messagebroker_instantiation():
+    with pytest.raises(AttributeError) as e:
+        MessageBroker(None)
+    str_class_path = '{}.{}'.format(_Backend.__module__, _Backend.__name__)
+    assert str(e.value) == 'backend must inherit from {}'.format(str_class_path)
+
+    assert MessageBroker(get_messagebroker_backend(BackendEnum.INMEMORY)) is not None
+
+def test_messagebroker(message_broker : MessageBroker): # pylint: disable=redefined-outer-name
+    output_list1 : List[Message] = []
+    consumer1 = Consumer(message_broker, {TOPIC_DEVICES, TOPIC_LINKS}, output_list1)
+    consumer1.start()
+
+    output_list2 : List[Message] = []
+    consumer2 = Consumer(message_broker, {TOPIC_DEVICES, TOPIC_SERVICES}, output_list2)
+    consumer2.start()
+
+    output_list3 : List[Message] = []
+    consumer3 = Consumer(message_broker, {TOPIC_SERVICES}, output_list3)
+    consumer3.start()
+
+    LOGGER.info('delay')
+    time.sleep(0.5)
+
+    message = Message(topic=TOPIC_DEVICES, content='new-device-01')
+    LOGGER.info('publish message={:s}'.format(str(message)))
+    message_broker.publish(message)
+
+    message = Message(topic=TOPIC_DEVICES, content='new-device-02')
+    LOGGER.info('publish message={:s}'.format(str(message)))
+    message_broker.publish(message)
+
+    message = Message(topic=TOPIC_LINKS,   content='new-link-01-02')
+    LOGGER.info('publish message={:s}'.format(str(message)))
+    message_broker.publish(message)
+
+    LOGGER.info('delay')
+    time.sleep(0.1)
+
+    message = Message(topic=TOPIC_DEVICES,  content='update-device-01')
+    LOGGER.info('publish message={:s}'.format(str(message)))
+    message_broker.publish(message)
+
+    message = Message(topic=TOPIC_DEVICES,  content='update-device-02')
+    LOGGER.info('publish message={:s}'.format(str(message)))
+    message_broker.publish(message)
+
+    message = Message(topic=TOPIC_SERVICES, content='new-service-01-02')
+    LOGGER.info('publish message={:s}'.format(str(message)))
+    message_broker.publish(message)
+
+    LOGGER.info('delay')
+    time.sleep(0.5)
+
+    LOGGER.info('terminate')
+    message_broker.terminate()
+
+    LOGGER.info('join')
+    consumer1.join()
+    consumer2.join()
+    consumer3.join()
+
+    LOGGER.info('output_list1={:s}'.format(str(output_list1)))
+    LOGGER.info('output_list2={:s}'.format(str(output_list2)))
+    LOGGER.info('output_list3={:s}'.format(str(output_list3)))
+
+    assert len(output_list1) == 5
+    assert output_list1[0].topic == TOPIC_DEVICES
+    assert output_list1[0].content == 'new-device-01'
+    assert output_list1[1].topic == TOPIC_DEVICES
+    assert output_list1[1].content == 'new-device-02'
+    assert output_list1[2].topic == TOPIC_LINKS
+    assert output_list1[2].content == 'new-link-01-02'
+    assert output_list1[3].topic == TOPIC_DEVICES
+    assert output_list1[3].content == 'update-device-01'
+    assert output_list1[4].topic == TOPIC_DEVICES
+    assert output_list1[4].content == 'update-device-02'
+
+    assert len(output_list2) == 5
+    assert output_list2[0].topic == TOPIC_DEVICES
+    assert output_list2[0].content == 'new-device-01'
+    assert output_list2[1].topic == TOPIC_DEVICES
+    assert output_list2[1].content == 'new-device-02'
+    assert output_list2[2].topic == TOPIC_DEVICES
+    assert output_list2[2].content == 'update-device-01'
+    assert output_list2[3].topic == TOPIC_DEVICES
+    assert output_list2[3].content == 'update-device-02'
+    assert output_list2[4].topic == TOPIC_SERVICES
+    assert output_list2[4].content == 'new-service-01-02'
+
+    assert len(output_list3) == 1
+    assert output_list3[0].topic == TOPIC_SERVICES
+    assert output_list3[0].content == 'new-service-01-02'

src/common/orm/Factory.py

@@ -10,8 +10,8 @@ LOGGER = logging.getLogger(__name__)
 BACKENDS = {
     BackendEnum.INMEMORY.value: InMemoryBackend,
     BackendEnum.REDIS.value: RedisBackend,
-    #BackendEnum.MONGO.value: MongoBackend,
-    #BackendEnum.RETHINK.value: RethinkBackend,
+    #BackendEnum.MONGODB.value: MongoDBBackend,
+    #BackendEnum.RETHINKDB.value: RethinkDBBackend,
     #BackendEnum.ETCD.value: EtcdBackend,
 }
 

src/common/orm/backend/BackendEnum.py

@@ -3,6 +3,6 @@ from enum import Enum
 class BackendEnum(Enum):
     INMEMORY = 'inmemory'
     REDIS = 'redis'
-    #MONGO = 'mongo'
-    #RETHINK = 'rethink'
+    #MONGODB = 'mongodb'
+    #RETHINKDB = 'rethinkdb'
     #ETCD = 'etcd'

src/common/orm/backend/inmemory/InMemoryBackend.py

@@ -14,7 +14,7 @@ from .Tools import get_dict, get_list, get_or_create_dict, get_or_create_list, g
 LOGGER = logging.getLogger(__name__)
 
 class InMemoryBackend(_Backend):
-    def __init__(self, **settings):
+    def __init__(self, **settings): # pylint: disable=super-init-not-called
         self._lock = threading.Lock()
         self._keys : Dict[str, Union[Set[str], List[str], Dict[str, str], str]]= {} # key => set/list/dict/string
 

src/common/orm/backend/redis/RedisBackend.py

@@ -14,7 +14,7 @@ def get_setting(settings : Dict[str, Any], name : str, default : Any) -> Any:
     return default if value is None else value
 
 class RedisBackend(_Backend):
-    def __init__(self, **settings) -> None:
+    def __init__(self, **settings) -> None: # pylint: disable=super-init-not-called
         host = get_setting(settings, 'REDIS_SERVICE_HOST', DEFAULT_SERVICE_HOST)
         port = get_setting(settings, 'REDIS_SERVICE_PORT', DEFAULT_SERVICE_PORT)
         dbid = get_setting(settings, 'REDIS_DATABASE_ID',  DEFAULT_DATABASE_ID )