<commit_before><?php

namespace Enqueue\Monitoring;

use Enqueue\Client\Config;
use Enqueue\Dsn\Dsn;
use InfluxDB\Client;
use InfluxDB\Database;
use InfluxDB\Driver\QueryDriverInterface;
use InfluxDB\Exception as InfluxDBException;
use InfluxDB\Point;

class InfluxDbStorage implements StatsStorage
{
    /**
     * @var array
     */
    private $config;

    /**
     * @var Client
     */
    private $client;

    /**
     * @var Database
     */
    private $database;

    /**
     * The config could be an array, string DSN, null or instance of InfluxDB\Client. In case of null it will attempt to connect to localhost.
     *
     * $config = [
     *   'dsn' => 'influxdb://127.0.0.1:8086',
     *   'host' => '127.0.0.1',
     *   'port' => '8086',
     *   'user' => '',
     *   'password' => '',
     *   'db' => 'enqueue',
     *   'measurementSentMessages' => 'sent-messages',
     *   'measurementConsumedMessages' => 'consumed-messages',
     *   'measurementConsumers' => 'consumers',
     *   'client' => null, # Client instance. Null by default.
     *   'retentionPolicy' => null,
     * ]
     *
     * or
     *
     * influxdb://127.0.0.1:8086?user=Jon&password=<PASSWORD>
     *
     * @param array|string|null $config
     */
    public function __construct($config = 'influxdb:')
    {
        if (false == class_exists(Client::class)) {
            throw new \LogicException('Seems client library is not installed. Please install "influxdb/influxdb-php"');
        }

        if (empty($config)) {
            $config = [];
        } elseif (is_string($config)) {
            $config = self::parseDsn($config);
        } elseif (is_array($config)) {
            $config = empty($config['dsn']) ? $config : self::parseDsn($config['dsn']);
        } elseif ($config instanceof Client) {
            // Passing Client instead of array config is deprecated because it prevents setting any configuration values
            // and causes library to use defaults.
            @trigger_error(
                sprintf('Passing %s as %s argument is deprecated. Pass it as "client" array property or use createWithClient instead',
                Client::class,
                __METHOD__
            ), E_USER_DEPRECATED);
            $this->client = $config;
            $config = [];
        } else {
            throw new \LogicException('The config must be either an array of options, a DSN string or null');
        }

        $config = array_replace([
            'host' => '127.0.0.1',
            'port' => '8086',
            'user' => '',
            'password' => '',
            'db' => 'enqueue',
            'measurementSentMessages' => 'sent-messages',
            'measurementConsumedMessages' => 'consumed-messages',
            'measurementConsumers' => 'consumers',
            'client' => null,
            'retentionPolicy' => null,
        ], $config);

        if (null !== $config['client']) {
            if (!$config['client'] instanceof Client) {
                throw new \InvalidArgumentException(sprintf(
                    '%s configuration property is expected to be an instance of %s class. %s was passed instead.',
                    'client',
                    Client::class,
                    gettype($config['client'])
                ));
            }
            $this->client = $config['client'];
        }

        $this->config = $config;
    }

    /**
     * @param Client $client
     * @param string $config
     *
     * @return InfluxDbStorage
     */
    public static function createWithClient(Client $client, $config = 'influxdb:'): self
    {
        if (is_string($config)) {
            $config = self::parseDsn($config);
        }
        $config['client'] = $client;

        return new static($config);
    }

    public function pushConsumerStats(ConsumerStats $stats): void
    {
        $points = [];

        foreach ($stats->getQueues() as $queue) {
            $tags = [
                'queue' => $queue,
                'consumerId' => $stats->getConsumerId(),
            ];

            $values = [
                'startedAtMs' => $stats->getStartedAtMs(),
                'started' => $stats->isStarted(),
                'finished' => $stats->isFinished(),
                'failed' => $stats->isFailed(),
                'received' => $stats->getReceived(),
                'acknowledged' => $stats->getAcknowledged(),
                'rejected' => $stats->getRejected(),
                'requeued' => $stats->getRequeued(),
                'memoryUsage' => $stats->getMemoryUsage(),
                'systemLoad' => $stats->getSystemLoad(),
            ];

            if ($stats->getFinishedAtMs()) {
                $values['finishedAtMs'] = $stats->getFinishedAtMs();
            }

            $points[] = new Point($this->config['measurementConsumers'], null, $tags, $values, $stats->getTimestampMs());
        }

        $this->doWrite($points);
    }

    public function pushConsumedMessageStats(ConsumedMessageStats $stats): void
    {
        $tags = [
            'queue' => $stats->getQueue(),
            'status' => $stats->getStatus(),
        ];

        $values = [
            'receivedAt' => $stats->getReceivedAtMs(),
            'processedAt' => $stats->getTimestampMs(),
            'redelivered' => $stats->isRedelivered(),
        ];<commit_msg>Send topic and command for consumed messages<commit_after><?php

namespace Enqueue\Monitoring;

use Enqueue\Client\Config;
use Enqueue\Dsn\Dsn;
use InfluxDB\Client;
use InfluxDB\Database;
use InfluxDB\Driver\QueryDriverInterface;
use InfluxDB\Exception as InfluxDBException;
use InfluxDB\Point;

class InfluxDbStorage implements StatsStorage
{
    /**
     * @var array
     */
    private $config;

    /**
     * @var Client
     */
    private $client;

    /**
     * @var Database
     */
    private $database;

    /**
     * The config could be an array, string DSN, null or instance of InfluxDB\Client. In case of null it will attempt to connect to localhost.
     *
     * $config = [
     *   'dsn' => 'influxdb://127.0.0.1:8086',
     *   'host' => '127.0.0.1',
     *   'port' => '8086',
     *   'user' => '',
     *   'password' => '',
     *   'db' => 'enqueue',
     *   'measurementSentMessages' => 'sent-messages',
     *   'measurementConsumedMessages' => 'consumed-messages',
     *   'measurementConsumers' => 'consumers',
     *   'client' => null, # Client instance. Null by default.
     *   'retentionPolicy' => null,
     * ]
     *
     * or
     *
     * influxdb://127.0.0.1:8086?user=Jon&password=<PASSWORD>
     *
     * @param array|string|null $config
     */
    public function __construct($config = 'influxdb:')
    {
        if (false == class_exists(Client::class)) {
            throw new \LogicException('Seems client library is not installed. Please install "influxdb/influxdb-php"');
        }

        if (empty($config)) {
            $config = [];
        } elseif (is_string($config)) {
            $config = self::parseDsn($config);
        } elseif (is_array($config)) {
            $config = empty($config['dsn']) ? $config : self::parseDsn($config['dsn']);
        } elseif ($config instanceof Client) {
            // Passing Client instead of array config is deprecated because it prevents setting any configuration values
            // and causes library to use defaults.
            @trigger_error(
                sprintf('Passing %s as %s argument is deprecated. Pass it as "client" array property or use createWithClient instead',
                Client::class,
                __METHOD__
            ), E_USER_DEPRECATED);
            $this->client = $config;
            $config = [];
        } else {
            throw new \LogicException('The config must be either an array of options, a DSN string or null');
        }

        $config = array_replace([
            'host' => '127.0.0.1',
            'port' => '8086',
            'user' => '',
            'password' => '',
            'db' => 'enqueue',
            'measurementSentMessages' => 'sent-messages',
            'measurementConsumedMessages' => 'consumed-messages',
            'measurementConsumers' => 'consumers',
            'client' => null,
            'retentionPolicy' => null,
        ], $config);

        if (null !== $config['client']) {
            if (!$config['client'] instanceof Client) {
                throw new \InvalidArgumentException(sprintf(
                    '%s configuration property is expected to be an instance of %s class. %s was passed instead.',
                    'client',
                    Client::class,
                    gettype($config['client'])
                ));
            }
            $this->client = $config['client'];
        }

        $this->config = $config;
    }

    /**
     * @param Client $client
     * @param string $config
     *
     * @return InfluxDbStorage
     */
    public static function createWithClient(Client $client, $config = 'influxdb:'): self
    {
        if (is_string($config)) {
            $config = self::parseDsn($config);
        }
        $config['client'] = $client;

        return new static($config);
    }

    public function pushConsumerStats(ConsumerStats $stats): void
    {
        $points = [];

        foreach ($stats->getQueues() as $queue) {
            $tags = [
                'queue' => $queue,
                'consumerId' => $stats->getConsumerId(),
            ];

            $values = [
                'startedAtMs' => $stats->getStartedAtMs(),
                'started' => $stats->isStarted(),
                'finished' => $stats->isFinished(),
                'failed' => $stats->isFailed(),
                'received' => $stats->getReceived(),
                'acknowledged' => $stats->getAcknowledged(),
                'rejected' => $stats->getRejected(),
                'requeued' => $stats->getRequeued(),
                'memoryUsage' => $stats->getMemoryUsage(),
                'systemLoad' => $stats->getSystemLoad(),
            ];

            if ($stats->getFinishedAtMs()) {
                $values['finishedAtMs'] = $stats->getFinishedAtMs();
            }

            $points[] = new Point($this->config['measurementConsumers'], null, $tags, $values, $stats->getTimestampMs());
        }

        $this->doWrite($points);
    }

    public function pushConsumedMessageStats(ConsumedMessageStats $stats): void
    {
        $tags = [
            'queue' => $stats->getQueue(),
            'status' => $stats->getStatus(),
        ];

        $properties = $stats->getProperties();

        if (false === empty($properties[Config::TOPIC])) {
            $tags['topic'] = $properties[Config::TOPIC];
        }

        if (false === empty($properties[Config::COMMAND])) {
            $tags['command'] = $properties[Config::COMMAND];
        }

        $values = [
            'receivedAt' => $stats->getReceivedAtMs(),
            'processedAt' => $stats->getTimestampMs(),
            'redelivered' => $stats->isRedelivered(),
        ];
{'fraction_non_alphanumeric': 0.14806110458284372, 'fraction_numerical': 0.0072313115791376665, 'mean_word_length': 1.8948194662480378, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 132, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 8, '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': '3849076', 'n_tokens_mistral': 3205, 'n_tokens_neox': 3069, 'n_words': 680}