<commit_before><?php

namespace AVAllAC\RabbitComponent;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AbstractConnection;

class Consumer
{
    protected $app;
    protected $receivedBreak;
    protected $socket;
    /** @var AMQPChannel $channel */
    protected $channel;

    public function __construct($app)
    {
        $this->app = $app;
    }

    protected function initSignalHandlers()
    {
        pcntl_signal(SIGTERM, function () {
            $this->receivedBreak = 1;
        });
        pcntl_signal(SIGINT, function () {
            $this->receivedBreak = 1;
        });
    }

    protected function initConsume($qName, MessageManager $manager)
    {
        $fnCallback = function ($rabbitMessage) use ($manager) {
            $code = $manager->handle($rabbitMessage);
            if ($code !== MessageManager::NO_ACK_MESSAGE) {
                $this->channel->basic_ack($rabbitMessage->delivery_info['delivery_tag']);
            }
        };
        $this->channel->basic_consume($qName, '', false, false, false, false, $fnCallback);

    }

    protected function checkExternalServices()
    {
        if ($this->app['amqp'] instanceof AbstractConnection) {
            $this->socket = $this->app['amqp']->getSocket();
        } else {
            throw new \Exception('require AbstractConnection on "amqp"');
        }
        if ($this->app['rabbitChannel'] instanceof AMQPChannel) {
            $this->channel = $this->app['rabbitChannel'];
        } else {
            throw new \Exception('require AMQPChannel on "rabbitChannel"');
        }
    }

    public function run($qName, $manager)
    {
        $this->initSignalHandlers();
        $this->checkExternalServices();
        $this->initConsume($qName, $manager);
        while (count($this->channel->callbacks) && !$this->receivedBreak) {
            $read = [$this->socket];
            $write = null;
            $except = null;
            $changeStreamsCount = stream_select($read, $write, $except, 60);
            pcntl_signal_dispatch();
            if (!$this->receivedBreak) {
                if ($changeStreamsCount === false) {
                    throw new \RuntimeException();
                } elseif ($changeStreamsCount > 0) {
                    $this->channel->wait();
                }
            }
        }
    }
}<commit_msg>lost pcntl_signal_dispatch<commit_after><?php

namespace AVAllAC\RabbitComponent;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AbstractConnection;

class Consumer
{
    protected $app;
    protected $receivedBreak;
    protected $socket;
    /** @var AMQPChannel $channel */
    protected $channel;

    public function __construct($app)
    {
        $this->app = $app;
    }

    protected function initSignalHandlers()
    {
        pcntl_signal(SIGTERM, function () {
            $this->receivedBreak = 1;
        });
        pcntl_signal(SIGINT, function () {
            $this->receivedBreak = 1;
        });
    }

    protected function initConsume($qName, MessageManager $manager)
    {
        $fnCallback = function ($rabbitMessage) use ($manager) {
            $code = $manager->handle($rabbitMessage);
            if ($code !== MessageManager::NO_ACK_MESSAGE) {
                $this->channel->basic_ack($rabbitMessage->delivery_info['delivery_tag']);
            }
        };
        $this->channel->basic_consume($qName, '', false, false, false, false, $fnCallback);

    }

    protected function checkExternalServices()
    {
        if ($this->app['amqp'] instanceof AbstractConnection) {
            $this->socket = $this->app['amqp']->getSocket();
        } else {
            throw new \Exception('require AbstractConnection on "amqp"');
        }
        if ($this->app['rabbitChannel'] instanceof AMQPChannel) {
            $this->channel = $this->app['rabbitChannel'];
        } else {
            throw new \Exception('require AMQPChannel on "rabbitChannel"');
        }
    }

    public function run($qName, $manager)
    {
        $this->initSignalHandlers();
        $this->checkExternalServices();
        $this->initConsume($qName, $manager);
        while (count($this->channel->callbacks) && !$this->receivedBreak) {
            $read = [$this->socket];
            $write = null;
            $except = null;
            $changeStreamsCount = stream_select($read, $write, $except, 60);
            pcntl_signal_dispatch();
            if (!$this->receivedBreak) {
                if ($changeStreamsCount === false) {
                    throw new \RuntimeException();
                } elseif ($changeStreamsCount > 0) {
                    $this->channel->wait();
                }
            }
            pcntl_signal_dispatch();
        }
    }
}
{'fraction_non_alphanumeric': 0.13609591922591502, 'fraction_numerical': 0.002103491796381994, 'mean_word_length': 2.1386138613861387, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 57, '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': '2922342', 'n_tokens_mistral': 1355, 'n_tokens_neox': 1288, 'n_words': 246}