<commit_before><?php
namespace Wisembly\AmqpBundle\Broker;

use InvalidArgumentException;

use AMQPQueue;
use AMQPChannel;
use AMQPExchange;
use AMQPConnection;

use AMQPException;
use AMQPChannelException;
use AMQPExchangeException;
use AMQPConnectionException;

use Swarrot\Broker\MessageProvider\PeclPackageMessageProvider;
use Swarrot\Broker\MessagePublisher\PeclPackageMessagePublisher;

use Wisembly\AmqpBundle\Gate;
use Wisembly\AmqpBundle\Connection;
use Wisembly\AmqpBundle\BrokerInterface;
use Wisembly\AmqpBundle\Exception\MessagingException;

class PeclBroker implements BrokerInterface
{
    const TYPES = [
        'direct' => AMQP_EX_TYPE_DIRECT,
        'fanout' => AMQP_EX_TYPE_FANOUT,
        'headers' => AMQP_EX_TYPE_HEADERS,
        'topic' => AMQP_EX_TYPE_TOPIC,
    ];

    /** @var AMQPChannel[] */
    private $channels = [];

    /** @var PeclPackageMessageProvider[][] */
    private $providers = [];

    /** @var PeclPackageMessagePublisher[][] */
    private $producers = [];

    /** {@inheritDoc} */
    public function getProvider(Gate $gate)
    {
        $name = $gate->getName();
        $connection = $gate->getConnection()->getName();

        if (!isset($this->providers[$connection])) {
            $this->providers[$connection] = [];
        }

        if (!isset($this->providers[$connection][$name])) {
            $channel = $this->getChannel($gate->getConnection());
            $this->declare($gate, $channel);
            $queue = new AMQPQueue($channel);
            $queue->setName($gate->getQueue());

            $this->providers[$connection][$name] = new PeclPackageMessageProvider($queue);
        }

        return $this->providers[$connection][$name];
    }

    /** {@inheritDoc} */
    public function getProducer(Gate $gate)
    {
        $name = $gate->getName();
        $connection = $gate->getConnection()->getName();

        if (!isset($this->producers[$connection])) {
            $this->producers[$connection] = [];
        }

        if (!isset($this->producers[$connection][$name])) {
            try {
                $channel = $this->getChannel($gate->getConnection());
                $this->declare($gate, $channel);
                $exchange = new AMQPExchange($channel);
                $exchange->setName($gate->getExchange());
            } catch (AMQPExchangeException $e) {
                throw new MessagingException($e);
            }

            $this->producers[$connection][$name] = new PeclPackageMessagePublisher($exchange);
        }

        return $this->producers[$connection][$name];
    }

    private function declare(Gate $gate, AMQPChannel $channel)
    {
        if (false === $gate->getAutoDeclare()) {
            return;
        }

        $options = $gate->getQueueOptions();
        $queue = new AMQPQueue($channel);
        $queue->setName($gate->getQueue());

        $flags = 0;
        if ($options['durable'] ?? false) {
            $flags |= \AMQP_DURABLE;
        }
        if ($options['passive'] ?? false) {
            $flags |= \AMQP_PASSIVE;
        }
        if ($options['exclusive'] ?? false) {
            $flags |= \AMQP_EXCLUSIVE;
        }
        if ($options['auto-delete'] ?? false) {
            $flags |= \AMQP_AUTODELETE;
        }

        $queue->setFlags($flags);
        $queue->setArguments($options['arguments'] ?? []);
        $queue->declareQueue();

        $options = $gate->getExchangeOptions();
        $exchange = new AMQPExchange($channel);
        $exchange->setName($gate->getExchange());

        $flags = 0;
        if ($options['durable'] ?? false) {
            $flags &= \AMQP_DURABLE;
        }
        if ($options['passive'] ?? false) {
            $flags &= \AMQP_PASSIVE;
        }

        $exchange->setFlags($flags);
        $exchange->setArguments($options['arguments'] ?? []);
        $exchange->setType(self::TYPES[$options['type'] ?? 'direct'] ?? 'direct');

        $queue->bind(
            $gate->getExchange(),
            $queue->getRoutingKey() ?: '',
            [] // arguments
        );
    }<commit_msg>fix bad method call<commit_after><?php
namespace Wisembly\AmqpBundle\Broker;

use InvalidArgumentException;

use AMQPQueue;
use AMQPChannel;
use AMQPExchange;
use AMQPConnection;

use AMQPException;
use AMQPChannelException;
use AMQPExchangeException;
use AMQPConnectionException;

use Swarrot\Broker\MessageProvider\PeclPackageMessageProvider;
use Swarrot\Broker\MessagePublisher\PeclPackageMessagePublisher;

use Wisembly\AmqpBundle\Gate;
use Wisembly\AmqpBundle\Connection;
use Wisembly\AmqpBundle\BrokerInterface;
use Wisembly\AmqpBundle\Exception\MessagingException;

class PeclBroker implements BrokerInterface
{
    const TYPES = [
        'direct' => AMQP_EX_TYPE_DIRECT,
        'fanout' => AMQP_EX_TYPE_FANOUT,
        'headers' => AMQP_EX_TYPE_HEADERS,
        'topic' => AMQP_EX_TYPE_TOPIC,
    ];

    /** @var AMQPChannel[] */
    private $channels = [];

    /** @var PeclPackageMessageProvider[][] */
    private $providers = [];

    /** @var PeclPackageMessagePublisher[][] */
    private $producers = [];

    /** {@inheritDoc} */
    public function getProvider(Gate $gate)
    {
        $name = $gate->getName();
        $connection = $gate->getConnection()->getName();

        if (!isset($this->providers[$connection])) {
            $this->providers[$connection] = [];
        }

        if (!isset($this->providers[$connection][$name])) {
            $channel = $this->getChannel($gate->getConnection());
            $this->declare($gate, $channel);
            $queue = new AMQPQueue($channel);
            $queue->setName($gate->getQueue());

            $this->providers[$connection][$name] = new PeclPackageMessageProvider($queue);
        }

        return $this->providers[$connection][$name];
    }

    /** {@inheritDoc} */
    public function getProducer(Gate $gate)
    {
        $name = $gate->getName();
        $connection = $gate->getConnection()->getName();

        if (!isset($this->producers[$connection])) {
            $this->producers[$connection] = [];
        }

        if (!isset($this->producers[$connection][$name])) {
            try {
                $channel = $this->getChannel($gate->getConnection());
                $this->declare($gate, $channel);
                $exchange = new AMQPExchange($channel);
                $exchange->setName($gate->getExchange());
            } catch (AMQPExchangeException $e) {
                throw new MessagingException($e);
            }

            $this->producers[$connection][$name] = new PeclPackageMessagePublisher($exchange);
        }

        return $this->producers[$connection][$name];
    }

    private function declare(Gate $gate, AMQPChannel $channel)
    {
        if (false === $gate->getAutoDeclare()) {
            return;
        }

        $options = $gate->getQueueOptions();
        $queue = new AMQPQueue($channel);
        $queue->setName($gate->getQueue());

        $flags = 0;
        if ($options['durable'] ?? false) {
            $flags |= \AMQP_DURABLE;
        }
        if ($options['passive'] ?? false) {
            $flags |= \AMQP_PASSIVE;
        }
        if ($options['exclusive'] ?? false) {
            $flags |= \AMQP_EXCLUSIVE;
        }
        if ($options['auto-delete'] ?? false) {
            $flags |= \AMQP_AUTODELETE;
        }

        $queue->setFlags($flags);
        $queue->setArguments($options['arguments'] ?? []);
        $queue->declareQueue();

        $options = $gate->getExchangeOptions();
        $exchange = new AMQPExchange($channel);
        $exchange->setName($gate->getExchange());

        $flags = 0;
        if ($options['durable'] ?? false) {
            $flags &= \AMQP_DURABLE;
        }
        if ($options['passive'] ?? false) {
            $flags &= \AMQP_PASSIVE;
        }

        $exchange->setFlags($flags);
        $exchange->setArguments($options['arguments'] ?? []);
        $exchange->setType(self::TYPES[$options['type'] ?? 'direct'] ?? 'direct');

        $queue->bind(
            $gate->getExchange(),
            $gate->getRoutingKey() ?: '',
            [] // arguments
        );
    }
{'fraction_non_alphanumeric': 0.16161863887185776, 'fraction_numerical': 0.0004904966278356836, 'mean_word_length': 2.438448566610455, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 95, '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': '2455812', 'n_tokens_mistral': 2478, 'n_tokens_neox': 2306, 'n_words': 379}