Laratrade\GDAX\WebSocket\Subscriber

Complexity

Total Complexity

8

Size/Duplication

Total Lines	115
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	5

Test Coverage

Coverage

100%

Importance

Changes

0

6 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	onConnect()	0	15	1
A	onDisconnect()	0	4	1
A	onError()	0	4	1
A	subscribe()	0	4	1
A	onMessage()	0	12	3

<?php

namespace Laratrade\GDAX\WebSocket;

use Exception;
use Illuminate\Contracts\Events\Dispatcher as DispatcherContract;
use Laratrade\GDAX\WebSocket\Contracts\Subscriber as SubscriberContract;
use Psr\Log\LoggerInterface as LoggerContract;
use Ratchet\Client\WebSocket;
use Ratchet\RFC6455\Messaging\MessageInterface as MessageContract;
use React\Promise\PromiseInterface as PromiseContract;

class Subscriber implements SubscriberContract
{
    /**
     * The logger instance.
     *
     * @var LoggerContract
     */
    protected $logger;

    /**
     * The dispatcher instance.
     *
     * @var DispatcherContract
     */
    protected $dispatcher;

    /**
     * The event mappings.
     *
     * @var array
     */
    protected $events;

    /**
     * Create a new subscriber instance.
     *
     * @param LoggerContract     $logger
     * @param DispatcherContract $dispatcher
     * @param array              $events
     */
    public function __construct(LoggerContract $logger, DispatcherContract $dispatcher, array $events)
    {
        $this->logger     = $logger;
        $this->dispatcher = $dispatcher;
        $this->events     = $events;
    }

    /**
     * Handle the connect event.
     *
     * @param WebSocket $webSocket
     *
     * @return void
     */
    public function onConnect(WebSocket $webSocket): void
    {
        $this->logger->info('websocket connect');

        $webSocket->on('message', [$this, 'onMessage']);
        $webSocket->on('close', [$this, 'onDisconnect']);

        $webSocket->send(json_encode([
            'type'        => 'subscribe',
            'product_ids' => [],
            'channels'    => [
                'ticker',
            ],
        ]));
    }

    /**
     * Handle the message event.
     *
     * @param MessageContract $message
     *
     * @return void
     */
    public function onMessage(MessageContract $message): void
    {
        $payload = json_decode($message->getPayload());

        $this->logger->info('websocket message', compact('payload'));

        if (!isset($payload->type) || !isset($this->events[$payload->type])) {
            return;
        }

        $this->dispatcher->dispatch(new $this->events[$payload->type]($payload));
    }

    /**
     * Handle the disconnect event.
     *
     * @param int|null    $code
     * @param string|null $reason
     *
     * @return void
     */
    public function onDisconnect(int $code = null, string $reason = null): void
    {
        $this->logger->warning('websocket disconnect', compact('code', 'reason'));
    }

    /**
     * Handle the error event.
     *
     * @param Exception $exception
     *
     * @return void
     */
    public function onError(Exception $exception): void
    {
        $this->logger->error('websocket error', compact('exception'));
    }

    /**
     * Register the listeners for the subscriber.
     *
     * @param PromiseContract $connection
     */
    public function subscribe(PromiseContract $connection): void
    {
        $connection->then([$this, 'onConnect'], [$this, 'onError']);
    }
}