Bernard\Queue\PersistentQueue

Complexity

Total Complexity

13

Size/Duplication

Total Lines	125
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Importance

Changes	10
Bugs	1	Features	1

9 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	10	1
A	register()	0	6	1
A	count()	0	6	1
A	close()	0	6	1
A	enqueue()	0	11	2
A	acknowledge()	0	10	2
A	dequeue()	0	15	2
A	peek()	0	8	1
A	getReceipt()	0	8	2

<?php

namespace Bernard\Queue;

use Bernard\Driver;
use Bernard\Envelope;
use Bernard\Serializer;

/**
 * @package Bernard
 */
class PersistentQueue extends AbstractQueue
{
    protected $driver;
    protected $serializer;
    protected $receipts;

    /**
     * @param string     $name
     * @param Driver     $driver
     * @param Serializer $serializer
     */
    public function __construct($name, Driver $driver, Serializer $serializer)
    {
        parent::__construct($name);

        $this->driver = $driver;
        $this->serializer = $serializer;
        $this->receipts = new \SplObjectStorage();

        $this->register();
    }

    /**
     * Register with the driver
     */
    public function register()
    {
        $this->errorIfClosed();

        $this->driver->createQueue($this->name);
    }

    /**
     * {@inheritdoc}
     */
    public function count()
    {
        $this->errorIfClosed();

        return $this->driver->countMessages($this->name);
    }

    /**
     * {@inheritdoc}
     */
    public function close()
    {
        parent::close();

        $this->driver->removeQueue($this->name);
    }

    /**
     * {@inheritdoc}
     */
    public function enqueue(Envelope $envelope)
    {
        $this->errorIfClosed();

        $receipt = $this->driver->pushMessage($this->name, $this->serializer->serialize($envelope));

        if ($receipt) {
            $envelope->setReceipt($receipt);
            $this->receipts->attach($envelope, $receipt);
        }
    }

    /**
     * {@inheritdoc}
     */
    public function acknowledge(Envelope $envelope)
    {
        $this->errorIfClosed();

        if ($this->receipts->contains($envelope)) {
            $this->driver->acknowledgeMessage($this->name, $this->receipts[$envelope]);

            $this->receipts->detach($envelope);
        }
    }

    /**
     * {@inheritdoc}
     */
    public function dequeue()
    {
        $this->errorIfClosed();

        list($serialized, $receipt) = $this->driver->popMessage($this->name);

        if ($serialized) {
            $envelope = $this->serializer->unserialize($serialized);

            $envelope->setReceipt($receipt);
            $this->receipts->attach($envelope, $receipt);

            return $envelope;
        }
    }

    /**
     * {@inheritdoc}
     */
    public function peek($index = 0, $limit = 20)
    {
        $this->errorIfClosed();

        $messages = $this->driver->peekQueue($this->name, $index, $limit);

        return array_map(array($this->serializer, 'unserialize'), $messages);
    }

    /**
     * @param Envelope $envelope The envelope.
     * @return mixed
     */
    public function getReceipt(Envelope $envelope)
    {
        if (!$this->receipts->contains($envelope)) {
            return null;
        }

        return $this->receipts->offsetGet($envelope);
    }
}