craftbeerpi4-pione/core/eventbus.py

import asyncio
import inspect
import logging
import json

class EventBus(object):
    class Node(object):
        __slots__ = '_children', '_content'

        def __init__(self):
            self._children = {}
            self._content = None

    class Content(object):
        def __init__(self, parent, topic, method, once):
            self.parent = parent
            self.method = method
            self.name = method.__name__
            self.once = once
            self.topic = topic

    def register(self, topic, method, once=False):

        if method in self.registry:
            raise RuntimeError("Method %s already registerd. Please unregister first!" % method.__name__)
        self.logger.info("Topic %s", topic)
        node = self._root
        for sym in topic.split('/'):
            node = node._children.setdefault(sym, self.Node())

        if not isinstance(node._content, list):
            node._content = []

        c = self.Content(node, topic, method, once)
        node._content.append(c)
        self.registry[method] = c

    def get_callbacks(self, key):
        try:
            node = self._root
            for sym in key.split('/'):
                node = node._children[sym]
            if node._content is None:
                raise KeyError(key)
            return node._content
        except KeyError:
            raise KeyError(key)

    def unregister(self, method):
        self.logger.info("Unregister %s", method.__name__)
        if method in self.registry:
            content = self.registry[method]
            clean_idx = None
            for idx, content_obj in enumerate(content.parent._content):
                if method == content_obj.method:
                    clean_idx = idx
                    break
            if clean_idx is not None:
                del content.parent._content[clean_idx]

    def __init__(self, loop, cbpi):
        self.logger = logging.getLogger(__name__)
        self.cbpi = cbpi
        self._root = self.Node()
        self.registry = {}
        self.docs = {}
        if loop is not None:
            self.loop = loop
        else:
            self.loop = asyncio.get_event_loop()

        print(self.loop)

    def fire(self, topic: str, **kwargs) -> None:
        self.logger.info("EMIT EVENT %s Data: %s", topic, kwargs)

        #self.cbpi.ws.send(json.dumps(dict(topic=topic, data=dict(**kwargs))))
        trx = dict(i=0)
        for e in self.iter_match(topic):
            content_array = e
            keep_idx = []
            for idx, content_obj in enumerate(content_array):

                if inspect.iscoroutinefunction(content_obj.method):
                    if hasattr(content_obj.method, "future"):

                        self.loop.create_task(content_obj.method(**kwargs, future=content_obj.method.future, topic=topic))
                    else:
                        self.loop.create_task(content_obj.method(**kwargs, topic = topic))
                else:
                    if hasattr(content_obj.method, "future"):
                        content_obj.method(**kwargs, future=content_obj.method.future, topic=topic)
                    else:
                        content_obj.method(**kwargs, topic = topic)


                if content_obj.once is False:
                    keep_idx.append(idx)

            # FILTER only elements with are required
            if len(keep_idx) < len(e):
                e[0].parent._content = [e[0].parent._content[i] for i in keep_idx]



    def dump(self):
        def rec(node, i=0):
            result = []
            if node._content is not None:
                for c in node._content:
                    result.append(dict(topic=c.topic, method=c.method.__name__, path=c.method.__module__, once=c.once))

            if node._children is not None:
                for c in node._children:
                    result = result + rec(node._children[c], i + 1)
            return result

        result = rec(self._root)

        return result

    def iter_match(self, topic):

        lst = topic.split('/')
        normal = not topic.startswith('$')

        def rec(node, i=0):
            if i == len(lst):
                if node._content is not None:
                    yield node._content
            else:
                part = lst[i]
                if part in node._children:
                    for content in rec(node._children[part], i + 1):
                        yield content
                if '+' in node._children and (normal or i > 0):
                    for content in rec(node._children['+'], i + 1):
                        yield content
            if '#' in node._children and (normal or i > 0):
                content = node._children['#']._content
                if content is not None:
                    yield content

        return rec(self._root)