bernardphp / bernard

src/Driver/IronMQ/Driver.php

<?php

namespace Bernard\Driver\IronMQ;

use Bernard\Driver\AbstractPrefetchDriver;
use IronMQ\IronMQ;

/**
 * Implements a Driver for use with Iron MQ:
 * https://github.com/iron-io/iron_mq_php.
 */
final class Driver extends AbstractPrefetchDriver
{
    private $ironmq;

    /**
     * @param IronMQ   $ironmq
     * @param int|null $prefetch
     */
    public function __construct(IronMQ $ironmq, $prefetch = null)
    {
        parent::__construct($prefetch);

        $this->ironmq = $ironmq;
    }

    /**
     * {@inheritdoc}
     */
    public function listQueues()
    {
        $queueNames = [];
        $page = 0;

        while ($queues = $this->ironmq->getQueues($page, 100)) {
            $queueNames += $this->pluck($queues, 'name');

            // If we get 100 results the probability of another page is high.
            if (count($queues) < 100) {
                break;
            }

            ++$page;
        }

        return $queueNames;
    }

    /**
     * {@inheritdoc}
     */
    public function createQueue($queueName)
    {
    }

    /**
     * {@inheritdoc}
     */
    public function countMessages($queueName)
    {
        if ($info = $this->ironmq->getQueue($queueName)) {
            return $info->size;
        }
    }

    /**
     * {@inheritdoc}
     */
    public function pushMessage($queueName, $message)
    {
        $this->ironmq->postMessage($queueName, $message);
    }

    /**
     * {@inheritdoc}
     */
    public function popMessage($queueName, $duration = 5)
    {
        if ($message = $this->cache->pop($queueName)) {
            return $message;
        }

        $timeout = IronMQ::GET_MESSAGE_TIMEOUT;

        $messages = $this->ironmq->getMessages($queueName, $this->prefetch, $timeout, $duration);

The method IronMQ\IronMQ::getMessages() has been deprecated with message: Use reserveMessages instead

        if (!$messages) {
            return [null, null];
        }

        foreach ($messages as $message) {
            $this->cache->push($queueName, [$message->body, $message->id]);
        }

        return $this->cache->pop($queueName);
    }

    /**
     * {@inheritdoc}
     */
    public function acknowledgeMessage($queueName, $receipt)
    {
        $this->ironmq->deleteMessage($queueName, $receipt);
    }

    /**
     * IronMQ does not support an offset when peeking messages.
     *
     * {@inheritdoc}
     */
    public function peekQueue($queueName, $index = 0, $limit = 20)
    {
        if ($messages = $this->ironmq->peekMessages($queueName, $limit)) {
            return $this->pluck($messages, 'body');
        }

        return [];
    }

    /**
     * {@inheritdoc}
     */
    public function removeQueue($queueName)
    {
        $this->ironmq->deleteQueue($queueName);
    }

    /**
     * {@inheritdoc}
     */
    public function info()
    {
        return [
            'prefetch' => $this->prefetch,
        ];
    }

    /**
     * The missing array_pluck but for objects array.
     *
     * @param array  $objects
     * @param string $property
     *
     * @return array
     */
    protected function pluck(array $objects, $property)
    {
        $function = function ($object) use ($property) {
            return $object->$property;
        };

        return array_map($function, $objects);
    }
}