Exceptions for json_decode

src/Consumer.php

@@ -63,9 +63,14 @@ public function __construct(
      */
     public function execute(AMQPMessage $message): int
     {
-        $data = json_decode($message->body, true);
-
         try {
+            $data = json_decode($message->body, true);
+            $jsonErrorCode = json_last_error();
+
+            if (JSON_ERROR_NONE !== $jsonErrorCode) {
+                throw new UnexpectedValueException('json_decode error: ' . json_last_error_msg(), $jsonErrorCode);
+            }
+
             if (!isset($data['id'])) {
                 throw new UnexpectedValueException(ConstantMessage::AMQP_DATA_DAMAGED);
             }

src/Factory/EntityFactory.php

@@ -6,6 +6,7 @@
 
 use JMS\Serializer\SerializerInterface;
 use Lamoda\QueueBundle\Entity\QueueEntityInterface;
+use Lamoda\QueueBundle\Exception\UnexpectedValueException;
 use Lamoda\QueueBundle\QueueInterface;
 
 class EntityFactory implements EntityFactoryInterface
@@ -22,15 +23,29 @@ public function __construct(SerializerInterface $serializer, string $entityClass
         $this->entityClass = $entityClass;
     }
 
+    /**
+     * @param QueueInterface $queueable
+     *
+     * @throws UnexpectedValueException
+     *
+     * @return QueueEntityInterface
+     */
     public function createQueue(QueueInterface $queueable): QueueEntityInterface
     {
         $name = $queueable->getQueue();
         $exchange = $queueable->getExchange();
         $jobName = get_class($queueable);
         $data = $this->serializer->serialize($queueable, 'json');
 
+        $data = json_decode($data, true);
+        $jsonErrorCode = json_last_error();
+
+        if (JSON_ERROR_NONE !== $jsonErrorCode) {
+            throw new UnexpectedValueException('json_decode error: ' . json_last_error_msg(), $jsonErrorCode);
+        }
+
         /** @var QueueEntityInterface $queueEntity */
-        $queueEntity = new $this->entityClass($name, $exchange, $jobName, json_decode($data, true));
+        $queueEntity = new $this->entityClass($name, $exchange, $jobName, $data);
 
         if (null !== $queueable->getScheduleAt()) {
             $queueEntity->setScheduled($queueable->getScheduleAt());

src/Factory/PublisherFactory.php

@@ -7,6 +7,7 @@
 use Lamoda\QueueBundle\ConstantMessage;
 use Lamoda\QueueBundle\Entity\QueueEntityInterface;
 use Lamoda\QueueBundle\Exception\RuntimeException;
+use Lamoda\QueueBundle\Exception\UnexpectedValueException;
 use Lamoda\QueueBundle\Publisher;
 use Lamoda\QueueBundle\QueueInterface;
 use Lamoda\QueueBundle\Service\DelayService;
@@ -67,6 +68,11 @@ public function get(string $exchangeName): Publisher
         return $this->publishers[$exchangeName];
     }
 
+    /**
+     * @param QueueInterface $queueable
+     *
+     * @throws UnexpectedValueException
+     */
     public function publish(QueueInterface $queueable): void
     {
         $exchangeName = $queueable->getExchange();

src/Publisher.php

@@ -5,6 +5,7 @@
 namespace Lamoda\QueueBundle;
 
 use Lamoda\QueueBundle\Entity\QueueEntityInterface;
+use Lamoda\QueueBundle\Exception\UnexpectedValueException;
 use Lamoda\QueueBundle\Service\DelayService;
 use Lamoda\QueueBundle\Service\QueueService;
 use OldSound\RabbitMqBundle\RabbitMq\Producer;
@@ -40,6 +41,13 @@ public function __construct(
         $this->delayService = $delayService;
     }
 
+    /**
+     * @param QueueInterface $queueable
+     *
+     * @throws UnexpectedValueException
+     *
+     * @return Publisher
+     */
     public function prepareJobForPublish(QueueInterface $queueable): self
     {
         $this->prepareQueueForPublish(

src/Service/QueueService.php

@@ -122,6 +122,13 @@ public function getToRepublish(int $limit, ?int $offset = null): array
         return $this->repository->getToRepublish($limit, $offset);
     }
 
+    /**
+     * @param QueueInterface $queueable
+     *
+     * @throws UnexpectedValueException
+     *
+     * @return QueueEntityInterface
+     */
     public function createQueue(QueueInterface $queueable): QueueEntityInterface
     {
         return $this->save($this->entityFactory->createQueue($queueable));

tests/unit/ConsumerTest.php

@@ -150,10 +150,17 @@ public function testExecute(QueueEntityInterface $queueEntity, AMQPMessage $mess
         $this->assertEquals($result, $mockConsumer->execute($message));
     }
 
-    public function dataExecuteBrokenMessage(): array
+    public function dataExecute(): array
     {
+        $id = 1;
+
         return [
-            [$this->getMessage(['name' => 'name'])],
+            [
+                $this->getQueue(),
+                $this->getMessage(json_encode(['id' => $id])),
+                Consumer::MSG_ACK,
+                $id,
+            ],
         ];
     }
 
@@ -164,11 +171,12 @@ public function dataExecuteBrokenMessage(): array
      *
      * @dataProvider dataExecuteBrokenMessage()
      */
-    public function testExecuteBrokenMessage(AMQPMessage $message): void
+    public function testExecuteBrokenMessage(AMQPMessage $message, string $expectedErrorMessage): void
     {
         $mockLogger = $this->getMockLogger(['alert']);
         $mockLogger->expects($this->once())
-            ->method('alert');
+            ->method('alert')
+            ->with($expectedErrorMessage);
 
         $mockConsumer = $this->getMockConsumer(['doExecute']);
 
@@ -179,16 +187,16 @@ public function testExecuteBrokenMessage(AMQPMessage $message): void
         $this->assertEquals($mockConsumer::MSG_REJECT, $mockConsumer->execute($message));
     }
 
-    public function dataExecute(): array
+    public function dataExecuteBrokenMessage(): array
     {
-        $id = 1;
-
         return [
             [
-                $this->getQueue(),
-                $this->getMessage(['id' => $id]),
-                Consumer::MSG_ACK,
-                $id,
+                $this->getMessage(json_encode(['name' => 'name'])),
+                'Data was damaged. Remove message from queue',
+            ],
+            [
+                $this->getMessage('abrakadabra'),
+                'json_decode error: Syntax error',
             ],
         ];
     }
@@ -255,7 +263,7 @@ protected function getQueue(): QueueEntityInterface
         return $queue;
     }
 
-    protected function getMessage(array $data = ['id' => 1]): AMQPMessage
+    protected function getMessage(string $data = '{"id": 1}'): AMQPMessage
     {
         /** @var \PhpAmqpLib\Channel\AMQPChannel | \PHPUnit_Framework_MockObject_MockObject $channelMock */
         $channelMock = $this->getMockBuilder(AMQPChannel::class)
@@ -272,7 +280,7 @@ protected function getMessage(array $data = ['id' => 1]): AMQPMessage
             'channel' => $channelMock,
             'delivery_tag' => uniqid(),
         ];
-        $message->setBody(json_encode($data));
+        $message->setBody($data);
 
         return $message;
     }

tests/unit/Factory/EntityFactoryTest.php

@@ -6,6 +6,7 @@
 
 use JMS\Serializer\SerializerInterface;
 use Lamoda\QueueBundle\Entity\QueueEntityInterface;
+use Lamoda\QueueBundle\Exception\UnexpectedValueException;
 use Lamoda\QueueBundle\Factory\EntityFactory;
 use Lamoda\QueueBundle\Job\AbstractJob;
 use Lamoda\QueueBundle\Tests\Unit\Job\StubJob;
@@ -59,6 +60,21 @@ public function dataCreateQueue(): array
         ];
     }
 
+    public function testJsonDecodeError(): void
+    {
+        $this->expectException(UnexpectedValueException::class);
+        $this->expectExceptionCode(4);
+        $this->expectExceptionMessage('json_decode error: Syntax error');
+
+        $serializer = $this->getJMSSerializer(['serialize']);
+        $serializer->expects($this->once())
+            ->method('serialize')
+            ->willReturn('abrakadabra');
+
+        $factory = $this->createFactory($serializer);
+        $factory->createQueue(new StubJob(1));
+    }
+
     private function createFactory(SerializerInterface $serializer): EntityFactory
     {
         return new EntityFactory($serializer, QueueEntity::class);