Genkgo\Mail\Queue\QueueProcessor

Complexity

Total Complexity

13

Size/Duplication

Total Lines	87
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	2

Test Coverage

Coverage

100%

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
C	process()	0	30	8
A	hasMessageExpired()	0	11	3
A	getSubmissionTimestamp()	0	6	1

<?php
declare(strict_types=1);

namespace Genkgo\Mail\Queue;

use Genkgo\Mail\Exception\AbstractProtocolException;
use Genkgo\Mail\Exception\EmptyQueueException;
use Genkgo\Mail\MessageInterface;
use Genkgo\Mail\Transport\QueueIfFailedTransport;
use Genkgo\Mail\TransportInterface;

/**
 * Class QueueProcessor
 * @package Genkgo\Mail\Queue
 */
final class QueueProcessor
{

    /**
     * @var TransportInterface
     */
    private $transport;
    /**
     * @var array|QueueInterface[]
     */
    private $queue;

    /**
     * @param TransportInterface $transport
     * @param QueueInterface[] $queue
     */
    public function __construct(TransportInterface $transport, iterable $queue)
    {
        $this->transport = $transport;
        $this->queue = $queue;
    }

    /**
     * @param int|null $messageLimit
     * @param int|null $timeLimit
     * @return int
     * @throws \Genkgo\Mail\Exception\QueueStoreException
     */
    public function process(?int $messageLimit = null, ?int $timeLimit = null): int
    {
        $count = 0;
        foreach ($this->queue as $queue) {
            try {
                while ($message = $queue->fetch()) {
                    if ($messageLimit && $count >= $messageLimit) {

The expression $messageLimit of type integer|null is loosely compared to true; this is ambiguous if the integer can be zero. You might want to explicitly use !== null instead.

                        $queue->store($message);

                        return $count;
                    }
                    try {
                        if (!$this->hasMessageExpired($message, (int) $timeLimit)) {
                            $this->transport->send($message);
                        }
                    } catch (AbstractProtocolException $e) {
                        $queue->store($message);

                        // do not continue transporting messages
                        // apparently our transport is not ready to receive messages yet
                        return $count;
                    }
                    ++$count;
                }
            } catch (EmptyQueueException $e) {
            }
        }

        return $count;
    }

    /**
     * @param MessageInterface $message
     * @param int $timeLimit
     * @return bool
     */
    private function hasMessageExpired(MessageInterface $message, int $timeLimit): bool
    {
        if ($timeLimit === 0) {
            return false;
        }
        if ($this->getSubmissionTimestamp($message) + $timeLimit > \time()) {
            return false;
        }

        return true;
    }

    /**
     * @param MessageInterface $message
     * @return int
     */
    private function getSubmissionTimestamp(MessageInterface $message): int
    {
        $queuedAt = $message->getHeader(QueueIfFailedTransport::QUEUED_HEADER);

        return (int) \strtotime(reset($queuedAt)->getValue()->getRaw());
    }
}