2amigos / mailer-library

src/Queue/Backend/Beanstalkd/BeanstalkdQueueStoreAdapter.php

<?php

namespace Da\Mailer\Queue\Backend\Beanstalkd;

use Da\Mailer\Exception\InvalidCallException;
use Da\Mailer\Queue\Backend\MailJobInterface;
use Da\Mailer\Queue\Backend\QueueStoreAdapterInterface;
use Pheanstalk\Job as PheanstalkJob;
use Pheanstalk\Pheanstalk;
use phpseclib3\Crypt\Random;

class BeanstalkdQueueStoreAdapter implements QueueStoreAdapterInterface
{
    /**
     * @var string the queue name
     */
    protected $queueName;
/**
     * @var int the time to run. Defaults to Pheanstalkd::DEFAULT_TTR.
     */
    protected $timeToRun;
/**
     * @var BeanstalkdQueueStoreConnection
     */
    protected $connection;
/**
     * @var int Reserves/locks a ready job in a watched tube. A timeout value of 0 will cause the server to immediately
     * return either a response or TIMED_OUT.  A positive value of timeout will limit the amount of time the client will
     * block on the reserve request until a job becomes available.
     *
     * We highly recommend a non-zero value. Defaults to 5.
     */
    protected $reserveTimeout;
/**
     * BeanstalkdQueueStoreAdapter constructor.
     *
     * @param BeanstalkdQueueStoreConnection $connection
     * @param string $queueName
     * @param int $timeToRun
     * @param int $reserveTimeOut
     */
    public function __construct(BeanstalkdQueueStoreConnection $connection, $queueName = 'mail_queue', $timeToRun = Pheanstalk::DEFAULT_TTR, $reserveTimeOut = 5)
    {
        $this->connection = $connection;
        $this->queueName = $queueName;
        $this->timeToRun = $timeToRun;
        $this->reserveTimeout = $reserveTimeOut;
        $this->init();
    }

    /**
     * @return BeanstalkdQueueStoreAdapter
     */
    public function init()
    {
        $this->getConnection()->connect();
        return $this;
    }

    /**
     * @return BeanstalkdQueueStoreConnection
     */
    public function getConnection()
    {
        return $this->connection;
    }

    /**
     * @param BeanstalkdMailJob|MailJobInterface $mailJob
     *
     * @return \Pheanstalk\Job
     */
    public function enqueue(MailJobInterface $mailJob)
    {
        $timestamp = $mailJob->getTimeToSend();

The method getTimeToSend() does not exist on Da\Mailer\Queue\Backend\MailJobInterface. It seems like you code against a sub-type of Da\Mailer\Queue\Backend\MailJobInterface such as Da\Mailer\Queue\Backend\Pdo\PdoMailJob or Da\Mailer\Queue\Backend\Beanstalkd\BeanstalkdMailJob or Da\Mailer\Queue\Backend\Redis\RedisMailJob.

        $payload = $this->createPayload($mailJob);
        $delay = (int) max(Pheanstalk::DEFAULT_DELAY, $timestamp - time());
        return $this->getConnection()

The expression return $this->getConnection()->getInstance()->useTube($this->queueName)->put($payload, Pheanstalk\Pheanstalk::DEFAULT_PRIORITY, $delay, $this->timeToRun) returns the type Pheanstalk\Job which is incompatible with the return type mandated by Da\Mailer\Queue\Backend\QueueStoreAdapterInterface::enqueue() of boolean.

            ->getInstance()
            ->useTube($this->queueName)
            ->put($payload, Pheanstalk::DEFAULT_PRIORITY, $delay, $this->timeToRun);
    }

    /**
     * @return BeanstalkdMailJob|null
     * @throws \Pheanstalk\Exception\DeadlineSoonException
     */
    public function dequeue()
    {
        $job = $this->getConnection()->getInstance()->watch($this->queueName)->reserveWithTimeout($this->reserveTimeout);
        if ($job instanceof PheanstalkJob) {
            $data = json_decode($job->getData(), true);
            return new BeanstalkdMailJob([
                    'id' => $data['id'],
                    'attempt' => $data['attempt'],
                    'message' => $data['message'],
                    'pheanstalkJob' => $job,
                ]);
        }

        return null;
    }

    /**
     * @param BeanstalkdMailJob|MailJobInterface $mailJob
     * @return null
     */
    public function ack(MailJobInterface $mailJob)
    {
        if ($mailJob->isNewRecord()) {
            throw new InvalidCallException('BeanstalkdMailJob cannot be a new object to be acknowledged');
        }

        $pheanstalk = $this->getConnection()->getInstance()->useTube($this->queueName);
        if ($mailJob->isCompleted()) {
            $pheanstalk->delete($mailJob->getPheanstalkJob());

The method getPheanstalkJob() does not exist on Da\Mailer\Queue\Backend\MailJobInterface. It seems like you code against a sub-type of Da\Mailer\Queue\Backend\MailJobInterface such as Da\Mailer\Queue\Backend\Beanstalkd\BeanstalkdMailJob.

            return null;
        }

        $timestamp = $mailJob->getTimeToSend();
        $delay = max(0, $timestamp - time());
// add back to the queue as it wasn't completed maybe due to some transitory error
        // could also be failed.
        $pheanstalk->release($mailJob->getPheanstalkJob(), Pheanstalk::DEFAULT_PRIORITY, $delay);
        return null;
    }

    /**
     *
     * @return bool
     */
    public function isEmpty()
    {
        $stats = $this->getConnection()->getInstance()->statsTube($this->queueName);
        return (int) $stats->current_jobs_delayed === 0

Accessing current_jobs_delayed on the interface Pheanstalk\Contract\ResponseInterface suggest that you code against a concrete implementation. How about adding an instanceof check?

        && (int) $stats->current_jobs_urgent === 0

Accessing current_jobs_urgent on the interface Pheanstalk\Contract\ResponseInterface suggest that you code against a concrete implementation. How about adding an instanceof check?

        && (int) $stats->current_jobs_ready === 0;

Accessing current_jobs_ready on the interface Pheanstalk\Contract\ResponseInterface suggest that you code against a concrete implementation. How about adding an instanceof check?

    }

    /**
     * @param BeanstalkdMailJob|MailJobInterface $mailJob
     *
     * @return string
     */
    protected function createPayload(MailJobInterface $mailJob)
    {
        return json_encode([
                'id' => $mailJob->isNewRecord() ? sha1(Random::string(32)) : $mailJob->getId(),

The method getId() does not exist on Da\Mailer\Queue\Backend\MailJobInterface. Since it exists in all sub-types, consider adding an abstract or default implementation to Da\Mailer\Queue\Backend\MailJobInterface.

                'attempt' => $mailJob->getAttempt(),

The method getAttempt() does not exist on Da\Mailer\Queue\Backend\MailJobInterface. Since it exists in all sub-types, consider adding an abstract or default implementation to Da\Mailer\Queue\Backend\MailJobInterface.

                'message' => $mailJob->getMessage(),
            ]);
    }
}