<commit_before><?php

namespace DelayedJobs\Broker\Driver;

use Cake\Core\InstanceConfigTrait;
use Cake\Core\Retry\CommandRetry;
use DelayedJobs\Broker\Driver\Retry\PhpAmqpLibReconnectStrategy;
use DelayedJobs\DelayedJob\Job;
use DelayedJobs\DelayedJob\ManagerInterface;
use DelayedJobs\DelayedJob\MessageBrokerInterface;
use DelayedJobs\Exception\BrokerReconnectionException;
use DelayedJobs\Traits\DebugLoggerTrait;
use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Connection\AMQPLazySocketConnection;
use PhpAmqpLib\Connection\AMQPSocketConnection;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exception\AMQPIOException;
use PhpAmqpLib\Exception\AMQPIOWaitException;
use PhpAmqpLib\Exception\AMQPTimeoutException;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

/**
 * Class PhpAmqpLibDriver
 */
class PhpAmqpLibDriver implements RabbitMqDriverInterface
{
    use DebugLoggerTrait;
    use InstanceConfigTrait;

    /**
     *
     */
    const TIMEOUT = 10; //In seconds

    /**
     * @var \PhpAmqpLib\Connection\AbstractConnection
     */
    protected $_connection = null;

    /**
     * @var \PhpAmqpLib\Channel\AMQPChannel
     */
    protected $_channel = null;

    /**
     * @var array
     */
    protected $_defaultConfig = [];

    /**
     * @var \DelayedJobs\DelayedJob\ManagerInterface
     */
    protected $_manager;

    protected $_consumeCallback;
    protected $_hearbeatCallback;

    /**
     * @param array $config
     * @param \DelayedJobs\DelayedJob\ManagerInterface $manager
     * @param \PhpAmqpLib\Connection\AbstractConnection|null $connection
     */
    public function __construct(array $config = [], ManagerInterface $manager, AbstractConnection $connection = null)
    {
        $this->setConfig($config);

        if ($connection) {
            $this->_connection = $connection;
        }

        $this->_manager = $manager;
    }

    public function __destroy()
    {
        if ($this->_channel) {
            $this->_channel->close();
        }
        if ($this->_connection && $this->_connection->isConnected()) {
            $this->_connection->close();
        }
    }

    /**
     * @return \PhpAmqpLib\Connection\AbstractConnection
     */
    public function getConnection()
    {
        if ($this->_connection && $this->_connection->isConnected()) {
            return $this->_connection;
        }

        $config = $this->getConfig();
        $this->_connection = new AMQPStreamConnection(
            $config['host'],
            $config['port'],
            $config['user'],
            $config['pass'],
            $config['path'],
            false,
            'AMQPLAIN',
            null,
            'en_US',
            self::TIMEOUT
        );

        return $this->_connection;
    }

    /**
     * @return void
     */
    public function clearChannel()
    {
        $this->_channel = null;
    }

    /**
     * @return \PhpAmqpLib\Channel\AMQPChannel
     */
    public function getChannel()
    {
        if ($this->_channel) {
            return $this->_channel;
        }<commit_msg>Increase heartbeat interval to 5 minutes (#37)<commit_after><?php

namespace DelayedJobs\Broker\Driver;

use Cake\Core\InstanceConfigTrait;
use Cake\Core\Retry\CommandRetry;
use DelayedJobs\Broker\Driver\Retry\PhpAmqpLibReconnectStrategy;
use DelayedJobs\DelayedJob\Job;
use DelayedJobs\DelayedJob\ManagerInterface;
use DelayedJobs\DelayedJob\MessageBrokerInterface;
use DelayedJobs\Exception\BrokerReconnectionException;
use DelayedJobs\Traits\DebugLoggerTrait;
use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Connection\AMQPLazySocketConnection;
use PhpAmqpLib\Connection\AMQPSocketConnection;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exception\AMQPIOException;
use PhpAmqpLib\Exception\AMQPIOWaitException;
use PhpAmqpLib\Exception\AMQPTimeoutException;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

/**
 * Class PhpAmqpLibDriver
 */
class PhpAmqpLibDriver implements RabbitMqDriverInterface
{
    use DebugLoggerTrait;
    use InstanceConfigTrait;

    /**
     * Connection timeout in seconds
     */
    const CONNECTION_TIMEOUT = 10;
    /**
     * Read/write timeout (in seconds)
     *
     * Must be at least double heartbeat
     */
    const READ_WRITE_TIMEOUT = 620;
    /**
     * Heartbeat in seconds
     */
    const HEARTBEAT = 300;

    /**
     * @var \PhpAmqpLib\Connection\AbstractConnection
     */
    protected $_connection = null;

    /**
     * @var \PhpAmqpLib\Channel\AMQPChannel
     */
    protected $_channel = null;

    /**
     * @var array
     */
    protected $_defaultConfig = [];

    /**
     * @var \DelayedJobs\DelayedJob\ManagerInterface
     */
    protected $_manager;

    protected $_consumeCallback;
    protected $_hearbeatCallback;

    /**
     * @param array $config
     * @param \DelayedJobs\DelayedJob\ManagerInterface $manager
     * @param \PhpAmqpLib\Connection\AbstractConnection|null $connection
     */
    public function __construct(array $config = [], ManagerInterface $manager, AbstractConnection $connection = null)
    {
        $this->setConfig($config);

        if ($connection) {
            $this->_connection = $connection;
        }

        $this->_manager = $manager;
    }

    public function __destroy()
    {
        if ($this->_channel) {
            $this->_channel->close();
        }
        if ($this->_connection && $this->_connection->isConnected()) {
            $this->_connection->close();
        }
    }

    /**
     * @return \PhpAmqpLib\Connection\AbstractConnection
     */
    public function getConnection()
    {
        if ($this->_connection && $this->_connection->isConnected()) {
            return $this->_connection;
        }

        $config = $this->getConfig();
        $this->_connection = new AMQPStreamConnection(
            $config['host'],
            $config['port'],
            $config['user'],
            $config['pass'],
            $config['path'],
            false,
            'AMQPLAIN',
            null,
            'en_US',
            self::TIMEOUT,
            self::READ_WRITE_TIMEOUT,
            null,
            true,
            self::HEARTBEAT
        );

        return $this->_connection;
    }

    /**
     * @return void
     */
    public function clearChannel()
    {
        $this->_channel = null;
    }

    /**
     * @return \PhpAmqpLib\Channel\AMQPChannel
     */
    public function getChannel()
    {
        if ($this->_channel) {
            return $this->_channel;
        }
{'fraction_non_alphanumeric': 0.11629315566323441, 'fraction_numerical': 0.001968503937007874, 'mean_word_length': 2.456305599162742, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 45, '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': '396380', 'n_tokens_mistral': 1996, 'n_tokens_neox': 1925, 'n_words': 335}