Bernard\Driver\MongoDB\Driver

Complexity

Total Complexity

12

Size/Duplication

Total Lines	137
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	0

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	listQueues()	0	4	1
A	createQueue()	0	6	1
A	countMessages()	0	7	1
A	pushMessage()	0	11	1
A	popMessage()	0	21	3
A	acknowledgeMessage()	0	7	1
A	peekQueue()	0	18	1
A	removeQueue()	0	5	1
A	info()	0	7	1

<?php

namespace Bernard\Driver\MongoDB;

use MongoCollection;
use MongoDate;
use MongoId;

/**
 * Driver supporting MongoDB
 *
 * @package Bernard
 */
final class Driver implements \Bernard\Driver
{
    private $messages;
    private $queues;

    /**
     * @param MongoCollection $queues   Collection where queues will be stored
     * @param MongoCollection $messages Collection where messages will be stored
     */
    public function __construct(MongoCollection $queues, MongoCollection $messages)
    {
        $this->queues = $queues;
        $this->messages = $messages;
    }

    /**
     * {@inheritdoc}
     */
    public function listQueues()
    {
        return $this->queues->distinct('_id');
    }

    /**
     * {@inheritdoc}
     */
    public function createQueue($queueName)
    {
        $data = ['_id' => (string) $queueName];

        $this->queues->update($data, $data, ['upsert' => true]);
    }

    /**
     * {@inheritdoc}
     */
    public function countMessages($queueName)
    {
        return $this->messages->count([
            'queue' => (string) $queueName,
            'visible' => true,
        ]);
    }

    /**
     * {@inheritdoc}
     */
    public function pushMessage($queueName, $message)
    {
        $data = [
            'queue' => (string) $queueName,
            'message' => (string) $message,
            'sentAt' => new MongoDate(),
            'visible' => true,
        ];

        $this->messages->insert($data);
    }

    /**
     * {@inheritdoc}
     */
    public function popMessage($queueName, $duration = 5)
    {
        $runtime = microtime(true) + $duration;

        while (microtime(true) < $runtime) {
            $result = $this->messages->findAndModify(
                ['queue' => (string) $queueName, 'visible' => true],
                ['$set' => ['visible' => false]],
                ['message' => 1],
                ['sort' => ['sentAt' => 1]]
            );

            if ($result) {

The expression $result of type array is implicitly converted to a boolean; are you sure this is intended? If so, consider using ! empty($expr) instead to make it clear that you intend to check for an array without elements.

                return [(string) $result['message'], (string) $result['_id']];
            }

            usleep(10000);
        }

        return [null, null];
    }

    /**
     * {@inheritdoc}
     */
    public function acknowledgeMessage($queueName, $receipt)
    {
        $this->messages->remove([
            '_id' => new MongoId((string) $receipt),
            'queue' => (string) $queueName,
        ]);
    }

    /**
     * {@inheritdoc}
     */
    public function peekQueue($queueName, $index = 0, $limit = 20)
    {
        $query = ['queue' => (string) $queueName, 'visible' => true];
        $fields = ['_id' => 0, 'message' => 1];

        $cursor = $this->messages
            ->find($query, $fields)
            ->sort(['sentAt' => 1])
            ->limit($limit)
            ->skip($index)
        ;

        $mapper = function ($result) {
            return (string) $result['message'];
        };

        return array_map($mapper, iterator_to_array($cursor, false));
    }

    /**
     * {@inheritdoc}
     */
    public function removeQueue($queueName)
    {
        $this->queues->remove(['_id' => $queueName]);
        $this->messages->remove(['queue' => (string) $queueName]);
    }

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