Merge pull request #7 from lamoda/feature/SCENTRE-5744

SCENTRE-5744 Handle queue in "in progress" status, some optimization
lamoda · Nov 7, 2019 · 0c22cfc · 0c22cfc
2 parents 202a596 + 817d4cd
commit 0c22cfc
Show file tree

Hide file tree

Showing 5 changed files with 55 additions and 83 deletions.
diff --git a/src/ConstantMessage.php b/src/ConstantMessage.php
@@ -20,10 +20,10 @@ class ConstantMessage
 
     public const PUBLISHER_NOT_FOUND = 'Publisher `%s` not found';
 
-    public const QUEUE_ENTITY_NOT_FOUND = 'The queue with id "%d" was not found';
-    public const QUEUE_ENTITY_NOT_FOUND_IN_STATUS_NEW = 'The queue "%s" with job "%s" was not found in status "new". Actual status is "%s"';
-    public const QUEUE_ATTEMPTS_REACHED = 'The queue "%s" has reached it\'s attempts count maximum';
-    public const QUEUE_CAN_NOT_REQUEUE = 'Can not requeue messages';
-    public const QUEUE_CAN_NOT_REPUBLISH = 'Can not republish messages';
-    public const QUEUE_SUCCESS_REPUBLISH = 'Queue republish successfully completed';
+    public const QUEUE_ENTITY_NOT_FOUND                    = 'The queue with id "%d" was not found';
+    public const QUEUE_ENTITY_NOT_FOUND_IN_SUITABLE_STATUS = 'The queue "%s" with job "%s" was not found in suitable status. Actual status is "%s"';
+    public const QUEUE_ATTEMPTS_REACHED                    = 'The queue "%s" has reached it\'s attempts count maximum';
+    public const QUEUE_CAN_NOT_REQUEUE                     = 'Can not requeue messages';
+    public const QUEUE_CAN_NOT_REPUBLISH                   = 'Can not republish messages';
+    public const QUEUE_SUCCESS_REPUBLISH                   = 'Queue republish successfully completed';
 }
diff --git a/src/Consumer.php b/src/Consumer.php
@@ -80,11 +80,7 @@ public function execute(AMQPMessage $message): int
             return $this->doExecute(
                 $this->queueService->getToProcess($data['id'])
             );
-        } catch (UnexpectedValueException $exception) {
-            $this->logger->alert($exception->getMessage(), $this->getMessageLogParams($message));
-
-            return self::MSG_REJECT;
-        } catch (AttemptsReachedException $exception) {
+        } catch (UnexpectedValueException | AttemptsReachedException $exception) {
             $this->logger->alert($exception->getMessage(), $this->getMessageLogParams($message));
 
             return self::MSG_REJECT;

diff --git a/src/Publisher.php b/src/Publisher.php
@@ -117,8 +117,6 @@ private function releaseQueues(array $queues): void
                         'trace' => $e->getTraceAsString(),
                     ]
                 );
-
-                continue;
             }
         }
     }

diff --git a/src/Service/QueueService.php b/src/Service/QueueService.php
@@ -30,6 +30,12 @@ class QueueService
     /** @var int */
     protected $maxAttempts;
 
+    /** @var array */
+    protected $queueSuitableStatuses = [
+        QueueEntityMappedSuperclass::STATUS_NEW,
+        QueueEntityMappedSuperclass::STATUS_IN_PROGRESS
+    ];
+
     public function __construct(
         QueueRepository $repository,
         EntityFactoryInterface $entityFactory,
@@ -64,45 +70,33 @@ public function getToRestore(int $limit, ?int $offset = null): array
      */
     public function getToProcess(int $id): QueueEntityInterface
     {
-        $this->repository->beginTransaction();
+        $queueEntity = $this->repository->find($id);
 
-        try {
-            $queueEntity = $this->repository->findOneBy(
-                [
-                    'id' => $id,
-                ]
-            );
-
-            if (!($queueEntity instanceof QueueEntityInterface)) {
-                throw new UnexpectedValueException(sprintf(ConstantMessage::QUEUE_ENTITY_NOT_FOUND, $id));
-            }
-
-            if (QueueEntityMappedSuperclass::STATUS_NEW !== $queueEntity->getStatus()) {
-                throw new UnexpectedValueException(sprintf(
-                    ConstantMessage::QUEUE_ENTITY_NOT_FOUND_IN_STATUS_NEW,
-                    $queueEntity->getName(),
-                    $queueEntity->getJobName(),
-                    $queueEntity->getStatusAsString()
-                ));
-            }
-
-            $attemptsReached = $queueEntity->isMaxAttemptsReached($this->maxAttempts);
-            if ($attemptsReached) {
-                $queueEntity->setAttemptsReached();
-            } else {
-                $queueEntity->setInProgress();
-            }
-
-            $this->repository->save($queueEntity);
-            $this->repository->commit();
-        } catch (Exception $exception) {
-            $this->repository->rollback();
-
-            throw $exception;
+        if (!($queueEntity instanceof QueueEntityInterface)) {
+            throw new UnexpectedValueException(sprintf(ConstantMessage::QUEUE_ENTITY_NOT_FOUND, $id));
         }
 
+        if (!in_array($queueEntity->getStatus(), $this->queueSuitableStatuses, true)) {
+            throw new UnexpectedValueException(sprintf(
+                ConstantMessage::QUEUE_ENTITY_NOT_FOUND_IN_SUITABLE_STATUS,
+                $queueEntity->getName(),
+                $queueEntity->getJobName(),
+                $queueEntity->getStatusAsString()
+            ));
+        }
+
+        $attemptsReached = $queueEntity->isMaxAttemptsReached($this->maxAttempts);
+        if ($attemptsReached) {
+            $queueEntity->setAttemptsReached();
+        } else {
+            $queueEntity->setInProgress();
+        }
+
+        $this->repository->save($queueEntity);
+
         if ($attemptsReached) {
             $this->eventDispatcher->dispatch(QueueAttemptsReachedEvent::NAME, new QueueAttemptsReachedEvent($queueEntity));
+
             throw new AttemptsReachedException(sprintf(ConstantMessage::QUEUE_ATTEMPTS_REACHED, $queueEntity->getName()));
         }
 
@@ -131,7 +125,7 @@ public function getToRepublish(int $limit, ?int $offset = null): array
      */
     public function createQueue(QueueInterface $queueable): QueueEntityInterface
     {
-        return $this->save($this->entityFactory->createQueue($queueable));
+        return $this->entityFactory->createQueue($queueable);
     }
 
     public function flush(QueueEntityInterface $entity = null): void

diff --git a/tests/unit/Service/QueueServiceTest.php b/tests/unit/Service/QueueServiceTest.php
@@ -42,33 +42,36 @@ public function testGetToRestore(): void
     }
 
     /**
+     * @param QueueEntity $queue
+     *
      * @throws \Exception
+     *
+     * @dataProvider dataGetToProcess()
      */
-    public function testGetToProcess(): void
+    public function testGetToProcess(QueueEntity $queue): void
     {
-        $queue = new QueueEntity('queue', 'exchange', 'ClassJob', ['id' => 1]);
-        $queue->setNew();
-
         $queueRepository = $this->getQueueRepository();
         $queueRepository
             ->expects($this->once())
-            ->method('beginTransaction');
-        $queueRepository
-            ->expects($this->once())
-            ->method('findOneBy')
+            ->method('find')
             ->willReturn($queue);
         $queueRepository
             ->expects($this->once())
             ->method('save')
             ->with($queue);
-        $queueRepository
-            ->expects($this->once())
-            ->method('commit');
 
         $this->assertEquals($queue, $this->createService($queueRepository)->getToProcess(1));
         $this->assertEquals(QueueEntity::STATUS_IN_PROGRESS_TITLE, $queue->getStatusAsString());
     }
 
+    public function dataGetToProcess(): array
+    {
+        return [
+            'new status' => [(new QueueEntity('queue', 'exchange', 'ClassJob', ['id' => 1]))->setNew()],
+            'in progress status' => [(new QueueEntity('queue', 'exchange', 'ClassJob', ['id' => 1]))->setInProgress()],
+        ];
+    }
+
     /**
      * @param null | QueueEntity $queueEntity
      * @param string $expectedExceptionMessage
@@ -83,16 +86,11 @@ public function testGetToProcessQueueNotFound(?QueueEntity $queueEntity, string
         $this->expectExceptionMessage($expectedExceptionMessage);
 
         $queueRepository = $this->getQueueRepository();
+
         $queueRepository
             ->expects($this->once())
-            ->method('beginTransaction');
-        $queueRepository
-            ->expects($this->once())
-            ->method('findOneBy')
+            ->method('find')
             ->willReturn($queueEntity);
-        $queueRepository
-            ->expects($this->once())
-            ->method('rollback');
 
         $this->createService($queueRepository)->getToProcess(1);
     }
@@ -106,7 +104,7 @@ public function dataGetToProcessQueueNotFound(): array
             ],
             'Status not NEW' => [
                 new QueueEntity('queue', 'exchange', 'ClassJob', ['id' => 1]),
-                'The queue "queue" with job "ClassJob" was not found in status "new". Actual status is "initial"',
+                'The queue "queue" with job "ClassJob" was not found in suitable status. Actual status is "initial"',
             ],
         ];
     }
@@ -126,18 +124,12 @@ public function testGetToProcessAttemptsReached(): void
         $queueRepository = $this->getQueueRepository();
         $queueRepository
             ->expects($this->once())
-            ->method('beginTransaction');
-        $queueRepository
-            ->expects($this->once())
-            ->method('findOneBy')
+            ->method('find')
             ->willReturn($queue);
         $queueRepository
             ->expects($this->once())
             ->method('save')
             ->with($queue);
-        $queueRepository
-            ->expects($this->once())
-            ->method('commit');
 
         $attemptsReachedEvent = new QueueAttemptsReachedEvent($queue);
 
@@ -184,11 +176,6 @@ public function testCreateQueue(): void
             ->willReturn($queue);
 
         $queueRepository = $this->getQueueRepository();
-        $queueRepository
-            ->expects($this->once())
-            ->method('save')
-            ->with($queue)
-            ->willReturnArgument(0);
 
         $this->assertEquals(
             $queue,
@@ -232,10 +219,7 @@ private function getQueueRepository()
     {
         return $this->getMockQueueRepository(
             [
-                'beginTransaction',
-                'rollback',
-                'commit',
-                'findOneBy',
+                'find',
                 'getToRestore',
                 'save',
                 'isTransactionActive',