<commit_before><?php

namespace Gdbots\Pbjx\Consumer;

use Gdbots\Common\Util\ClassUtils;
use Gdbots\Common\Util\NumberUtils;
use Gdbots\Common\Util\StringUtils;
use Gdbots\Pbj\Message;
use Gdbots\Pbj\Mixin\Command;
use Gdbots\Pbj\Mixin\DomainEvent;
use Gdbots\Pbj\Mixin\Request;
use Gdbots\Pbj\Mixin\Response;
use Gdbots\Pbjx\PbjxEvents;
use Gdbots\Pbjx\ServiceLocator;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

// todo: implement "replay" logic for messages?
abstract class AbstractConsumer
{
    /* @var bool */
    private $isRunning = false;

    /* @var bool */
    private $boundSignals = false;

    /** @var ServiceLocator */
    protected $locator;

    /** @var LoggerInterface */
    protected $logger;

    /* @var string */
    protected $consumerName;

    /**
     * @param ServiceLocator $locator
     * @param LoggerInterface $logger
     */
    public function __construct(ServiceLocator $locator, LoggerInterface $logger = null)
    {
        $this->locator = $locator;
        $this->logger = $logger ?: new NullLogger();
        $this->consumerName = StringUtils::toSnakeFromCamel(
            str_replace('Consumer', '', ClassUtils::getShortName($this))
        );
    }

    /**
     * Runs the consumer until a signal is caught or the max runtime is reached.
     * @param int $maxRuntime
     */
    final public function run($maxRuntime = 300)
    {
        $maxRuntime = NumberUtils::bound($maxRuntime, 10, 86400);
        $start = time();
        $this->logger->notice(
            sprintf('Starting [%s] consumer, will run for up to [%d] seconds.', $this->consumerName, $maxRuntime)
        );
        $this->setup();

        $this->isRunning = true;
        $this->bindSignals();
        $dispatcher = $this->locator->getDispatcher();

        do {
            try {
                $this->work();
            } catch(\Exception $e) {
                $this->stop();
                $this->logger->critical(
                    sprintf(
                        'Consumer [%s] caught an exception with message [%s], shutting down.',
                        $this->consumerName,
                        $e->getMessage()
                    )
                );
                $dispatcher->dispatch(PbjxEvents::CONSUMER_HANDLING_EXCEPTION);
                break;
            }

            if ($this->isRunning && $maxRuntime > 0 && time() - $start > $maxRuntime) {
                $this->logger->notice(
                    sprintf(
                        'Consumer [%s] has been running for more than [%d] seconds, shutting down.',
                        $this->consumerName,
                        $maxRuntime
                    )
                );
                $this->stop();
            }
        } while ($this->isRunning());

        $this->teardown();
        $dispatcher->dispatch(PbjxEvents::CONSUMER_AFTER_TEARDOWN);
        $this->stop();
    }

    /**
     * Stops the consumer.  This allows it to finish its current work
     * and perform a clean shutdown if possible.
     */
    final public function stop()
    {
        $this->isRunning = false;
    }

    /**
     * @return bool
     */
    final public function isRunning()
    {
        return $this->isRunning;
    }

    /**
     * This runs prior to the run loop starting.
     */
    protected function setup()
    {
        // override for custom setup.
    }

    /**
     * Override in the concrete consumer to perform the actual work.
     * This is called repeatly in a while loop when the consumer is running.
     */
    protected function work()
    {
        // concrete consumer should do something.
    }

    /**
     * This runs after to the run loop stops.
     */
    protected function teardown()
    {
        // override for custom teardown.
    }

    /**
     * @param Message $message
     * @return Message|null
     */
    final protected function handleMessage(Message $message)
    {
        if ($message instanceof Command) {
            $this->handleCommand($message);
            return null;
        }

        if ($message instanceof DomainEvent) {
            $this->handleEvent($message);
            return null;
        }

        if ($message instanceof Request) {
            return $this->handleRequest($message);
        }
    }

    /**
     * @param Command $command
     */
    final protected function handleCommand(Command $command)
    {
        $this->locator->getCommandBus()->receiveCommand($command);
    }

    /**
     * @param DomainEvent $domainEvent
     */
    final protected function handleEvent(DomainEvent $domainEvent)
    {
        $this->locator->getEventBus()->receiveEvent($domainEvent);
    }

    /**
     * @param Request $request
     * @return Response
     */
    final protected function handleRequest(Request $request)
    {
        return $this->locator->getRequestBus()->receiveRequest($request);
    }<commit_msg>handle command, event, etc. methods don't need to be public<commit_after><?php

namespace Gdbots\Pbjx\Consumer;

use Gdbots\Common\Util\ClassUtils;
use Gdbots\Common\Util\NumberUtils;
use Gdbots\Common\Util\StringUtils;
use Gdbots\Pbj\Message;
use Gdbots\Pbj\Mixin\Command;
use Gdbots\Pbj\Mixin\DomainEvent;
use Gdbots\Pbj\Mixin\Request;
use Gdbots\Pbj\Mixin\Response;
use Gdbots\Pbjx\PbjxEvents;
use Gdbots\Pbjx\ServiceLocator;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

// todo: implement "replay" logic for messages?
abstract class AbstractConsumer
{
    /* @var bool */
    private $isRunning = false;

    /* @var bool */
    private $boundSignals = false;

    /** @var ServiceLocator */
    protected $locator;

    /** @var LoggerInterface */
    protected $logger;

    /* @var string */
    protected $consumerName;

    /**
     * @param ServiceLocator $locator
     * @param LoggerInterface $logger
     */
    public function __construct(ServiceLocator $locator, LoggerInterface $logger = null)
    {
        $this->locator = $locator;
        $this->logger = $logger ?: new NullLogger();
        $this->consumerName = StringUtils::toSnakeFromCamel(
            str_replace('Consumer', '', ClassUtils::getShortName($this))
        );
    }

    /**
     * Runs the consumer until a signal is caught or the max runtime is reached.
     * @param int $maxRuntime
     */
    final public function run($maxRuntime = 300)
    {
        $maxRuntime = NumberUtils::bound($maxRuntime, 10, 86400);
        $start = time();
        $this->logger->notice(
            sprintf('Starting [%s] consumer, will run for up to [%d] seconds.', $this->consumerName, $maxRuntime)
        );
        $this->setup();

        $this->isRunning = true;
        $this->bindSignals();
        $dispatcher = $this->locator->getDispatcher();

        do {
            try {
                $this->work();
            } catch(\Exception $e) {
                $this->stop();
                $this->logger->critical(
                    sprintf(
                        'Consumer [%s] caught an exception with message [%s], shutting down.',
                        $this->consumerName,
                        $e->getMessage()
                    )
                );
                $dispatcher->dispatch(PbjxEvents::CONSUMER_HANDLING_EXCEPTION);
                break;
            }

            if ($this->isRunning && $maxRuntime > 0 && time() - $start > $maxRuntime) {
                $this->stop();
                $this->logger->notice(
                    sprintf(
                        'Consumer [%s] has been running for more than [%d] seconds, shutting down.',
                        $this->consumerName,
                        $maxRuntime
                    )
                );
                break;
            }
        } while ($this->isRunning());

        $this->teardown();
        $dispatcher->dispatch(PbjxEvents::CONSUMER_AFTER_TEARDOWN);
        $this->stop();
    }

    /**
     * Stops the consumer.  This allows it to finish its current work
     * and perform a clean shutdown if possible.
     */
    final public function stop()
    {
        $this->isRunning = false;
    }

    /**
     * @return bool
     */
    final public function isRunning()
    {
        return $this->isRunning;
    }

    /**
     * This runs prior to the run loop starting.
     */
    protected function setup()
    {
        // override for custom setup.
    }

    /**
     * Override in the concrete consumer to perform the actual work.
     * This is called repeatly in a while loop when the consumer is running.
     */
    protected function work()
    {
        // concrete consumer should do something.
    }

    /**
     * This runs after to the run loop stops.
     */
    protected function teardown()
    {
        // override for custom teardown.
    }

    /**
     * @param Message $message
     * @return Message|null
     */
    final protected function handleMessage(Message $message)
    {
        if ($message instanceof Command) {
            $this->handleCommand($message);
            return null;
        }

        if ($message instanceof DomainEvent) {
            $this->handleEvent($message);
            return null;
        }

        if ($message instanceof Request) {
            return $this->handleRequest($message);
        }
    }

    /**
     * @param Command $command
     */
    private function handleCommand(Command $command)
    {
        $this->locator->getCommandBus()->receiveCommand($command);
    }

    /**
     * @param DomainEvent $domainEvent
     */
    private function handleEvent(DomainEvent $domainEvent)
    {
        $this->locator->getEventBus()->receiveEvent($domainEvent);
    }

    /**
     * @param Request $request
     * @return Response
     */
    private function handleRequest(Request $request)
    {
        return $this->locator->getRequestBus()->receiveRequest($request);
    }
{'fraction_non_alphanumeric': 0.11979746835443038, 'fraction_numerical': 0.0022278481012658227, 'mean_word_length': 1.8568122649696268, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 89, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3216881', 'n_tokens_mistral': 2829, 'n_tokens_neox': 2677, 'n_words': 671}