aswi2021vochomurka-gitlab/aswi2021vochomurka/service/subscriber.py @ 64ca87d0

import time
from typing import Dict

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.base import STATE_STOPPED

from aswi2021vochomurka.model.Message import Message
from aswi2021vochomurka.service.file_manager import FileManager
from aswi2021vochomurka.service.subscriber_callback import SubscriberCallback
from aswi2021vochomurka.service.subscriber_params import SubscriberParams


class Subscriber:
    """
    Subscriber is responsible for establishing communication with broker and notifying
    about new message via callback
    Subscriber must be started via 'start' method and stopped via 'stop' method
    """
    callback: SubscriberCallback
    params: SubscriberParams

    scheduler: BackgroundScheduler
    files: Dict[str, FileManager] = {}

    def __init__(self, callback: SubscriberCallback, params: SubscriberParams):
        """
        Constructor
        :param callback: callback
        :param params: params
        """
        self.callback = callback
        self.params = params

    def start(self):
        """
        Start subscriber
        """
        # start scheduler to check closed topics
        self.scheduler = BackgroundScheduler()
        self.scheduler.add_job(self.check_closed_topics, 'interval', seconds=self.params.closeLimit)
        self.scheduler.start()

    def stop(self):
        """
        Stop subscriber
        """
        if self.scheduler.state != STATE_STOPPED:
            self.scheduler.shutdown()
        self.close_files()

    def close_files(self):
        """
        Close all open files
        """
        for topic in self.files:
            self.files.get(topic).close()
        self.files.clear()

    def check_closed_topics(self):
        """
        May be called periodically for checking for expired timeout for closing topic
        """
        t = time.time()
        for topic in list(self.files):
            file = self.files.get(topic)
            if t - file.lastUpdate > self.params.closeLimit:
                self.callback.onCloseTopic(topic)
                file.close()
                self.files.pop(topic)

    def write_to_file(self, message: Message):
        """
        Write message to file
        :param message: message
        """
        if message.topic in self.files:
            # file exist, just append message
            self.files.get(message.topic).write(message)
        else:
            # new message for this topic, create new file
            fm = FileManager(message.topic, message)
            self.files[message.topic] = fm