Move source files directly under src/ directory

src/Consumer/AMQPEntryConsumer.php

@@ -0,0 +1,14 @@
+<?php
+
+namespace Wallabag\CoreBundle\Consumer;
+
+use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface;
+use PhpAmqpLib\Message\AMQPMessage;
+
+class AMQPEntryConsumer extends AbstractConsumer implements ConsumerInterface
+{
+    public function execute(AMQPMessage $msg)
+    {
+        return $this->handleMessage($msg->body);
+    }
+}

src/Consumer/AbstractConsumer.php

@@ -0,0 +1,90 @@
+<?php
+
+namespace Wallabag\CoreBundle\Consumer;
+
+use Doctrine\ORM\EntityManagerInterface;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Wallabag\CoreBundle\Entity\Entry;
+use Wallabag\CoreBundle\Entity\Tag;
+use Wallabag\CoreBundle\Event\EntrySavedEvent;
+use Wallabag\CoreBundle\Import\AbstractImport;
+use Wallabag\CoreBundle\Repository\UserRepository;
+
+abstract class AbstractConsumer
+{
+    protected $em;
+    protected $userRepository;
+    protected $import;
+    protected $eventDispatcher;
+    protected $logger;
+
+    public function __construct(EntityManagerInterface $em, UserRepository $userRepository, AbstractImport $import, EventDispatcherInterface $eventDispatcher, ?LoggerInterface $logger = null)
+    {
+        $this->em = $em;
+        $this->userRepository = $userRepository;
+        $this->import = $import;
+        $this->eventDispatcher = $eventDispatcher;
+        $this->logger = $logger ?: new NullLogger();
+    }
+
+    /**
+     * Handle a message and save it.
+     *
+     * @param string $body Message from the queue (in json)
+     *
+     * @return bool
+     */
+    protected function handleMessage($body)
+    {
+        $storedEntry = json_decode($body, true);
+
+        $user = $this->userRepository->find($storedEntry['userId']);
+
+        // no user? Drop message
+        if (null === $user) {
+            $this->logger->warning('Unable to retrieve user', ['entry' => $storedEntry]);
+
+            // return true to skip message
+            return true;
+        }
+
+        $this->import->setUser($user);
+
+        if (false === $this->import->validateEntry($storedEntry)) {
+            $this->logger->warning('Entry is invalid', ['entry' => $storedEntry]);
+
+            // return true to skip message
+            return true;
+        }
+
+        $entry = $this->import->parseEntry($storedEntry);
+
+        if (null === $entry) {
+            $this->logger->warning('Entry already exists', ['entry' => $storedEntry]);
+
+            // return true to skip message
+            return true;
+        }
+
+        try {
+            $this->em->flush();
+
+            // entry saved, dispatch event about it!
+            $this->eventDispatcher->dispatch(new EntrySavedEvent($entry), EntrySavedEvent::NAME);
+
+            // clear only affected entities
+            $this->em->clear(Entry::class);
+            $this->em->clear(Tag::class);
+        } catch (\Exception $e) {
+            $this->logger->warning('Unable to save entry', ['entry' => $storedEntry, 'exception' => $e]);
+
+            return false;
+        }
+
+        $this->logger->info('Content with url imported! (' . $entry->getUrl() . ')');
+
+        return true;
+    }
+}

src/Consumer/RabbitMQConsumerTotalProxy.php

@@ -0,0 +1,114 @@
+<?php
+
+namespace Wallabag\CoreBundle\Consumer;
+
+use OldSound\RabbitMqBundle\RabbitMq\Consumer;
+
+/**
+ * A proxy class only used to count messages in a queue while lazy loading RabbitMQ services.
+ * Only used in ImportController.
+ */
+class RabbitMQConsumerTotalProxy
+{
+    private Consumer $pocketConsumer;
+    private Consumer $readabilityConsumer;
+    private Consumer $wallabagV1Consumer;
+    private Consumer $wallabagV2Consumer;
+    private Consumer $firefoxConsumer;
+    private Consumer $chromeConsumer;
+    private Consumer $instapaperConsumer;
+    private Consumer $pinboardConsumer;
+    private Consumer $deliciousConsumer;
+    private Consumer $elcuratorConsumer;
+    private Consumer $shaarliConsumer;
+    private Consumer $pocketHtmlConsumer;
+
+    public function __construct(
+        Consumer $pocketConsumer,
+        Consumer $readabilityConsumer,
+        Consumer $wallabagV1Consumer,
+        Consumer $wallabagV2Consumer,
+        Consumer $firefoxConsumer,
+        Consumer $chromeConsumer,
+        Consumer $instapaperConsumer,
+        Consumer $pinboardConsumer,
+        Consumer $deliciousConsumer,
+        Consumer $elcuratorConsumer,
+        Consumer $shaarliConsumer,
+        Consumer $pocketHtmlConsumer
+    ) {
+        $this->pocketConsumer = $pocketConsumer;
+        $this->readabilityConsumer = $readabilityConsumer;
+        $this->wallabagV1Consumer = $wallabagV1Consumer;
+        $this->wallabagV2Consumer = $wallabagV2Consumer;
+        $this->firefoxConsumer = $firefoxConsumer;
+        $this->chromeConsumer = $chromeConsumer;
+        $this->instapaperConsumer = $instapaperConsumer;
+        $this->pinboardConsumer = $pinboardConsumer;
+        $this->deliciousConsumer = $deliciousConsumer;
+        $this->elcuratorConsumer = $elcuratorConsumer;
+        $this->shaarliConsumer = $shaarliConsumer;
+        $this->pocketHtmlConsumer = $pocketHtmlConsumer;
+    }
+
+    /**
+     * Count message in RabbitMQ queue.
+     *
+     * It get one message without acking it (so it'll stay in the queue)
+     * which will include the total of *other* messages in the queue.
+     * Adding one to that messages will result in the full total message.
+     *
+     * @param string $importService The import service related: pocket, readability, wallabag_v1 or wallabag_v2
+     */
+    public function getTotalMessage(string $importService): int
+    {
+        switch ($importService) {
+            case 'pocket':
+                $consumer = $this->pocketConsumer;
+                break;
+            case 'readability':
+                $consumer = $this->readabilityConsumer;
+                break;
+            case 'wallabag_v1':
+                $consumer = $this->wallabagV1Consumer;
+                break;
+            case 'wallabag_v2':
+                $consumer = $this->wallabagV2Consumer;
+                break;
+            case 'firefox':
+                $consumer = $this->firefoxConsumer;
+                break;
+            case 'chrome':
+                $consumer = $this->chromeConsumer;
+                break;
+            case 'instapaper':
+                $consumer = $this->instapaperConsumer;
+                break;
+            case 'pinboard':
+                $consumer = $this->pinboardConsumer;
+                break;
+            case 'delicious':
+                $consumer = $this->deliciousConsumer;
+                break;
+            case 'elcurator':
+                $consumer = $this->elcuratorConsumer;
+                break;
+            case 'shaarli':
+                $consumer = $this->shaarliConsumer;
+                break;
+            case 'pocket_html':
+                $consumer = $this->pocketHtmlConsumer;
+                break;
+            default:
+                return 0;
+        }
+
+        $message = $consumer->getChannel()->basic_get('wallabag.import.' . $importService);
+
+        if (null === $message) {
+            return 0;
+        }
+
+        return $message->delivery_info['message_count'] + 1;
+    }
+}

src/Consumer/RedisEntryConsumer.php

@@ -0,0 +1,46 @@
+<?php
+
+namespace Wallabag\CoreBundle\Consumer;
+
+use Simpleue\Job\Job;
+
+class RedisEntryConsumer extends AbstractConsumer implements Job
+{
+    /**
+     * Handle one message by one message.
+     *
+     * @param string $job Content of the message (directly from Redis)
+     *
+     * @return bool
+     */
+    public function manage($job)
+    {
+        return $this->handleMessage($job);
+    }
+
+    /**
+     * Should tell if the given job will kill the worker.
+     * We don't want to stop it :).
+     *
+     * @param string $job Content of the message (directly from Redis)
+     *
+     * @return false
+     */
+    public function isStopJob($job)
+    {
+        return false;
+    }
+
+    /**
+     * This abstract method is only used when we use one queue for multiple job type.
+     * We don't do that, so we'll always return true.
+     *
+     * @param string $job Content of the message (directly from Redis)
+     *
+     * @return true
+     */
+    public function isMyJob($job)
+    {
+        return true;
+    }
+}