PHP code example of makise-co / pool

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

makise-co / pool example snippets

declare(strict_types=1);

ionConfig;
use MakiseCo\Connection\ConnectionConfigInterface;
use MakiseCo\Connection\ConnectionInterface;
use MakiseCo\Connection\ConnectorInterface;
use MakiseCo\Pool\Exception\BorrowTimeoutException;
use MakiseCo\Pool\Pool;
use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use Swoole\Coroutine\Http\Client;

use function Swoole\Coroutine\run;

class HttpResponse
{
    public string $content;
    public int $statusCode;

    public function __construct(string $content, int $statusCode)
    {
        $this->content = $content;
        $this->statusCode = $statusCode;
    }
}

interface HttpConnectionInterface extends ConnectionInterface
{
    /**
     * Perform GET request
     *
     * @param string $path
     *
     * @return HttpResponse
     *
     * @throws RuntimeException on connection problems
     */
    public function get(string $path): HttpResponse;
}

class HttpConnectionConfig extends ConnectionConfig
{
    private float $timeout;
    private bool $ssl;

    public function __construct(string $host, int $port, ?bool $ssl = null, float $timeout = 30)
    {
        $this->timeout = $timeout;

        if ($ssl === null) {
            $ssl = $port === 443;
        }

        $this->ssl = $ssl;

        parent::__construct($host, $port, null, null, null);
    }

    public function getTimeout(): float
    {
        return $this->timeout;
    }

    public function getSsl(): bool
    {
        return $this->ssl;
    }

    public function getConnectionString(): string
    {
        return '';
    }
}

class HttpConnection implements HttpConnectionInterface
{
    private Client $client;
    private int $lastUsedAt;
    private bool $isClosed = false;

    public function __construct(Client $client)
    {
        $this->client = $client;
        $this->lastUsedAt = time();
    }

    public function __destruct()
    {
        $this->close();
    }

    public function isAlive(): bool
    {
        return !$this->isClosed;
    }

    public function close(): void
    {
        if ($this->isClosed) {
            return;
        }

        $this->isClosed = true;
        $this->client->close();
    }

    public function getLastUsedAt(): int
    {
        return $this->lastUsedAt;
    }

    public function resetSession(): void
    {
    }

    public static function connect(HttpConnectionConfig $config): self
    {
        $client = new Client($config->getHost(), $config->getPort(), $config->getSsl());
        $client->set(['timeout' => $config->getTimeout()]);

        return new self($client);
    }

    /**
     * @param string $path
     * @return HttpResponse
     *
     * @throws RuntimeException on connection errors
     */
    public function get(string $path): HttpResponse
    {
        $this->lastUsedAt = time();

        $this->client->get($path);

        $code = $this->client->getStatusCode();
        $this->checkStatusCode($code);

        return new HttpResponse(
            $this->client->getBody(),
            $this->client->getStatusCode()
        );
    }

    private function checkStatusCode(int $code): void
    {
        if ($code === SWOOLE_HTTP_CLIENT_ESTATUS_CONNECT_FAILED) {
            throw new RuntimeException('Connection failed');
        }

        if ($code === SWOOLE_HTTP_CLIENT_ESTATUS_REQUEST_TIMEOUT) {
            throw new RuntimeException('Request timeout');
        }

        if ($code === SWOOLE_HTTP_CLIENT_ESTATUS_SERVER_RESET) {
            throw new RuntimeException('Server has closed connection unexpectedly');
        }
    }
}

class HttpConnector implements ConnectorInterface
{
    /**
     * @param HttpConnectionConfig|ConnectionConfigInterface $config
     * @return HttpConnection
     */
    public function connect(ConnectionConfigInterface $config): HttpConnection
    {
        return HttpConnection::connect($config);
    }
}

class HttpPool extends Pool implements HttpConnectionInterface
{
    protected function createDefaultConnector(): HttpConnector
    {
        return new HttpConnector();
    }

    /**
     * {@inheritDoc}
     *
     * @throws BorrowTimeoutException
     */
    public function get(string $path): HttpResponse
    {
        $connection = $this->pop();

        try {
            return $connection->get($path);
        } finally {
            $this->push($connection);
        }
    }
}

run(static function () {
    $httpPool = new HttpPool(new HttpConnectionConfig('google.com', 80));
    $httpPool->setMaxActive(4);
    $httpPool->init();

    $tasks = [
        '/',
        '/help',
        '/search',
        '/test',
        '/query',
        '/images',
        '/videos',
        '/mail',
    ];

    $ch = new Channel();

    $start = microtime(true);

    foreach ($tasks as $task) {
        Coroutine::create(static function (Channel $ch, string $path) use ($httpPool) {
            $result = new class {
                public string $task;
                public ?HttpResponse $success = null;
                public ?Throwable $fail = null;
            };

            $result->task = $path;

            try {
                $result->success = $httpPool->get($path);
                $ch->push($result);
            } catch (Throwable $e) {
                $result->fail = $e;
                $ch->push($result);
            }
        }, $ch, $task);
    }

    $results = [];
    for ($i = 0, $iMax = \count($tasks); $i < $iMax; $i++) {
        $results[] = $ch->pop();
    }

    $end = microtime(true);

    foreach ($results as $result) {
        if ($result->fail !== null) {
            printf("Task: %s failed with: %s\n", $result->task, $result->fail->getMessage());

            continue;
        }

        printf("Task: %s returned %d status code\n", $result->task, $result->success->statusCode);
    }

    printf("\nResults fetched in %.4f secs\n\n", round($end - $start, 4));

    $stats = $httpPool->getStats();

    printf("Connections limit = %d\n", $stats->maxActive);
    printf("Connections count = %d\n", $stats->totalCount);
    printf("Idle connections = %d\n", $stats->idle);
    printf("Busy connections = %d\n", $stats->inUse);

    printf("Total wait time for an available connections = %f secs\n", $stats->waitDuration);
    printf("Total wait count = %d\n", $stats->waitCount);
    printf("Average wait time per one connection = %f secs\n", $stats->waitDuration / $stats->waitCount);

    $httpPool->close();
});