<commit_before><?php

/*
 * This file is part of the Symfony package.
 *
 * (c) <NAME> <<EMAIL>>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Symfony\Component\Messenger\Bridge\Redis\Transport;

use Symfony\Component\Messenger\Exception\InvalidArgumentException;
use Symfony\Component\Messenger\Exception\LogicException;
use Symfony\Component\Messenger\Exception\TransportException;

/**
 * A Redis connection.
 *
 * @author <NAME> <<EMAIL>>
 * @author <NAME> <<EMAIL>>
 * @author <NAME> <<EMAIL>>
 *
 * @internal
 * @final
 */
class Connection
{
    private const DEFAULT_OPTIONS = [
        'stream' => 'messages',
        'group' => 'symfony',
        'consumer' => 'consumer',
        'auto_setup' => true,
        'delete_after_ack' => false,
        'delete_after_reject' => true,
        'stream_max_entries' => 0, // any value higher than 0 defines an approximate maximum number of stream entries
        'dbindex' => 0,
        'tls' => false,
        'redeliver_timeout' => 3600, // Timeout before redeliver messages still in pending state (seconds)
        'claim_interval' => 60000, // Interval by which pending/abandoned messages should be checked
        'lazy' => false,
        'auth' => null,
        'serializer' => \Redis::SERIALIZER_PHP,
    ];

    private $connection;
    private $stream;
    private $queue;
    private $group;
    private $consumer;
    private $autoSetup;
    private $maxEntries;
    private $redeliverTimeout;
    private $nextClaim = 0;
    private $claimInterval;
    private $deleteAfterAck;
    private $deleteAfterReject;
    private $couldHavePendingMessages = true;

    /**
     * @param \Redis|\RedisCluster|null $redis
     */
    public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], $redis = null)
    {
        if (version_compare(phpversion('redis'), '4.3.0', '<')) {
            throw new LogicException('The redis transport requires php-redis 4.3.0 or higher.');
        }

        $host = $connectionCredentials['host'] ?? '127.0.0.1';
        $port = $connectionCredentials['port'] ?? 6379;
        $serializer = $redisOptions['serializer'] ?? \Redis::SERIALIZER_PHP;
        $dbIndex = $configuration['dbindex'] ?? self::DEFAULT_OPTIONS['dbindex'];
        $auth = $connectionCredentials['auth'] ?? null;
        if ('' === $auth) {
            $auth = null;
        }

        $lazy = $configuration['lazy'] ?? self::DEFAULT_OPTIONS['lazy'];
        if (\is_array($host) || $redis instanceof \RedisCluster) {
            $hosts = \is_string($host) ? [$host.':'.$port] : $host; // Always ensure we have an array
            $initializer = static function ($redis) use ($hosts, $auth, $serializer) {
                return self::initializeRedisCluster($redis, $hosts, $auth, $serializer);
            };
            $redis = $lazy ? new RedisClusterProxy($redis, $initializer) : $initializer($redis);
        } else {
            $redis = $redis ?? new \Redis();
            $initializer = static function ($redis) use ($host, $port, $auth, $serializer, $dbIndex) {
                return self::initializeRedis($redis, $host, $port, $auth, $serializer, $dbIndex);
            };
            $redis = $lazy ? new RedisProxy($redis, $initializer) : $initializer($redis);
        }

        $this->connection = $redis;

        foreach (['stream', 'group', 'consumer'] as $key) {
            if (isset($configuration[$key]) && '' === $configuration[$key]) {
                throw new InvalidArgumentException(sprintf('"%s" should be configured, got an empty string.', $key));
            }
        }

        $this->stream = $configuration['stream'] ?? self::DEFAULT_OPTIONS['stream'];
        $this->group = $configuration['group'] ?? self::DEFAULT_OPTIONS['group'];
        $this->consumer = $configuration['consumer'] ?? self::DEFAULT_OPTIONS['consumer'];
        $this->queue = $this->stream.'__queue';
        $this->autoSetup = $configuration['auto_setup'] ?? self::DEFAULT_OPTIONS['auto_setup'];
        $this->maxEntries = $configuration['stream_max_entries'] ?? self::DEFAULT_OPTIONS['stream_max_entries'];
        $this->deleteAfterAck = $configuration['delete_after_ack'] ?? self::DEFAULT_OPTIONS['delete_after_ack'];
        $this->deleteAfterReject = $configuration['delete_after_reject'] ?? self::DEFAULT_OPTIONS['delete_after_reject'];
        $this->redeliverTimeout = ($configuration['redeliver_timeout'] ?? self::DEFAULT_OPTIONS['redeliver_timeout']) * 1000;
        $this->claimInterval = $configuration['claim_interval'] ?? self::DEFAULT_OPTIONS['claim_interval'];
    }

    private static function initializeRedis(\Redis $redis, string $host, int $port, ?string $auth, int $serializer, int $dbIndex): \Redis
    {
        $redis->connect($host, $port);
        $redis->setOption(\Redis::OPT_SERIALIZER, $serializer);

        if (null !== $auth && !$redis->auth($auth)) {
            throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError());
        }

        if ($dbIndex && !$redis->select($dbIndex)) {
            throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError());
        }

        return $redis;
    }

    private static function initializeRedisCluster(?\RedisCluster $redis, array $hosts, ?string $auth, int $serializer): \RedisCluster
    {
        if (null === $redis) {
            $redis = new \RedisCluster(null, $hosts, 0.0, 0.0, false, $auth);
        }

        $redis->setOption(\Redis::OPT_SERIALIZER, $serializer);<commit_msg>bug #43174 [Messenger] relax parameter type (xabbuh)<commit_after><?php

/*
 * This file is part of the Symfony package.
 *
 * (c) <NAME> <<EMAIL>>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Symfony\Component\Messenger\Bridge\Redis\Transport;

use Symfony\Component\Messenger\Exception\InvalidArgumentException;
use Symfony\Component\Messenger\Exception\LogicException;
use Symfony\Component\Messenger\Exception\TransportException;

/**
 * A Redis connection.
 *
 * @author <NAME> <<EMAIL>>
 * @author <NAME> <<EMAIL>>
 * @author <NAME> <<EMAIL>>
 *
 * @internal
 * @final
 */
class Connection
{
    private const DEFAULT_OPTIONS = [
        'stream' => 'messages',
        'group' => 'symfony',
        'consumer' => 'consumer',
        'auto_setup' => true,
        'delete_after_ack' => false,
        'delete_after_reject' => true,
        'stream_max_entries' => 0, // any value higher than 0 defines an approximate maximum number of stream entries
        'dbindex' => 0,
        'tls' => false,
        'redeliver_timeout' => 3600, // Timeout before redeliver messages still in pending state (seconds)
        'claim_interval' => 60000, // Interval by which pending/abandoned messages should be checked
        'lazy' => false,
        'auth' => null,
        'serializer' => \Redis::SERIALIZER_PHP,
    ];

    private $connection;
    private $stream;
    private $queue;
    private $group;
    private $consumer;
    private $autoSetup;
    private $maxEntries;
    private $redeliverTimeout;
    private $nextClaim = 0;
    private $claimInterval;
    private $deleteAfterAck;
    private $deleteAfterReject;
    private $couldHavePendingMessages = true;

    /**
     * @param \Redis|\RedisCluster|null $redis
     */
    public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], $redis = null)
    {
        if (version_compare(phpversion('redis'), '4.3.0', '<')) {
            throw new LogicException('The redis transport requires php-redis 4.3.0 or higher.');
        }

        $host = $connectionCredentials['host'] ?? '127.0.0.1';
        $port = $connectionCredentials['port'] ?? 6379;
        $serializer = $redisOptions['serializer'] ?? \Redis::SERIALIZER_PHP;
        $dbIndex = $configuration['dbindex'] ?? self::DEFAULT_OPTIONS['dbindex'];
        $auth = $connectionCredentials['auth'] ?? null;
        if ('' === $auth) {
            $auth = null;
        }

        $lazy = $configuration['lazy'] ?? self::DEFAULT_OPTIONS['lazy'];
        if (\is_array($host) || $redis instanceof \RedisCluster) {
            $hosts = \is_string($host) ? [$host.':'.$port] : $host; // Always ensure we have an array
            $initializer = static function ($redis) use ($hosts, $auth, $serializer) {
                return self::initializeRedisCluster($redis, $hosts, $auth, $serializer);
            };
            $redis = $lazy ? new RedisClusterProxy($redis, $initializer) : $initializer($redis);
        } else {
            $redis = $redis ?? new \Redis();
            $initializer = static function ($redis) use ($host, $port, $auth, $serializer, $dbIndex) {
                return self::initializeRedis($redis, $host, $port, $auth, $serializer, $dbIndex);
            };
            $redis = $lazy ? new RedisProxy($redis, $initializer) : $initializer($redis);
        }

        $this->connection = $redis;

        foreach (['stream', 'group', 'consumer'] as $key) {
            if (isset($configuration[$key]) && '' === $configuration[$key]) {
                throw new InvalidArgumentException(sprintf('"%s" should be configured, got an empty string.', $key));
            }
        }

        $this->stream = $configuration['stream'] ?? self::DEFAULT_OPTIONS['stream'];
        $this->group = $configuration['group'] ?? self::DEFAULT_OPTIONS['group'];
        $this->consumer = $configuration['consumer'] ?? self::DEFAULT_OPTIONS['consumer'];
        $this->queue = $this->stream.'__queue';
        $this->autoSetup = $configuration['auto_setup'] ?? self::DEFAULT_OPTIONS['auto_setup'];
        $this->maxEntries = $configuration['stream_max_entries'] ?? self::DEFAULT_OPTIONS['stream_max_entries'];
        $this->deleteAfterAck = $configuration['delete_after_ack'] ?? self::DEFAULT_OPTIONS['delete_after_ack'];
        $this->deleteAfterReject = $configuration['delete_after_reject'] ?? self::DEFAULT_OPTIONS['delete_after_reject'];
        $this->redeliverTimeout = ($configuration['redeliver_timeout'] ?? self::DEFAULT_OPTIONS['redeliver_timeout']) * 1000;
        $this->claimInterval = $configuration['claim_interval'] ?? self::DEFAULT_OPTIONS['claim_interval'];
    }

    /**
     * @param string|string[]|null $auth
     */
    private static function initializeRedis(\Redis $redis, string $host, int $port, $auth, int $serializer, int $dbIndex): \Redis
    {
        $redis->connect($host, $port);
        $redis->setOption(\Redis::OPT_SERIALIZER, $serializer);

        if (null !== $auth && !$redis->auth($auth)) {
            throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError());
        }

        if ($dbIndex && !$redis->select($dbIndex)) {
            throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError());
        }

        return $redis;
    }

    private static function initializeRedisCluster(?\RedisCluster $redis, array $hosts, ?string $auth, int $serializer): \RedisCluster
    {
        if (null === $redis) {
            $redis = new \RedisCluster(null, $hosts, 0.0, 0.0, false, $auth);
        }

        $redis->setOption(\Redis::OPT_SERIALIZER, $serializer);
{'fraction_non_alphanumeric': 0.15396769662921347, 'fraction_numerical': 0.006934691011235955, 'mean_word_length': 3.145924308588064, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 93, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '2405535', 'n_tokens_mistral': 3418, 'n_tokens_neox': 3309, 'n_words': 751}