BenTools\MercurePHP\Hub\Hub

Complexity

Total Complexity

7

Size/Duplication

Total Lines	83
Duplicated Lines	0 %

Test Coverage

Coverage

30.95%

Importance

Changes	1
Bugs	0	Features	0

7 Methods

Rating	Name	Duplication	Size	Complexity
A	stop()	0	5	1
A	serve()	0	7	1
A	createSocketConnection()	0	9	1
A	getShutdownSignal()	0	3	1
A	__construct()	0	13	1
A	__invoke()	0	4	1
A	run()	0	20	1

<?php

namespace BenTools\MercurePHP\Hub;

use BenTools\MercurePHP\Configuration\Configuration;
use BenTools\MercurePHP\Metrics\MetricsHandlerInterface;
use BenTools\MercurePHP\Security\CORS;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
use Psr\Log\LoggerInterface;
use React\EventLoop\LoopInterface;
use React\Http;
use React\Promise\PromiseInterface;
use React\Socket;
use React\Socket\ConnectionInterface;

final class Hub implements HubInterface
{
    private array $config;
    private LoopInterface $loop;
    private LoggerInterface $logger;
    private RequestHandler $requestHandler;
    private CORS $cors;
    private MetricsHandlerInterface $metricsHandler;
    private ?int $shutdownSignal;

    public function __construct(
        array $config,
        LoopInterface $loop,
        RequestHandler $requestHandler,
        MetricsHandlerInterface $metricsHandler,
        LoggerInterface $logger
    ) {
        $this->config = $config;
        $this->loop = $loop;
        $this->requestHandler = $requestHandler;
        $this->metricsHandler = $metricsHandler;
        $this->logger = $logger;
        $this->cors = new CORS($config);
    }

    public function run(): void
    {
        $localAddress = $this->config[Configuration::ADDR];
        $this->shutdownSignal = null;
        $this->metricsHandler->resetUsers($localAddress);
        $this->loop->addSignal(SIGINT, function ($signal) {
            $this->stop($signal, $this->loop);
        });
        $this->loop->addPeriodicTimer(
            15,
            fn() => $this->metricsHandler->getNbUsers()->then(
                function (int $nbUsers) {
                    $memory = \memory_get_usage(true) / 1024 / 1024;
                    $this->logger->debug("Users: {$nbUsers} - Memory: {$memory}MB");
                }
            )
        );

        $socket = $this->createSocketConnection($localAddress, $this->loop);
        $this->serve($localAddress, $socket, $this->loop);
    }

    public function getShutdownSignal(): ?int
    {
        return $this->shutdownSignal;
    }

    public function __invoke(ServerRequestInterface $request): PromiseInterface
    {
        return $this->requestHandler->handle($request)
            ->then(fn(ResponseInterface $response) => $this->cors->decorateResponse($request, $response));
    }

    private function createSocketConnection(string $localAddress, LoopInterface $loop): Socket\Server
    {
        $socket = new Socket\Server($localAddress, $loop);
        $socket->on('connection', function (ConnectionInterface $connection) use ($localAddress) {
            $this->metricsHandler->incrementUsers($localAddress);
            $connection->on('close', fn() => $this->metricsHandler->decrementUsers($localAddress));
        });

        return $socket;
    }

    private function serve(string $localAddress, Socket\Server $socket, LoopInterface $loop): void
    {
        $server = new Http\Server($loop, $this);
        $server->listen($socket);

        $this->logger->info("Server running at http://" . $localAddress);
        $loop->run();
    }

    private function stop(int $signal, LoopInterface $loop): void
    {
        $this->shutdownSignal = $signal;
        $loop->futureTick(function () use ($loop) {
            $loop->stop();
        });
    }
}