ekinhbayar\GitAmp\Websocket\Handler

Complexity

Total Complexity

12

Size/Duplication

Total Lines	100
Duplicated Lines	0 %

Test Coverage

Coverage

68.75%

Importance

Changes	16
Bugs	2	Features	0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	handleHandshake()	0	7	2
A	__construct()	0	5	1
A	onStart()	0	13	2
A	emit()	0	9	2
A	handleClient()	0	18	2
A	sendConnectedUsersCount()	0	3	1
A	processDisconnectingClient()	0	14	1
A	onStop()	0	3	1

<?php declare(strict_types=1);

namespace ekinhbayar\GitAmp\Websocket;

use Amp\Delayed;
use Amp\Http\Server\HttpServer;
use Amp\Http\Server\Request;
use Amp\Http\Server\Response;
use Amp\Http\Status;
use Amp\Promise;
use Amp\Success;
use Amp\Websocket\Client;
use Amp\Websocket\Server\ClientHandler;
use Amp\Websocket\Server\Gateway;
use Amp\Websocket\Server\WebsocketServerObserver;
use ekinhbayar\GitAmp\Provider\Listener;
use ekinhbayar\GitAmp\Response\Results;
use Psr\Log\LoggerInterface;
use function Amp\asyncCall;
use function Amp\call;

class Handler implements ClientHandler, WebsocketServerObserver
{
    private Gateway $gateway;

    private string $origin;

    private Listener $provider;

    private ?Results $lastEvents = null;

    private LoggerInterface $logger;

    public function __construct(string $origin, Listener $provider, LoggerInterface $logger)
    {
        $this->origin   = $origin;
        $this->provider = $provider;
        $this->logger   = $logger;
    }

    public function handleHandshake(Gateway $gateway, Request $request, Response $response): Promise
    {
        if ($request->getHeader('origin') !== $this->origin) {
            return $gateway->getErrorHandler()->handleError(Status::FORBIDDEN, 'Forbidden Origin', $request);
        }

        return new Success($response);
    }

    public function handleClient(Gateway $gateway, Client $client, Request $request, Response $response): Promise
    {
        return call(function () use ($gateway, $client, $request, $response) {

The import $response is not used and could be removed.

The import $request is not used and could be removed.

            $client->onClose(function (Client $client, int $code, string $reason) use ($gateway) {
                yield $this->processDisconnectingClient($gateway, $client, $code, $reason);
            });

            $this->logger->info(
                \sprintf('Client %d connected. Total clients: %d', $client->getId(), count($gateway->getClients())),
            );

            yield $this->sendConnectedUsersCount(\count($gateway->getClients()));

            if ($this->lastEvents) {
                $client->send($this->lastEvents->jsonEncode());
            }

            yield $client->receive();
        });
    }

    private function processDisconnectingClient(Gateway $gateway, Client  $client, int $code, string $reason): Promise
    {
        return call(function () use ($gateway, $client, $code, $reason) {
            $this->logger->info(
                \sprintf(
                    'Client %d disconnected. Code: %d Reason: %s. Total clients: %d',
                    $client->getId(),
                    $code,
                    $reason,
                    count($gateway->getClients()),
                )
            );

            yield $this->sendConnectedUsersCount(count($gateway->getClients()));
        });
    }

    private function emit(Results $events): Promise
    {
        if (!$events->hasEvents()) {
            return new Success();
        }

        $this->lastEvents = $events;

        return $this->gateway->broadcast($events->jsonEncode());
    }

    private function sendConnectedUsersCount(int $count): Promise
    {
        return $this->gateway->broadcast(\json_encode(['connectedUsers' => $count]));
    }

    public function onStart(HttpServer $server, Gateway $gateway): Promise
    {
        $this->gateway = $gateway;

        asyncCall(function () {
            while (true) {
                $this->emit(yield $this->provider->listen());

                yield new Delayed(25000);
            }
        });

        return new Success();
    }

    public function onStop(HttpServer $server, Gateway $gateway): Promise
    {
        return new Success();
    }
}