add files

Author: makasim

PhpFpmConnectionFactory.php

@@ -0,0 +1,36 @@
+<?php
+namespace Makasim\PhpFpm;
+
+use Enqueue\Dsn\Dsn;
+use hollodotme\FastCGI\Client;
+use hollodotme\FastCGI\SocketConnections\NetworkSocket;
+use hollodotme\FastCGI\SocketConnections\UnixDomainSocket;
+use Interop\Queue\ConnectionFactory;
+use Interop\Queue\Context;
+
+class PhpFpmConnectionFactory implements ConnectionFactory
+{
+    /**
+     * @var string
+     */
+    private $dsn;
+
+    public function __construct(string $dsn)
+    {
+        $this->dsn = $dsn;
+    }
+
+    public function createContext(): Context
+    {
+        $dsn = new Dsn($this->dsn);
+        if ('unix' == $dsn->getSchemeProtocol()) {
+            $socket = new UnixDomainSocket($dsn->getPath());
+        } else if ('tcp' == $dsn->getSchemeProtocol()) {
+            $socket = new NetworkSocket($dsn->getHost(), $dsn->getPort());
+        } else {
+            throw new \LogicException('Protocol is not supported');
+        }
+
+        return new PhpFpmContext(new Client($socket));
+    }
+}

PhpFpmConsumer.php

@@ -0,0 +1,46 @@
+<?php
+namespace Makasim\PhpFpm;
+
+use Interop\Queue\Consumer;
+use Interop\Queue\Message;
+use Interop\Queue\Queue;
+
+class PhpFpmConsumer implements Consumer
+{
+    /**
+     * @var PhpFpmDestination
+     */
+    private $destination;
+
+    public function __construct(PhpFpmDestination $destination)
+    {
+        $this->destination = $destination;
+    }
+
+    public function getQueue(): Queue
+    {
+        return $this->destination;
+    }
+
+    public function receive(int $timeout = 0): ?Message
+    {
+        return $this->receiveNoWait();
+    }
+
+    public function receiveNoWait(): ?Message
+    {
+        if ($_POST['message']) {
+            return PhpFpmMessage::jsonUnserialize($_POST['message']);
+        }
+
+        return null;
+    }
+
+    public function acknowledge(Message $message): void
+    {
+    }
+
+    public function reject(Message $message, bool $requeue = false): void
+    {
+    }
+}

PhpFpmContext.php

@@ -0,0 +1,73 @@
+<?php
+namespace Makasim\PhpFpm;
+
+use hollodotme\FastCGI\Client;
+use Interop\Queue\ConnectionFactory;
+use Interop\Queue\Consumer;
+use Interop\Queue\Context;
+use Interop\Queue\Destination;
+use Interop\Queue\Exception\PurgeQueueNotSupportedException;
+use Interop\Queue\Exception\SubscriptionConsumerNotSupportedException;
+use Interop\Queue\Exception\TemporaryQueueNotSupportedException;
+use Interop\Queue\Message;
+use Interop\Queue\Producer;
+use Interop\Queue\Queue;
+use Interop\Queue\SubscriptionConsumer;
+use Interop\Queue\Topic;
+
+class PhpFpmContext implements Context
+{
+    /**
+     * @var Client
+     */
+    private $cgiClient;
+
+    public function __construct(Client $cgiClient)
+    {
+        $this->cgiClient = $cgiClient;
+    }
+
+    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    {
+        return new PhpFpmMessage($body, $properties, $headers);
+    }
+
+    public function createTopic(string $topicName): Topic
+    {
+        return new PhpFpmDestination($topicName);
+    }
+
+    public function createQueue(string $queueName): Queue
+    {
+        return new PhpFpmDestination($queueName);
+    }
+
+    public function createProducer(): Producer
+    {
+        return new PhpFpmProducer($this->cgiClient);
+    }
+
+    public function createConsumer(Destination $destination): Consumer
+    {
+        return new PhpFpmConsumer($destination);
+    }
+
+    public function close(): void
+    {
+    }
+
+    public function createTemporaryQueue(): Queue
+    {
+        throw TemporaryQueueNotSupportedException::providerDoestNotSupportIt();
+    }
+
+    public function createSubscriptionConsumer(): SubscriptionConsumer
+    {
+        throw SubscriptionConsumerNotSupportedException::providerDoestNotSupportIt();
+    }
+
+    public function purgeQueue(Queue $queue): void
+    {
+        throw PurgeQueueNotSupportedException::providerDoestNotSupportIt();
+    }
+}

PhpFpmDestination.php

@@ -0,0 +1,35 @@
+<?php
+namespace Makasim\PhpFpm;
+
+use Interop\Queue\ConnectionFactory;
+use Interop\Queue\Context;
+use Interop\Queue\Queue;
+use Interop\Queue\Topic;
+
+class PhpFpmDestination implements Queue, Topic
+{
+    /**
+     * @var string
+     */
+    private $name;
+
+    public function __construct(string $name)
+    {
+        $this->name = $name;
+    }
+
+    public function getName(): string
+    {
+        return $this->name;
+    }
+
+    public function getQueueName(): string
+    {
+        return $this->getName();
+    }
+
+    public function getTopicName(): string
+    {
+        return $this->getName();
+    }
+}

PhpFpmMessage.php

@@ -0,0 +1,42 @@
+<?php
+namespace Makasim\PhpFpm;
+
+use Interop\Queue\Message;
+use Interop\Queue\MessageTrait;
+
+class PhpFpmMessage implements Message, \JsonSerializable
+{
+    public function __construct(string $body = '', array $properties = [], array $headers = [])
+    {
+        $this->body = $body;
+        $this->properties = $properties;
+        $this->headers = $headers;
+
+        $this->redelivered = false;
+    }
+
+    use MessageTrait;
+
+    public function jsonSerialize(): array
+    {
+        return [
+            'body' => $this->getBody(),
+            'properties' => $this->getProperties(),
+            'headers' => $this->getHeaders(),
+        ];
+    }
+
+    public static function jsonUnserialize(string $json): self
+    {
+        $data = json_decode($json, true);
+        if (JSON_ERROR_NONE !== json_last_error()) {
+            throw new \InvalidArgumentException(sprintf(
+                'The malformed json given. Error %s and message %s',
+                json_last_error(),
+                json_last_error_msg()
+            ));
+        }
+
+        return new self($data['body'], $data['properties'], $data['headers']);
+    }
+}

PhpFpmProducer.php

@@ -0,0 +1,82 @@
+<?php
+namespace Makasim\PhpFpm;
+
+use hollodotme\FastCGI\Client;
+use hollodotme\FastCGI\Requests\PostRequest;
+use Interop\Queue\Destination;
+use Interop\Queue\Exception\DeliveryDelayNotSupportedException;
+use Interop\Queue\Exception\Exception;
+use Interop\Queue\Exception\PriorityNotSupportedException;
+use Interop\Queue\Exception\TimeToLiveNotSupportedException;
+use Interop\Queue\Message;
+use Interop\Queue\Producer;
+
+class PhpFpmProducer implements Producer
+{
+    /**
+     * @var Client
+     */
+    private $cgiClient;
+
+    public function __construct(Client $cgiClient)
+    {
+        $this->cgiClient = $cgiClient;
+    }
+
+    /**
+     * @param PhpFpmDestination $destination
+     * @param PhpFpmMessage $message
+     */
+    public function send(Destination $destination, Message $message): void
+    {
+        $request = new PostRequest($destination->getName(), http_build_query(['message' => json_encode($message)]));
+
+        try {
+            $this->cgiClient->sendAsyncRequest($request);
+        } catch (\Exception $e) {
+            throw new Exception($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    public function setDeliveryDelay(int $deliveryDelay = null): Producer
+    {
+        if (null !== $deliveryDelay) {
+            throw DeliveryDelayNotSupportedException::providerDoestNotSupportIt();
+        }
+
+        return $this;
+    }
+
+    public function getDeliveryDelay(): ?int
+    {
+        return null;
+    }
+
+    public function setPriority(int $priority = null): Producer
+    {
+        if (null !== $priority) {
+            throw PriorityNotSupportedException::providerDoestNotSupportIt();
+        }
+
+        return $this;
+    }
+
+    public function getPriority(): ?int
+    {
+        return null;
+    }
+
+    public function setTimeToLive(int $timeToLive = null): Producer
+    {
+        if (null !== $timeToLive) {
+            throw TimeToLiveNotSupportedException::providerDoestNotSupportIt();
+        }
+
+        return $this;
+    }
+
+    public function getTimeToLive(): ?int
+    {
+        return null;
+    }
+}