<commit_before><?php

declare(strict_types=1);

namespace Enqueue\Dbal;

use Doctrine\DBAL\Connection;
use Interop\Queue\Consumer;
use Interop\Queue\SubscriptionConsumer;

class DbalSubscriptionConsumer implements SubscriptionConsumer
{
    use DbalConsumerHelperTrait;

    /**
     * @var DbalContext
     */
    private $context;

    /**
     * an item contains an array: [DbalConsumer $consumer, callable $callback];.
     *
     * @var array
     */
    private $subscribers;

    /**
     * @var \Doctrine\DBAL\Connection
     */
    private $dbal;

    /**
     * Default 20 minutes in milliseconds.
     *
     * @var int
     */
    private $redeliveryDelay;

    /**
     * Time to wait between subscription requests in milliseconds.
     *
     * @var int
     */
    private $pollingInterval = 200;

    public function __construct(DbalContext $context)
    {
        $this->context = $context;
        $this->dbal = $this->context->getDbalConnection();
        $this->subscribers = [];

        $this->redeliveryDelay = 1200000;
    }

    /**
     * Get interval between retrying failed messages in milliseconds.
     */
    public function getRedeliveryDelay(): int
    {
        return $this->redeliveryDelay;
    }

    public function setRedeliveryDelay(int $redeliveryDelay): self
    {
        $this->redeliveryDelay = $redeliveryDelay;

        return $this;
    }

    public function getPollingInterval(): int
    {
        return $this->pollingInterval;
    }

    public function setPollingInterval(int $msec): self
    {
        $this->pollingInterval = $msec;

        return $this;
    }

    public function consume(int $timeout = 0): void
    {
        if (empty($this->subscribers)) {
            throw new \LogicException('No subscribers');
        }

        $queueNames = [];
        foreach (array_keys($this->subscribers) as $queueName) {
            $queueNames[$queueName] = $queueName;
        }

        $timeout /= 1000;
        $now = time();
        $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds

        $currentQueueNames = [];
        while (true) {
            if (empty($currentQueueNames)) {
                $currentQueueNames = $queueNames;
            }

            $this->removeExpiredMessages();
            $this->redeliverMessages();

            if ($message = $this->fetchMessage($currentQueueNames, $redeliveryDelay)) {
                /**
                 * @var DbalConsumer
                 * @var callable     $callback
                 */
                [$consumer, $callback] = $this->subscribers[$message->getQueue()];

                if (false === call_user_func($callback, $message, $consumer)) {
                    return;
                }

                unset($currentQueueNames[$message->getQueue()]);
            } else {
                $currentQueueNames = [];

                usleep($this->getPollingInterval() * 1000);
            }

            if ($timeout && microtime(true) >= $now + $timeout) {
                return;
            }
        }
    }

    /**
     * @param DbalConsumer $consumer
     */
    public function subscribe(Consumer $consumer, callable $callback): void
    {
        if (false == $consumer instanceof DbalConsumer) {
            throw new \InvalidArgumentException(sprintf('The consumer must be instance of "%s" got "%s"', DbalConsumer::class, get_class($consumer)));
        }

        $queueName = $consumer->getQueue()->getQueueName();
        if (array_key_exists($queueName, $this->subscribers)) {
            if ($this->subscribers[$queueName][0] === $consumer && $this->subscribers[$queueName][1] === $callback) {
                return;<commit_msg>just work on the next message based on priority and published_at<commit_after><?php

declare(strict_types=1);

namespace Enqueue\Dbal;

use Doctrine\DBAL\Connection;
use Interop\Queue\Consumer;
use Interop\Queue\SubscriptionConsumer;

class DbalSubscriptionConsumer implements SubscriptionConsumer
{
    use DbalConsumerHelperTrait;

    /**
     * @var DbalContext
     */
    private $context;

    /**
     * an item contains an array: [DbalConsumer $consumer, callable $callback];.
     *
     * @var array
     */
    private $subscribers;

    /**
     * @var \Doctrine\DBAL\Connection
     */
    private $dbal;

    /**
     * Default 20 minutes in milliseconds.
     *
     * @var int
     */
    private $redeliveryDelay;

    /**
     * Time to wait between subscription requests in milliseconds.
     *
     * @var int
     */
    private $pollingInterval = 200;

    public function __construct(DbalContext $context)
    {
        $this->context = $context;
        $this->dbal = $this->context->getDbalConnection();
        $this->subscribers = [];

        $this->redeliveryDelay = 1200000;
    }

    /**
     * Get interval between retrying failed messages in milliseconds.
     */
    public function getRedeliveryDelay(): int
    {
        return $this->redeliveryDelay;
    }

    public function setRedeliveryDelay(int $redeliveryDelay): self
    {
        $this->redeliveryDelay = $redeliveryDelay;

        return $this;
    }

    public function getPollingInterval(): int
    {
        return $this->pollingInterval;
    }

    public function setPollingInterval(int $msec): self
    {
        $this->pollingInterval = $msec;

        return $this;
    }

    public function consume(int $timeout = 0): void
    {
        if (empty($this->subscribers)) {
            throw new \LogicException('No subscribers');
        }

        $queueNames = [];
        foreach (array_keys($this->subscribers) as $queueName) {
            $queueNames[$queueName] = $queueName;
        }

        $timeout /= 1000;
        $now = time();
        $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds

        while (true) {
            $this->removeExpiredMessages();
            $this->redeliverMessages();

            if ($message = $this->fetchMessage($queueNames, $redeliveryDelay)) {
                /**
                 * @var DbalConsumer
                 * @var callable     $callback
                 */
                [$consumer, $callback] = $this->subscribers[$message->getQueue()];

                if (false === call_user_func($callback, $message, $consumer)) {
                    return;
                }
            } else {
                usleep($this->getPollingInterval() * 1000);
            }

            if ($timeout && microtime(true) >= $now + $timeout) {
                return;
            }
        }
    }

    /**
     * @param DbalConsumer $consumer
     */
    public function subscribe(Consumer $consumer, callable $callback): void
    {
        if (false == $consumer instanceof DbalConsumer) {
            throw new \InvalidArgumentException(sprintf('The consumer must be instance of "%s" got "%s"', DbalConsumer::class, get_class($consumer)));
        }

        $queueName = $consumer->getQueue()->getQueueName();
        if (array_key_exists($queueName, $this->subscribers)) {
            if ($this->subscribers[$queueName][0] === $consumer && $this->subscribers[$queueName][1] === $callback) {
                return;
{'fraction_non_alphanumeric': 0.12432357430276121, 'fraction_numerical': 0.007770223393922576, 'mean_word_length': 2.050359712230216, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 54, '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': '7375421', 'n_tokens_mistral': 2109, 'n_tokens_neox': 1958, 'n_words': 430}