src/Wallabag/ImportBundle/Consumer/AbstractConsumer.php

   1 <?php
   2
   3 namespace Wallabag\ImportBundle\Consumer;
   4
   5 use Doctrine\ORM\EntityManager;
   6 use Psr\Log\LoggerInterface;
   7 use Psr\Log\NullLogger;
   8 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
   9 use Wallabag\CoreBundle\Entity\Entry;
  10 use Wallabag\CoreBundle\Entity\Tag;
  11 use Wallabag\CoreBundle\Event\EntrySavedEvent;
  12 use Wallabag\ImportBundle\Import\AbstractImport;
  13 use Wallabag\UserBundle\Repository\UserRepository;
  14
  15 abstract class AbstractConsumer
  16 {
  17     protected $em;
  18     protected $userRepository;
  19     protected $import;
  20     protected $eventDispatcher;
  21     protected $logger;
  22
  23     public function __construct(EntityManager $em, UserRepository $userRepository, AbstractImport $import, EventDispatcherInterface $eventDispatcher, LoggerInterface $logger = null)
  24     {
  25         $this->em = $em;
  26         $this->userRepository = $userRepository;
  27         $this->import = $import;
  28         $this->eventDispatcher = $eventDispatcher;
  29         $this->logger = $logger ?: new NullLogger();
  30     }
  31
  32     /**
  33      * Handle a message and save it.
  34      *
  35      * @param string $body Message from the queue (in json)
  36      *
  37      * @return bool
  38      */
  39     protected function handleMessage($body)
  40     {
  41         $storedEntry = json_decode($body, true);
  42
  43         $user = $this->userRepository->find($storedEntry['userId']);
  44
  45         // no user? Drop message
  46         if (null === $user) {
  47             $this->logger->warning('Unable to retrieve user', ['entry' => $storedEntry]);
  48
  49             // return true to skip message
  50             return true;
  51         }
  52
  53         $this->import->setUser($user);
  54
  55         if (false === $this->import->validateEntry($storedEntry)) {
  56             $this->logger->warning('Entry is invalid', ['entry' => $storedEntry]);
  57
  58             // return true to skip message
  59             return true;
  60         }
  61
  62         $entry = $this->import->parseEntry($storedEntry);
  63
  64         if (null === $entry) {
  65             $this->logger->warning('Entry already exists', ['entry' => $storedEntry]);
  66
  67             // return true to skip message
  68             return true;
  69         }
  70
  71         try {
  72             $this->em->flush();
  73
  74             // entry saved, dispatch event about it!
  75             $this->eventDispatcher->dispatch(EntrySavedEvent::NAME, new EntrySavedEvent($entry));
  76
  77             // clear only affected entities
  78             $this->em->clear(Entry::class);
  79             $this->em->clear(Tag::class);
  80         } catch (\Exception $e) {
  81             $this->logger->warning('Unable to save entry', ['entry' => $storedEntry, 'exception' => $e]);
  82
  83             return false;
  84         }
  85
  86         $this->logger->info('Content with url imported! (' . $entry->getUrl() . ')');
  87
  88         return true;
  89     }
  90 }