ApiClients\Client\Twitter\StreamingClient

Complexity

Total Complexity

6

Size/Duplication

Total Lines	75
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	5

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	sample()	0	4	1
A	filtered()	0	4	1
A	stream()	0	19	1
A	loopUpHydrateClassConstant()	0	9	2

<?php declare(strict_types=1);

namespace ApiClients\Client\Twitter;

use ApiClients\Foundation\Hydrator\CommandBus\Command\BuildSyncFromAsyncCommand;
use ApiClients\Foundation\Resource\ResourceInterface;
use ApiClients\Tools\CommandBus\CommandBus;
use React\EventLoop\LoopInterface;
use ReflectionObject;
use Rx\ObservableInterface;
use Rx\React\Promise;

final class StreamingClient
{
    /**
     * @var LoopInterface
     */
    protected $loop;

    /**
     * @var CommandBus
     */
    protected $commandBus;

    /**
     * @var AsyncStreamingClient
     */
    protected $client;

    /**
     * @var array
     */
    protected $hydrateClassConstantCache = [];

The variable name $hydrateClassConstantCache exceeds the maximum configured length of 20.

    /**
     * StreamingClient constructor.
     * @param LoopInterface $loop
     * @param CommandBus $commandBus
     * @param AsyncStreamingClient $client
     */
    public function __construct(LoopInterface $loop, CommandBus $commandBus, AsyncStreamingClient $client)
    {
        $this->loop = $loop;
        $this->commandBus = $commandBus;
        $this->client = $client;
    }

    public function sample(callable $listener)
    {
        $this->stream($this->client->sample(), $listener);
    }

    public function filtered(callable $listener, array $filter = [])
    {
        $this->stream($this->client->filtered($filter), $listener);
    }

    protected function stream(ObservableInterface $observable, callable $listener)
    {
        $observable->flatMap(function (ResourceInterface $resource) {
            return Promise::toObservable(
                $this->commandBus->handle(
                    new BuildSyncFromAsyncCommand(
                        $this->loopUpHydrateClassConstant($resource),
                        $resource
                    )
                )
            );
        })->subscribeCallback(
            $listener,
            function ($error) {
                throw $error;
            }
        );
        $this->loop->run();
    }

    protected function loopUpHydrateClassConstant(ResourceInterface $resource)
    {
        $class = get_class($resource);
        if (!isset($this->hydrateClassConstantCache[$class])) {
            $this->hydrateClassConstantCache[$class] = (new ReflectionObject($resource))->getConstant('HYDRATE_CLASS');
        }

        return $this->hydrateClassConstantCache[$class];
    }
}