tg_update_polling_master_slave.py

class MasterSlave(Model):
    # There should be only one Updater in polling mode, so we have to check that via this model
    collection = 'master_slave'
    ttl = 10
    holder_thread = None

    @classmethod
    def ensure_indexes(cls):
        cls.get_collection().create_index([
            ('created', pymongo.DESCENDING),
        ], background=True, expireAfterSeconds=cls.ttl)

    @classmethod
    def poll_master(cls) -> bool:
        if cls.holder_thread:
            logger.debug('Already is the master of the Universe')
            return True

        logger.debug('Start to polling master')
        collection = cls.get_collection()

        border_time = datetime.utcnow() - timedelta(seconds=cls.ttl)
        logger.debug('Check is master exists after %s', border_time)

        res = collection.find_one({'created': {'$gte': border_time}})
        logger.debug('Current master record: %s', res)
        if res is not None:
            # A master exists
            return False

        # No master record, become the master
        cls.become_master()
        return True

    @classmethod
    def become_master(cls):
        logger.info('Become the master')
        collection = cls.get_collection()

        master_id = str(uuid.uuid4())
        collection.insert_one({'master_id': master_id, 'created': datetime.utcnow()})

        def master_holder():
            while True:
                logger.debug('Renew master record')
                collection.update_one({'master_id': master_id}, {'$set': {'created': datetime.utcnow()}})
                time.sleep(cls.ttl / 2)

        cls.holder_thread = Thread(target=master_holder, name='MasterHolder')
        cls.holder_thread.start()

    @classmethod
    def join(cls):
        if not cls.holder_thread:
            logger.warning('There is no active holder thread, do not join')
            return

        cls.holder_thread.join()