baseagent.php

<?php
 
namespace Bitrix\Calendar\Core\Queue\Agent;
 
use Bitrix\Calendar\Core\Queue\Interfaces;
use Bitrix\Calendar\Core\Queue\Interfaces\Processor;
use CAgent;
use Exception;
 
abstract class BaseAgent
{
    private const TIME_LIMIT = 10;
 
    protected bool $isEscalated = false;
    public static function runAgent(): string
    {
        try
        {
            $runner = new static();
            $runner->run();
            self::modifyAgent($runner);
 
        }
        catch(Exception $e)
        {
            // TODO: log it
        }
 
        return static::getAgentName();
    }
 
    protected static function getAgentName(): string
    {
        return static::class . "::runAgent();";
    }
 
    protected static function getModule(): string
    {
        return 'calendar';
    }
 
    private static function modifyAgent(BaseAgent $runner)
    {
        $agent = CAgent::getList(
            [],
            [
                'MODULE_ID' => self::getModule(),
                '=NAME' =>self::getAgentName(),
            ]
        )->Fetch();
        if ($agent)
        {
            $interval = $runner->isEscalated
                ? $runner->getEscalatedInterval()
                : $runner->getInterval()
            ;
            if ((int)$agent['AGENT_INTERVAL'] !== $interval)
            {
                CAgent::Update($agent['ID'],['AGENT_INTERVAL' => $interval]);
            }
        }
    }
 
    protected function run()
    {
        $consumer = $this->getConsumer();
        $processor = $this->getProcessor();
 
        $startTime = time();
        $this->deescalateMe();
        while ($message = $consumer->receive())
        {
            $result = $processor->process($message);
 
            if ($result === Interfaces\Processor::ACK)
            {
                $consumer->acknowledge($message);
            }
            else if ($result === Interfaces\Processor::REJECT)
            {
                $consumer->reject($message);
            }
 
            $this->escalateMe();
            if ((time() - $startTime) > $this->getTimeLimit())
            {
                break;
            }
        }
    }
 
    protected function escalateMe(): void
    {
        $this->isEscalated = true;
    }
 
    protected function deescalateMe(): void
    {
        $this->isEscalated = false;
    }
 
    protected function getInterval(): int
    {
        return 3600;
    }
 
    protected function getEscalatedInterval(): int
    {
        return 60;
    }
 
    protected function getTimeLimit(): int
    {
        return self::TIME_LIMIT;
    }
 
    abstract protected function getConsumer(): Interfaces\Consumer;
 
    abstract protected function getProcessor(): Interfaces\Processor;
}