<commit_before><?php

declare(strict_types=1);

namespace Enqueue\Redis;

use Interop\Queue\Destination;
use Interop\Queue\Exception\InvalidDestinationException;
use Interop\Queue\Exception\InvalidMessageException;
use Interop\Queue\Exception\PriorityNotSupportedException;
use Interop\Queue\Message;
use Interop\Queue\Producer;
use Ramsey\Uuid\Uuid;

class RedisProducer implements Producer
{
    /**
     * @var RedisContext
     */
    private $context;

    /**
     * @var int|null
     */
    private $timeToLive;

    /**
     * @var int
     */
    private $deliveryDelay;

    /**
     * @param RedisContext $context
     */
    public function __construct(RedisContext $context)
    {
        $this->context = $context;
    }

    /**
     * @param RedisDestination $destination
     * @param RedisMessage     $message
     */
    public function send(Destination $destination, Message $message): void
    {
        InvalidDestinationException::assertDestinationInstanceOf($destination, RedisDestination::class);
        InvalidMessageException::assertMessageInstanceOf($message, RedisMessage::class);

        $message->setMessageId(Uuid::uuid4()->toString());
        $message->setHeader('attempts', 0);

        if (null !== $this->timeToLive && null === $message->getTimeToLive()) {
            $message->setTimeToLive($this->timeToLive);
        }

        if (null !== $this->deliveryDelay && null === $message->getDeliveryDelay()) {
            $message->setDeliveryDelay($this->deliveryDelay);
        }

        if ($message->getTimeToLive()) {
            $message->setHeader('expires_at', time() + $message->getTimeToLive());
        }

        $payload = $this->context->getSerializer()->toString($message);

        if ($message->getDeliveryDelay()) {
            $deliveryAt = time() + $message->getDeliveryDelay();
            $this->context->getRedis()->zadd($destination->getName().':delayed', $payload, $deliveryAt);
        } else {
            $this->context->getRedis()->lpush($destination->getName(), $payload);
        }
    }

    /**
     * @return self
     */
    public function setDeliveryDelay(int $deliveryDelay = null): Producer
    {
        $this->deliveryDelay = $deliveryDelay;

        return $this;
    }

    public function getDeliveryDelay(): ?int
    {
        return $this->deliveryDelay;
    }

    /**
     * @return RedisProducer<commit_msg>Fix $deliveryAt in RedisProducer::send by converting deliveryDelay milliseconds to seconds<commit_after><?php

declare(strict_types=1);

namespace Enqueue\Redis;

use Interop\Queue\Destination;
use Interop\Queue\Exception\InvalidDestinationException;
use Interop\Queue\Exception\InvalidMessageException;
use Interop\Queue\Exception\PriorityNotSupportedException;
use Interop\Queue\Message;
use Interop\Queue\Producer;
use Ramsey\Uuid\Uuid;

class RedisProducer implements Producer
{
    /**
     * @var RedisContext
     */
    private $context;

    /**
     * @var int|null
     */
    private $timeToLive;

    /**
     * @var int
     */
    private $deliveryDelay;

    /**
     * @param RedisContext $context
     */
    public function __construct(RedisContext $context)
    {
        $this->context = $context;
    }

    /**
     * @param RedisDestination $destination
     * @param RedisMessage     $message
     */
    public function send(Destination $destination, Message $message): void
    {
        InvalidDestinationException::assertDestinationInstanceOf($destination, RedisDestination::class);
        InvalidMessageException::assertMessageInstanceOf($message, RedisMessage::class);

        $message->setMessageId(Uuid::uuid4()->toString());
        $message->setHeader('attempts', 0);

        if (null !== $this->timeToLive && null === $message->getTimeToLive()) {
            $message->setTimeToLive($this->timeToLive);
        }

        if (null !== $this->deliveryDelay && null === $message->getDeliveryDelay()) {
            $message->setDeliveryDelay($this->deliveryDelay);
        }

        if ($message->getTimeToLive()) {
            $message->setHeader('expires_at', time() + $message->getTimeToLive());
        }

        $payload = $this->context->getSerializer()->toString($message);

        if ($message->getDeliveryDelay()) {
            $deliveryAt = time() + $message->getDeliveryDelay() / 1000;
            $this->context->getRedis()->zadd($destination->getName().':delayed', $payload, $deliveryAt);
        } else {
            $this->context->getRedis()->lpush($destination->getName(), $payload);
        }
    }

    /**
     * @return self
     */
    public function setDeliveryDelay(int $deliveryDelay = null): Producer
    {
        $this->deliveryDelay = $deliveryDelay;

        return $this;
    }

    public function getDeliveryDelay(): ?int
    {
        return $this->deliveryDelay;
    }

    /**
     * @return RedisProducer
{'fraction_non_alphanumeric': 0.14227142271422713, 'fraction_numerical': 0.002050020500205002, 'mean_word_length': 2.875297855440826, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 63, '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': '4092681', 'n_tokens_mistral': 1402, 'n_tokens_neox': 1375, 'n_words': 235}