<commit_before><?php

namespace Itkg\DelayEventBundle\Command;

use Itkg\DelayEventBundle\Handler\LockHandlerInterface;
use Itkg\DelayEventBundle\Model\Event;
use Itkg\DelayEventBundle\Model\Lock;
use Itkg\DelayEventBundle\Processor\EventProcessor;
use Itkg\DelayEventBundle\Repository\EventRepository;
use Itkg\DelayEventBundle\Repository\LockRepository;
use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;

/**
 * Class ProcessDynamicChannelCommand
 */
class ProcessDynamicChannelCommand extends ContainerAwareCommand
{
    /**
     * @var EventRepository
     */
    private $eventRepository;

    /**
     * @var EventProcessor
     */
    private $eventProcessor;

    /**
     * @var LockHandlerInterface
     */
    private $lockHandler;

    /**
     * @var array
     */
    private $channels;

    /**
     * @var LockRepository
     */
    private $lockRepository;

    /**
     * ProcessEventCommand constructor.
     *
     * @param EventRepository      $eventRepository
     * @param LockRepository       $lockRepository
     * @param array                $channels
     * @param null|string          $name
     */
    public function __construct(
        EventRepository $eventRepository,
        LockRepository $lockRepository,
        array $channels = [],
        $name = null
    ) {
        $this->eventRepository = $eventRepository;
        $this->lockRepository = $lockRepository;
        $this->channels = $channels;

        parent::__construct($name);
    }

    /**
     * {@inheritdoc}
     */
    protected function configure()
    {
        $this
            ->setName('itkg_delay_event:process_dynamic_channel')
            ->setDescription('Process dynamic channel')
            ->addArgument(
                'channel',
                InputArgument::REQUIRED,
                'Dynamic channel to process'
            )
            ->addOption(
                'concurrent-jobs-count',
                'c',
                InputOption::VALUE_OPTIONAL,
                'Maximum concurrent jobs count for this channel. Default 5',
                5
            );
    }

    /**
     * {@inheritdoc}
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $channel = $input->getArgument('channel');

        $commandline = 'php console --env=%s itkg_delay_event:process -c %s -g %s';

        if (!isset($this->channels[$channel])) {
            $output->writeln(
                sprintf(
                    '<error>Channel <info>%s</info> is not configured.</error>',
                    $channel
                )
            );
            return;
        }

        if (!$this->channels[$channel]['dynamic']) {
            $output->writeln(
                sprintf(
                    '<error>Channel <info>%s</info> is not dynamic.</error>',
                    $channel
                )
            );
            return;
        }

        $fieldGroupIdentifierList = $this->eventRepository->findDistinctFieldGroupIdentifierByChannel(
            $this->channels[$channel]['include'],
            $this->channels[$channel]['exclude']
        )->toArray();

        $locks = $this->lockRepository->findAll();
        $lockNames = array();
        $currentMachineLockNames = array();
        /** @var Lock $lock */
        foreach ($locks as $lock) {
            if ($lock->isCommandLocked()) {
                $lockName = str_replace(sprintf('%s_', $channel), '', $lock->getChannel());
                $lockNames[] = $lockName;
                if ($lock->getLockedBy() === $this->getHostName()) {
                    $currentMachineLockNames[] = $lockName;
                }
            }
        }

        $concurrentAvailableSlotsCount = $this->calculateAvailableSlotsCount(
            $fieldGroupIdentifierList,
            $currentMachineLockNames,
            $input->getOption('concurrent-jobs-count')
        );

        if ($concurrentAvailableSlotsCount <= 0) {
            $output->writeln(
                sprintf(
                    '<info>Maximum concurrent jobs limit for %s is reached.</info>',
                    $channel
                )
            );
        }

        $groupFieldIdentifierListToProcess = array_slice(
            array_diff(
                $fieldGroupIdentifierList,
                $lockNames<commit_msg>Fix available slots count calcul<commit_after><?php

namespace Itkg\DelayEventBundle\Command;

use Itkg\DelayEventBundle\Handler\LockHandlerInterface;
use Itkg\DelayEventBundle\Model\Event;
use Itkg\DelayEventBundle\Model\Lock;
use Itkg\DelayEventBundle\Processor\EventProcessor;
use Itkg\DelayEventBundle\Repository\EventRepository;
use Itkg\DelayEventBundle\Repository\LockRepository;
use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;

/**
 * Class ProcessDynamicChannelCommand
 */
class ProcessDynamicChannelCommand extends ContainerAwareCommand
{
    /**
     * @var EventRepository
     */
    private $eventRepository;

    /**
     * @var EventProcessor
     */
    private $eventProcessor;

    /**
     * @var LockHandlerInterface
     */
    private $lockHandler;

    /**
     * @var array
     */
    private $channels;

    /**
     * @var LockRepository
     */
    private $lockRepository;

    /**
     * ProcessEventCommand constructor.
     *
     * @param EventRepository      $eventRepository
     * @param LockRepository       $lockRepository
     * @param array                $channels
     * @param null|string          $name
     */
    public function __construct(
        EventRepository $eventRepository,
        LockRepository $lockRepository,
        array $channels = [],
        $name = null
    ) {
        $this->eventRepository = $eventRepository;
        $this->lockRepository = $lockRepository;
        $this->channels = $channels;

        parent::__construct($name);
    }

    /**
     * {@inheritdoc}
     */
    protected function configure()
    {
        $this
            ->setName('itkg_delay_event:process_dynamic_channel')
            ->setDescription('Process dynamic channel')
            ->addArgument(
                'channel',
                InputArgument::REQUIRED,
                'Dynamic channel to process'
            )
            ->addOption(
                'concurrent-jobs-count',
                'c',
                InputOption::VALUE_OPTIONAL,
                'Maximum concurrent jobs count for this channel. Default 5',
                5
            );
    }

    /**
     * {@inheritdoc}
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $channel = $input->getArgument('channel');

        $commandline = 'php console --env=%s itkg_delay_event:process -c %s -g %s';

        if (!isset($this->channels[$channel])) {
            $output->writeln(
                sprintf(
                    '<error>Channel <info>%s</info> is not configured.</error>',
                    $channel
                )
            );
            return;
        }

        if (!$this->channels[$channel]['dynamic']) {
            $output->writeln(
                sprintf(
                    '<error>Channel <info>%s</info> is not dynamic.</error>',
                    $channel
                )
            );
            return;
        }

        $fieldGroupIdentifierList = $this->eventRepository->findDistinctFieldGroupIdentifierByChannel(
            $this->channels[$channel]['include'],
            $this->channels[$channel]['exclude']
        )->toArray();

        $locks = $this->lockRepository->findAll();
        $lockNames = array();
        $currentMachineLockNames = array();
        /** @var Lock $lock */
        foreach ($locks as $lock) {
            $lockName = str_replace(sprintf('%s_', $channel), '', $lock->getChannel());
            if ($lock->isCommandLocked() && in_array($lockName, $fieldGroupIdentifierList)) {
                $lockNames[] = $lockName;
                if ($lock->getLockedBy() === $this->getHostName()) {
                    $currentMachineLockNames[] = $lockName;
                }
            }
        }

        $concurrentAvailableSlotsCount = $this->calculateAvailableSlotsCount(
            $fieldGroupIdentifierList,
            $currentMachineLockNames,
            $input->getOption('concurrent-jobs-count')
        );

        if ($concurrentAvailableSlotsCount <= 0) {
            $output->writeln(
                sprintf(
                    '<info>Maximum concurrent jobs limit for %s is reached.</info>',
                    $channel
                )
            );
        }

        $groupFieldIdentifierListToProcess = array_slice(
            array_diff(
                $fieldGroupIdentifierList,
                $lockNames
{'fraction_non_alphanumeric': 0.11547268563951586, 'fraction_numerical': 0.0006542361792607131, 'mean_word_length': 1.8879093198992443, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 75, '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': '4655793', 'n_tokens_mistral': 2354, 'n_tokens_neox': 2280, 'n_words': 421}