import threading
from typing import Optional
import logging
from olc.config.properties_loader import OlcConfigManager
from olc.bean.tag_group import TagGroup
from olc.cache.safe__ttl_cache import SafeTTLCache
from olc.event.sny_event import Event
from olc.statistic.statistic import IStatistic, OlcStatistic, StatisticSnapshot
logger = logging.getLogger(__name__)
class OlcStatisticFactory:
_instance = None
_lock = threading.RLock()
def __init__(self):
self._cache = SafeTTLCache[IStatistic](
maxsize=OlcConfigManager.get_instance().get_sdk_config().statistic_max_size,
ttl=OlcConfigManager.get_instance().get_sdk_config().statistic_expire)
self._create_lock = threading.Lock()
Event.get_instance().olc_clean_rule.connect(self.subscribe_clean)
@classmethod
def get_instance(cls):
"""
统计管理器单例
:return:
"""
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = cls()
return cls._instance
def get_statistic(self, tag_group: TagGroup) -> Optional[IStatistic]:
if tag_group is None:
return None
return self._cache.get(tag_group)
def get_and_create(self, tag_group: TagGroup) -> IStatistic:
if tag_group is None:
raise ValueError("tag_group cannot be None")
result = self._cache.get(tag_group)
if result is None:
with self._create_lock:
result = self._cache.get(tag_group)
if result is None:
result = OlcStatistic()
self._cache.put(tag_group, result)
return result
def get_all_snapshots(self) -> dict[TagGroup, StatisticSnapshot]:
items = self._cache.get_all_items()
return {tag_group: statistic.snapshot() for tag_group, statistic in items}
def subscribe_clean(self, sender, **kwargs):
"""
清理使用的钩子函数
:param sender:
:param kwargs:
:return:
"""
logger.info("clean statistic cache")
self._cache.clear()