Bernard\Driver\MongoDBDriver

Complexity

Total Complexity

15

Size/Duplication

Total Lines	182
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	2

Importance

Changes	10
Bugs	1	Features	3

12 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	listQueues()	0	4	1
A	createQueue()	0	7	1
A	countMessages()	0	7	1
A	pushMessage()	0	12	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
A	configureQueueOptions()	0	6	1
B	configurePushOptions()	0	29	2

<?php

namespace Bernard\Driver;

use MongoCollection;
use MongoDate;
use MongoId;
use Symfony\Component\OptionsResolver\OptionsResolver;

/**
 * Driver supporting MongoDB
 *
 * @package Bernard
 */
class MongoDBDriver extends AbstractDriver
{
    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, array $options = [])
    {
        $options = $this->validateQueueOptions($options);
        $data = ['_id' => (string) $queueName];

        $this->queues->update($data, $data, $options);
    }

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

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

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

    /**
     * {@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,
        ];
    }

    /**
     * {@inheritdoc}
     */
    public function configureQueueOptions(OptionsResolver $resolver)
    {
        $resolver->setDefaults(array(
            'upsert' => true
        ));
    }

    /**
     * {@inheritdoc}
     */
    public function configurePushOptions(OptionsResolver $resolver)
    {
        //BC layer to support 2.3+ and 2.7+/3.0+ versions
        if (interface_exists('Symfony\Component\OptionsResolver\OptionsResolverInterface')) {
            //2.3+
            $resolver

The method setOptional() does not seem to exist on object<Symfony\Component\OptionsResolver\OptionsResolver>.

                ->setDefaults(array(
                    'fsync' => false,
                    'j' => false,
                ))
                ->setOptional(array(
                    'socketTimeoutMS',
                    'w',
                    'wTimeoutMS',
                ))
            ;
        } else {
            //2.7+
            $resolver
                ->setDefaults(array(
                    'fsync' => false,
                    'j' => false,
                ))
                ->setDefined('socketTimeoutMS')
                ->setDefined('w')
                ->setDefined('wTimeoutMS')
            ;
        }
    }
}