kraken/backend/core/sig_chain/utils.py

from abc import ABCMeta
from abc import abstractmethod
import logging
import threading

logger = logging.getLogger(__name__)

class Singleton(object):
    _instance_lock = threading.Lock()
    _init_flag = False

    def __new__(cls, *args, **kw):
        with Singleton._instance_lock:
            if not hasattr(cls, '_instance'):
                orig = super(Singleton, cls)
                cls._instance = orig.__new__(cls, *args, **kw)
        return cls._instance

    @classmethod
    def clear_instance(cls):
        cls._init_flag = False
        if hasattr(cls, '_instance'):
            del cls._instance

    def __init__(self) -> None:
        if Singleton._init_flag:
            return
        Singleton._init_flag = True


class Observable(object):

    def __init__(self) -> None:
        self._observers = []

    def add_observer(self, observer):
        if observer not in self._observers:
            self._observers.append(observer)
        else:
            logger.error('Add observer %s failed !', observer)

    def remove_observer(self, observer):
        try:
            self._observers.remove(observer)
        except ValueError:
            logger.error('Failed to remove: %s', observer)

    @abstractmethod
    def notify_observers(self):
        return


class Observer(metaclass=ABCMeta):  # Observer

    @abstractmethod
    def update(self):
        pass