FastCGI

cli/src/Command/Queue.php

@@ -7,8 +7,10 @@ use Psr\Http\Client\ClientExceptionInterface;
 use Psr\Http\Client\ClientInterface;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Console;
+use Incoviba\Service\FastCGI;
 use Incoviba\Service\Job;
 use Incoviba\Common\Alias\Command;
+use Incoviba\Exception\Client\FastCGI as FastCGIException;
 
 #[Console\Attribute\AsCommand(
     name: 'queue',
@@ -16,10 +18,11 @@ use Incoviba\Common\Alias\Command;
 )]
 class Queue extends Command
 {
-    public function __construct(ClientInterface $client, LoggerInterface $logger,
-                                protected Job $jobService,
+    public function __construct(ClientInterface        $client, LoggerInterface $logger,
+                                protected Job          $jobService,
                                 protected DateTimeZone $timezone,
-                                ?string $name = null)
+                                protected FastCGI      $fastcgi,
+                                ?string                $name = null)
     {
         parent::__construct($client, $logger, $name);
     }
@@ -55,20 +58,30 @@ class Queue extends Command
                 $errors ++;
             }
         }
+        $responses = $this->fastcgi->awaitResponses();
+        foreach ($responses as $response) {
+            if ((int) floor($response->getStatusCode() / 100) !== 2) {
+                $this->logger->error("Error running job", [
+                    'status' => $response->getStatusCode(),
+                    'body' => $response->getBody()->getContents(),
+                    'headers' => $response->getHeaders(),
+                ]);
+                $errors ++;
+            }
+        }
         return $errors === 0 ? Console\Command\Command::SUCCESS : Console\Command\Command::FAILURE;
     }
     protected function runJob(Console\Output\OutputInterface $output, int $job_id): int
     {
         $uri = "/api/queue/run/{$job_id}";
         $output->writeln("GET {$uri}");
-        try {
-            $response = $this->client->get($uri);
-        } catch (ClientExceptionInterface $exception) {
-            $this->logger->error($exception);
-            return Console\Command\Command::FAILURE;
-        }
-        $output->writeln("Response Code: {$response->getStatusCode()}");
 
-        return ((int) floor($response->getStatusCode() / 100) === 2) ? Console\Command\Command::SUCCESS : Console\Command\Command::FAILURE;
+        try {
+            $this->fastcgi->get($uri);
+            return self::SUCCESS;
+        } catch (FastCGIException $exception) {
+            $this->logger->error($exception->getMessage(), ['uri' => $uri, 'exception' =>$exception]);
+            return self::FAILURE;
+        }
     }
 }

cli/src/Exception/Client/FastCGI.php

@@ -0,0 +1,55 @@
+<?php
+namespace Incoviba\Exception\Client;
+
+use Throwable;
+use Psr\Http\Client\ClientExceptionInterface;
+
+
+class FastCGI implements ClientExceptionInterface
+{
+    public function __construct(protected ?Throwable $previous) {}
+
+    public function getMessage(): string
+    {
+        $message = "Could not send request";
+        if ($this->previous !== null) {
+            $message .= ": {$this->previous->getMessage()}";
+        }
+        return $message;
+    }
+
+    public function getCode()
+    {
+        return $this->previous?->getCode() ?? 500;
+    }
+
+    public function getFile(): string
+    {
+        return $this->previous?->getFile() ?? '';
+    }
+
+    public function getLine(): int
+    {
+        return $this->previous?->getLine() ?? 0;
+    }
+
+    public function getTrace(): array
+    {
+        return $this->previous?->getTrace() ?? [];
+    }
+
+    public function getTraceAsString(): string
+    {
+        return $this->previous?->getTraceAsString() ?? '';
+    }
+
+    public function getPrevious(): ?Throwable
+    {
+        return $this->previous;
+    }
+
+    public function __toString()
+    {
+        return $this->getMessage();
+    }
+}

cli/src/Service/FastCGI.php

@@ -0,0 +1,99 @@
+<?php
+namespace Incoviba\Service;
+
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\LoggerInterface;
+use hollodotme\FastCGI as FCGI;
+use Incoviba\Exception\Client\FastCGI as FastCGIException;
+
+class FastCGI implements LoggerAwareInterface
+{
+    public function __construct(protected string $hostname, protected int $port,
+                                protected string $documentRoot,
+                                protected int $connectionTimeout = 5000, protected int $readTimeout = 5000)
+    {
+        $this->client = new FCGI\Client();
+    }
+
+    public LoggerInterface $logger {
+        get {
+            return $this->logger;
+        }
+        set(LoggerInterface $logger) {
+            $this->logger = $logger;
+        }
+    }
+
+    protected FCGI\Client $client;
+    protected FCGI\Interfaces\ConfiguresSocketConnection $socket;
+    public function connect(): self
+    {
+        $this->socket = new FCGI\SocketConnections\NetworkSocket($this->hostname, $this->port, $this->connectionTimeout, $this->readTimeout);
+        return $this;
+    }
+
+    protected array $socketIds = [];
+
+    /**
+     * @throws FastCGIException
+     */
+    public function sendRequest(FCGI\Interfaces\ProvidesRequestData $request): self
+    {
+        if (!isset($this->socket)) {
+            $this->connect();
+        }
+        try {
+            $this->socketIds []= $this->client->sendAsyncRequest($this->socket, $request);
+        } catch (FCGI\Exceptions\FastCGIClientException $exception) {
+            $this->logger->error($exception->getMessage());
+            throw new FastCGIException($exception);
+        }
+        return $this;
+    }
+
+    /**
+     * @return array
+     */
+    public function awaitResponses(): array
+    {
+        $responses = [];
+        while ($this->client->hasUnhandledResponses()) {
+            try {
+                $readyResponses = $this->client->readReadyResponses(3000);
+            } catch (FCGI\Exceptions\FastCGIClientException $exception) {
+                $this->logger->error($exception->getMessage());
+                continue;
+            }
+            foreach ($readyResponses as $response) {
+                $responses []= $response;
+            }
+        }
+        return $responses;
+    }
+
+    /**
+     * @param string $uri
+     * @return FastCGI
+     * @throws FastCGIException
+     */
+    public function get(string $uri): self
+    {
+        $request = new FCGI\Requests\GetRequest($this->documentRoot, '');
+        $request->setRequestUri($uri);
+        return $this->sendRequest($request);
+    }
+
+    /**
+     * @param string $uri
+     * @param ?array $data
+     * @return FastCGI
+     * @throws FastCGIException
+     */
+    public function post(string $uri, ?array $data): self
+    {
+        $content = new FCGI\RequestContents\JsonData($data ?? []);
+        $request = FCGI\Requests\PostRequest::newWithRequestContent($this->documentRoot, $content);
+        $request->setRequestUri($uri);
+        return $this->sendRequest($request);
+    }
+}