Sonata\NotificationBundle\Document\MessageManager

Complexity

Total Complexity

33

Size/Duplication

Total Lines	211
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	5

Importance

Changes

0

9 Methods

Rating	Name	Duplication	Size	Complexity
A	save()	0	9	3
A	findByTypes()	0	12	2
A	findByAttempts()	0	22	3
B	countStates()	0	27	3
A	cleanup()	0	12	1
A	cancel()	0	10	4
A	restart()	0	14	4
C	getPager()	0	37	7
B	prepareStateQuery()	0	25	6

<?php

/*
 * This file is part of the Sonata Project package.
 *
 * (c) Thomas Rabaix <thomas.rabaix@sonata-project.org>
 * (c) Salma Khemiri Chakroun <chakroun.salma@gmail.com>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Sonata\NotificationBundle\Document;

use Sonata\CoreBundle\Model\BaseDocumentManager;
use Sonata\DatagridBundle\Pager\Doctrine\Pager;
use Sonata\DatagridBundle\ProxyQuery\Doctrine\ProxyQuery;
use NotificationEngine\Sonata\NotificationBundle\Model\MessageInterface;
use NotificationEngine\Sonata\NotificationBundle\Model\MessageManagerInterface;
use Doctrine\ODM\MongoDB\Cursor;
use Doctrine\MongoDB\ArrayIterator;

class MessageManager extends BaseDocumentManager implements MessageManagerInterface
{
    /**
     * {@inheritDoc}
     */
    public function save($message, $andFlush = true)
    {
        //Hack for ConsumerHandlerCommand->optimize()
        if ($message->getId() && !$this->dm->getUnitOfWork()->isInIdentityMap($message)) {
            $this->getDocumentManager()->getUnitOfWork()->merge($message);

It seems like you code against a concrete implementation and not the interface Doctrine\Common\Persistence\ObjectManager as the method getUnitOfWork() does only exist in the following implementations of said interface: Doctrine\ORM\Decorator\EntityManagerDecorator, Doctrine\ORM\EntityManager.

        }
        
        parent::save($message, $andFlush);
    }

    /**
     * {@inheritDoc}
     */
    public function findByTypes(array $types, $state, $batchSize)
    {
        $query = $this->prepareStateQuery($state, $types, $batchSize);

        $result = $query->getQuery()->execute();
        
        if ($result instanceof Cursor) {

The class Doctrine\ODM\MongoDB\Cursor does not exist. Did you forget a USE statement, or did you not list all dependencies?

            $result = $result->toArray();
        }
        
        return $result;
    }

    /**
     * {@inheritDoc}
     */
    public function findByAttempts(array $types, $state, $batchSize, $maxAttempts = null, $attemptDelay = 10)
    {
        $query = $this->prepareStateQuery($state, $types, $batchSize);

        if ($maxAttempts) {
            $now = new \DateTime();
            $delayDate = $now->add(\DateInterval::createFromDateString(($attemptDelay * -1) . ' second'));
            
            $query
                ->field('restartCount')->lt($maxAttempts)
                ->field('updatedAt')->lt($delayDate);
            
        }

        $result = $query->getQuery()->execute();
        
        if ($result instanceof Cursor) {

The class Doctrine\ODM\MongoDB\Cursor does not exist. Did you forget a USE statement, or did you not list all dependencies?

            $result = $result->toArray();
        }
        
        return $result;
    }

    /**
     * {@inheritDoc}
     */
    public function countStates()
    {        
        $result = $this->getRepository()->createQueryBuilder('message')
                                        ->group(array('state' => 1), array('count' => 0))
                                        ->reduce('function (curr, result) { result.count++; }')
                                        ->getQuery()
                                        ->execute();
        
        $states = array(
            MessageInterface::STATE_DONE        => 0,
            MessageInterface::STATE_ERROR       => 0,
            MessageInterface::STATE_IN_PROGRESS => 0,
            MessageInterface::STATE_OPEN        => 0,
        );
        
        if ($result instanceof ArrayIterator) 

The class Doctrine\MongoDB\ArrayIterator does not exist. Did you forget a USE statement, or did you not list all dependencies?

        {
            $result = $result->toArray();
            
            foreach ($result as $data) 
            {
                $states[$data['state']] = $data['count'];
            }
        }

        return $states;
    }

    /**
     * {@inheritDoc}
     */
    public function cleanup($maxAge)
    {
        $date = new \DateTime('now');
        $date->sub(new \DateInterval(sprintf('PT%sS', $maxAge)));
        
        $qb = $this->getRepository()->createQueryBuilder('message')

$qb is not used, you could remove the assignment.

                                    ->remove()
                                    ->field('state')->equals(MessageInterface::STATE_DONE)
                                    ->field('completedAt')->lt($date)
                                    ->getQuery()
                                    ->execute();
    }

    /**
     * {@inheritdoc}
     */
    public function cancel(MessageInterface $message, $force = false)
    {
        if (($message->isRunning() || $message->isError()) && !$force) {
            return;
        }

        $message->setState(MessageInterface::STATE_CANCELLED);

        $this->save($message);
    }

    /**
     * {@inheritdoc}
     */
    public function restart(MessageInterface $message)
    {
        if ($message->isOpen() || $message->isRunning() || $message->isCancelled()) {
            return;
        }

        $this->cancel($message, true);

        $newMessage = clone $message;
        $newMessage->setRestartCount($message->getRestartCount() + 1);
        $newMessage->setType($message->getType());

        return $newMessage;
    }

    /**
     * {@inheritdoc}
     */
    public function getPager(array $criteria, $page, $limit = 10, array $sort = array())
    {
        $query = $this->getRepository()
            ->createQueryBuilder('message');

        $fields = $this->getDocumentManager()->getClassMetadata($this->class)->getFieldNames();
        foreach ($sort as $field => $direction) {
            if (!in_array($field, $fields)) {
                throw new \RuntimeException(sprintf("Invalid sort field '%s' in '%s' class", $field, $this->class));
            }
        }
        if (count($sort) == 0) {
            $sort = array('type' => 'ASC');
        }
        foreach ($sort as $field => $direction) {
            $query->sort(sprintf('m.%s', $field), strtoupper($direction));
        }

        $parameters = array();

$parameters is not used, you could remove the assignment.

        if (isset($criteria['type'])) {
            $query->field('type')->equals($criteria['type']);
        }

        if (isset($criteria['state'])) {
            $query->field('state')->equals($criteria['state']);
        }

        $pager = new Pager();

        $pager->setMaxPerPage($limit);
        $pager->setQuery(new ProxyQuery($query));
        $pager->setPage($page);
        $pager->init();

        return $pager;
    }

    /**
     * @param int   $state
     * @param array $types
     * @param int   $batchSize
     * @param array $parameters

There is no parameter named $parameters. Was it maybe removed?

     *
     * @return QueryBuilder
     */
    protected function prepareStateQuery($state, $types, $batchSize)
    {
        $query = $this->getRepository()
            ->createQueryBuilder('message')
            ->field('state')->equals($state)
            ->sort('createdAt');
        
        if (count($types) > 0) {
            if (array_key_exists('exclude', $types) || array_key_exists('include', $types)) {
                if (array_key_exists('exclude', $types)) {
                    $query->field('type')->notIn(array($types['exclude']));
                }

                if (array_key_exists('include', $types)) {
                    $query->field('type')->in(array($types['include']));
                }
            } else {
                $query->field('type')->in(array($types));
            }
        }

        $query->limit($batchSize);

        return $query;
    }
}