Migrate pheanstalk transport

Author: makasim

pkg/pheanstalk/PheanstalkConnectionFactory.php

@@ -3,6 +3,7 @@
 namespace Enqueue\Pheanstalk;
 
 use Interop\Queue\PsrConnectionFactory;
+use Interop\Queue\PsrContext;
 use Pheanstalk\Pheanstalk;
 
 class PheanstalkConnectionFactory implements PsrConnectionFactory
@@ -49,19 +50,14 @@ public function __construct($config = 'beanstalk:')
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @return PheanstalkContext
      */
-    public function createContext()
+    public function createContext(): PsrContext
     {
         return new PheanstalkContext($this->establishConnection());
     }
 
-    /**
-     * @return Pheanstalk
-     */
-    private function establishConnection()
+    private function establishConnection(): Pheanstalk
     {
         if (false == $this->connection) {
             $this->connection = new Pheanstalk(
@@ -75,12 +71,7 @@ private function establishConnection()
         return $this->connection;
     }
 
-    /**
-     * @param string $dsn
-     *
-     * @return array
-     */
-    private function parseDsn($dsn)
+    private function parseDsn(string $dsn): array
     {
         $dsnConfig = parse_url($dsn);
         if (false === $dsnConfig) {
@@ -112,10 +103,7 @@ private function parseDsn($dsn)
         ]);
     }
 
-    /**
-     * @return array
-     */
-    private function defaultConfig()
+    private function defaultConfig(): array
     {
         return [
             'host' => 'localhost',

pkg/pheanstalk/PheanstalkConsumer.php

@@ -5,6 +5,7 @@
 use Interop\Queue\InvalidMessageException;
 use Interop\Queue\PsrConsumer;
 use Interop\Queue\PsrMessage;
+use Interop\Queue\PsrQueue;
 use Pheanstalk\Job;
 use Pheanstalk\Pheanstalk;
 
@@ -20,32 +21,24 @@ class PheanstalkConsumer implements PsrConsumer
      */
     private $pheanstalk;
 
-    /**
-     * @param PheanstalkDestination $destination
-     * @param Pheanstalk            $pheanstalk
-     */
     public function __construct(PheanstalkDestination $destination, Pheanstalk $pheanstalk)
     {
         $this->destination = $destination;
         $this->pheanstalk = $pheanstalk;
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @return PheanstalkDestination
      */
-    public function getQueue()
+    public function getQueue(): PsrQueue
     {
         return $this->destination;
     }
 
     /**
-     * {@inheritdoc}
-     *
-     * @return PheanstalkMessage|null
+     * @return PheanstalkMessage
      */
-    public function receive($timeout = 0)
+    public function receive(int $timeout = 0): ?PsrMessage
     {
         if (0 === $timeout) {
             while (true) {
@@ -58,26 +51,26 @@ public function receive($timeout = 0)
                 return $this->convertJobToMessage($job);
             }
         }
+
+        return null;
     }
 
     /**
-     * {@inheritdoc}
-     *
-     * @return PheanstalkMessage|null
+     * @return PheanstalkMessage
      */
-    public function receiveNoWait()
+    public function receiveNoWait(): ?PsrMessage
     {
         if ($job = $this->pheanstalk->reserveFromTube($this->destination->getName(), 0)) {
             return $this->convertJobToMessage($job);
         }
+
+        return null;
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @param PheanstalkMessage $message
      */
-    public function acknowledge(PsrMessage $message)
+    public function acknowledge(PsrMessage $message): void
     {
         InvalidMessageException::assertMessageInstanceOf($message, PheanstalkMessage::class);
 
@@ -89,11 +82,9 @@ public function acknowledge(PsrMessage $message)
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @param PheanstalkMessage $message
      */
-    public function reject(PsrMessage $message, $requeue = false)
+    public function reject(PsrMessage $message, bool $requeue = false): void
     {
         $this->acknowledge($message);
 
@@ -102,12 +93,7 @@ public function reject(PsrMessage $message, $requeue = false)
         }
     }
 
-    /**
-     * @param Job $job
-     *
-     * @return PheanstalkMessage
-     */
-    private function convertJobToMessage(Job $job)
+    private function convertJobToMessage(Job $job): PheanstalkMessage
     {
         $stats = $this->pheanstalk->statsJob($job);
 

pkg/pheanstalk/PheanstalkContext.php

@@ -3,8 +3,17 @@
 namespace Enqueue\Pheanstalk;
 
 use Interop\Queue\InvalidDestinationException;
+use Interop\Queue\PsrConsumer;
 use Interop\Queue\PsrContext;
 use Interop\Queue\PsrDestination;
+use Interop\Queue\PsrMessage;
+use Interop\Queue\PsrProducer;
+use Interop\Queue\PsrQueue;
+use Interop\Queue\PsrSubscriptionConsumer;
+use Interop\Queue\PsrTopic;
+use Interop\Queue\PurgeQueueNotSupportedException;
+use Interop\Queue\SubscriptionConsumerNotSupportedException;
+use Interop\Queue\TemporaryQueueNotSupportedException;
 use Pheanstalk\Pheanstalk;
 
 class PheanstalkContext implements PsrContext
@@ -14,79 +23,76 @@ class PheanstalkContext implements PsrContext
      */
     private $pheanstalk;
 
-    /**
-     * @param Pheanstalk $pheanstalk
-     */
     public function __construct(Pheanstalk $pheanstalk)
     {
         $this->pheanstalk = $pheanstalk;
     }
 
     /**
-     * {@inheritdoc}
+     * @return PheanstalkMessage
      */
-    public function createMessage($body = '', array $properties = [], array $headers = [])
+    public function createMessage(string $body = '', array $properties = [], array $headers = []): PsrMessage
     {
         return new PheanstalkMessage($body, $properties, $headers);
     }
 
     /**
-     * {@inheritdoc}
+     * @return PheanstalkDestination
      */
-    public function createTopic($topicName)
+    public function createTopic(string $topicName): PsrTopic
     {
         return new PheanstalkDestination($topicName);
     }
 
     /**
-     * {@inheritdoc}
+     * @return PheanstalkDestination
      */
-    public function createQueue($queueName)
+    public function createQueue(string $queueName): PsrQueue
     {
         return new PheanstalkDestination($queueName);
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function createTemporaryQueue()
+    public function createTemporaryQueue(): PsrQueue
     {
-        throw new \LogicException('Not implemented');
+        throw TemporaryQueueNotSupportedException::providerDoestNotSupportIt();
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @return PheanstalkProducer
      */
-    public function createProducer()
+    public function createProducer(): PsrProducer
     {
         return new PheanstalkProducer($this->pheanstalk);
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @param PheanstalkDestination $destination
      *
      * @return PheanstalkConsumer
      */
-    public function createConsumer(PsrDestination $destination)
+    public function createConsumer(PsrDestination $destination): PsrConsumer
     {
         InvalidDestinationException::assertDestinationInstanceOf($destination, PheanstalkDestination::class);
 
         return new PheanstalkConsumer($destination, $this->pheanstalk);
     }
 
-    public function close()
+    public function close(): void
     {
         $this->pheanstalk->getConnection()->disconnect();
     }
 
-    /**
-     * @return Pheanstalk
-     */
-    public function getPheanstalk()
+    public function createSubscriptionConsumer(): PsrSubscriptionConsumer
+    {
+        throw SubscriptionConsumerNotSupportedException::providerDoestNotSupportIt();
+    }
+
+    public function purgeQueue(PsrQueue $queue): void
+    {
+        throw PurgeQueueNotSupportedException::providerDoestNotSupportIt();
+    }
+
+    public function getPheanstalk(): Pheanstalk
     {
         return $this->pheanstalk;
     }

pkg/pheanstalk/PheanstalkDestination.php

@@ -12,35 +12,23 @@ class PheanstalkDestination implements PsrQueue, PsrTopic
      */
     private $destinationName;
 
-    /**
-     * @param string $destinationName
-     */
-    public function __construct($destinationName)
+    public function __construct(string $destinationName)
     {
         $this->destinationName = $destinationName;
     }
 
-    /**
-     * @return string
-     */
-    public function getName()
+    public function getName(): string
     {
         return $this->destinationName;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getQueueName()
+    public function getQueueName(): string
     {
-        return $this->getName();
+        return $this->destinationName;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getTopicName()
+    public function getTopicName(): string
     {
-        return $this->getName();
+        return $this->destinationName;
     }
 }