GpsLab\Component\Command\Queue\PubSub\PredisCommandQueue

Complexity

Total Complexity

5

Size/Duplication

Total Lines	85
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	3

Importance

Changes

0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	11	1
A	publish()	0	7	1
A	subscribe()	0	22	3

<?php

/**
 * GpsLab component.
 *
 * @author    Peter Gribanov <info@peter-gribanov.ru>
 * @copyright Copyright (c) 2011, Peter Gribanov
 * @license   http://opensource.org/licenses/MIT
 */

namespace GpsLab\Component\Command\Queue\PubSub;

use GpsLab\Component\Command\Command;
use Psr\Log\LoggerInterface;
use Superbalist\PubSub\Redis\RedisPubSubAdapter;
use Symfony\Component\Serializer\SerializerInterface;

class PredisCommandQueue implements CommandQueue
{
    const FORMAT = 'predis';

    /**
     * @var RedisPubSubAdapter
     */
    private $client;

    /**
     * @var SerializerInterface
     */
    private $serializer;

    /**
     * @var LoggerInterface
     */
    private $logger;

    /**
     * @var string
     */
    private $queue_name = '';

    /**
     * @param RedisPubSubAdapter  $client
     * @param SerializerInterface $serializer
     * @param LoggerInterface     $logger
     * @param string              $queue_name
     */
    public function __construct(
        RedisPubSubAdapter $client,
        SerializerInterface $serializer,
        LoggerInterface $logger,
        $queue_name
    ) {
        $this->client = $client;
        $this->serializer = $serializer;
        $this->logger = $logger;
        $this->queue_name = $queue_name;
    }

    /**
     * Publish command to queue.
     *
     * @param Command $command
     *
     * @return bool
     */
    public function publish(Command $command)
    {
        $massage = $this->serializer->serialize($command, self::FORMAT);
        $this->client->publish($this->queue_name, $massage);

        return true;
    }

    /**
     * Subscribe on command queue.
     *
     * @param callable $handler
     */
    public function subscribe(callable $handler)
    {
        $this->client->subscribe($this->queue_name, function ($message) use ($handler) {
            try {
                $command = $this->serializer->deserialize($message, Command::class, self::FORMAT);
            } catch (\Exception $e) { // catch only deserialize exception
                // it's a critical error
                // it is necessary to react quickly to it
                $this->logger->critical(
                    'Failed denormalize a command in the Redis queue',
                    [$message, $e->getMessage()]
                );

                // try denormalize in later
                $this->client->publish($this->queue_name, $message);
            }

            if (isset($command)) {
                call_user_func($handler, $command);
            }
        });
    }
}