Merge pull request #1941 from wallabag/v2-asynchronous-jobs

Use asynchronous jobs for imports

27 files changed, 842 insertions, 330 deletions

diff --git a/src/Wallabag/ImportBundle/Command/ImportCommand.php b/src/Wallabag/ImportBundle/Command/ImportCommand.php
index a4aa8531..20ecc6e1 100644
--- a/src/Wallabag/ImportBundle/Command/ImportCommand.php
+++ b/src/Wallabag/ImportBundle/Command/ImportCommand.php
@@ -26,6 +26,10 @@ class ImportCommand extends ContainerAwareCommand
     {
         $output->writeln('Start : '.(new \DateTime())->format('d-m-Y G:i:s').' ---');
 
+        if (!file_exists($input->getArgument('filepath'))) {
+            throw new Exception(sprintf('File "%s" not found', $input->getArgument('filepath')));
+        }
+
         $em = $this->getContainer()->get('doctrine')->getManager();
         // Turning off doctrine default logs queries for saving memory
         $em->getConnection()->getConfiguration()->setSQLLogger(null);
@@ -43,9 +47,9 @@ class ImportCommand extends ContainerAwareCommand
         }
 
         $wallabag->setMarkAsRead($input->getOption('markAsRead'));
+        $wallabag->setUser($user);
 
         $res = $wallabag
-            ->setUser($user)
             ->setFilepath($input->getArgument('filepath'))
             ->import();
 
diff --git a/src/Wallabag/ImportBundle/Command/RedisWorkerCommand.php b/src/Wallabag/ImportBundle/Command/RedisWorkerCommand.php
new file mode 100644
index 00000000..5f90e00f
--- /dev/null
+++ b/src/Wallabag/ImportBundle/Command/RedisWorkerCommand.php
@@ -0,0 +1,44 @@
+<?php
+
+namespace Wallabag\ImportBundle\Command;
+
+use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
+use Symfony\Component\Config\Definition\Exception\Exception;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Simpleue\Worker\QueueWorker;
+
+class RedisWorkerCommand extends ContainerAwareCommand
+{
+    protected function configure()
+    {
+        $this
+            ->setName('wallabag:import:redis-worker')
+            ->setDescription('Launch Redis worker')
+            ->addArgument('serviceName', InputArgument::REQUIRED, 'Service to use: wallabag_v1, wallabag_v2, pocket or readability')
+            ->addOption('maxIterations', '', InputOption::VALUE_OPTIONAL, 'Number of iterations before stoping', false)
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $output->writeln('Worker started at: '.(new \DateTime())->format('d-m-Y G:i:s'));
+        $output->writeln('Waiting for message ...');
+
+        $serviceName = $input->getArgument('serviceName');
+
+        if (!$this->getContainer()->has('wallabag_import.queue.redis.'.$serviceName) || !$this->getContainer()->has('wallabag_import.consumer.redis.'.$serviceName)) {
+            throw new Exception(sprintf('No queue or consumer found for service name: "%s"', $input->getArgument('serviceName')));
+        }
+
+        $worker = new QueueWorker(
+            $this->getContainer()->get('wallabag_import.queue.redis.'.$serviceName),
+            $this->getContainer()->get('wallabag_import.consumer.redis.'.$serviceName),
+            $input->getOption('maxIterations')
+        );
+
+        $worker->start();
+    }
+}
diff --git a/src/Wallabag/ImportBundle/Consumer/AMQPEntryConsumer.php b/src/Wallabag/ImportBundle/Consumer/AMQPEntryConsumer.php
new file mode 100644
index 00000000..5aaa8c03
--- /dev/null
+++ b/src/Wallabag/ImportBundle/Consumer/AMQPEntryConsumer.php
@@ -0,0 +1,17 @@
+<?php
+
+namespace Wallabag\ImportBundle\Consumer;
+
+use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface;
+use PhpAmqpLib\Message\AMQPMessage;
+
+class AMQPEntryConsumer extends AbstractConsumer implements ConsumerInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function execute(AMQPMessage $msg)
+    {
+        return $this->handleMessage($msg->body);
+    }
+}
diff --git a/src/Wallabag/ImportBundle/Consumer/AbstractConsumer.php b/src/Wallabag/ImportBundle/Consumer/AbstractConsumer.php
new file mode 100644
index 00000000..2b85ad76
--- /dev/null
+++ b/src/Wallabag/ImportBundle/Consumer/AbstractConsumer.php
@@ -0,0 +1,74 @@
+<?php
+
+namespace Wallabag\ImportBundle\Consumer;
+
+use Doctrine\ORM\EntityManager;
+use Wallabag\ImportBundle\Import\AbstractImport;
+use Wallabag\UserBundle\Repository\UserRepository;
+use Wallabag\CoreBundle\Entity\Entry;
+use Wallabag\CoreBundle\Entity\Tag;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+
+abstract class AbstractConsumer
+{
+    protected $em;
+    protected $userRepository;
+    protected $import;
+    protected $logger;
+
+    public function __construct(EntityManager $em, UserRepository $userRepository, AbstractImport $import, LoggerInterface $logger = null)
+    {
+        $this->em = $em;
+        $this->userRepository = $userRepository;
+        $this->import = $import;
+        $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 false;
+        }
+
+        $this->import->setUser($user);
+
+        $entry = $this->import->parseEntry($storedEntry);
+
+        if (null === $entry) {
+            $this->logger->warning('Unable to parse entry', ['entry' => $storedEntry]);
+
+            return false;
+        }
+
+        try {
+            $this->em->flush();
+
+            // 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;
+    }
+}
diff --git a/src/Wallabag/ImportBundle/Consumer/RedisEntryConsumer.php b/src/Wallabag/ImportBundle/Consumer/RedisEntryConsumer.php
new file mode 100644
index 00000000..450b71ff
--- /dev/null
+++ b/src/Wallabag/ImportBundle/Consumer/RedisEntryConsumer.php
@@ -0,0 +1,29 @@
+<?php
+
+namespace Wallabag\ImportBundle\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 :).
+     */
+    public function isStopJob($job)
+    {
+        return false;
+    }
+}
diff --git a/src/Wallabag/ImportBundle/Controller/ImportController.php b/src/Wallabag/ImportBundle/Controller/ImportController.php
index c1486e38..ea4f7d7b 100644
--- a/src/Wallabag/ImportBundle/Controller/ImportController.php
+++ b/src/Wallabag/ImportBundle/Controller/ImportController.php
@@ -16,4 +16,66 @@ class ImportController extends Controller
             'imports' => $this->get('wallabag_import.chain')->getAll(),
         ]);
     }
+
+    /**
+     * Display how many messages are queue (both in Redis and RabbitMQ).
+     * Only for admins.
+     */
+    public function checkQueueAction()
+    {
+        $nbRedisMessages = null;
+        $nbRabbitMessages = null;
+
+        if (!$this->get('security.authorization_checker')->isGranted('ROLE_SUPER_ADMIN')) {
+            return $this->render('WallabagImportBundle:Import:check_queue.html.twig', [
+                'nbRedisMessages' => $nbRedisMessages,
+                'nbRabbitMessages' => $nbRabbitMessages,
+            ]);
+        }
+
+        if ($this->get('craue_config')->get('import_with_rabbitmq')) {
+            $nbRabbitMessages = $this->getTotalMessageInRabbitQueue('pocket')
+                + $this->getTotalMessageInRabbitQueue('readability')
+                + $this->getTotalMessageInRabbitQueue('wallabag_v1')
+                + $this->getTotalMessageInRabbitQueue('wallabag_v2')
+            ;
+        } elseif ($this->get('craue_config')->get('import_with_redis')) {
+            $redis = $this->get('wallabag_core.redis.client');
+
+            $nbRedisMessages = $redis->llen('wallabag.import.pocket')
+                + $redis->llen('wallabag.import.readability')
+                + $redis->llen('wallabag.import.wallabag_v1')
+                + $redis->llen('wallabag.import.wallabag_v2')
+            ;
+        }
+
+        return $this->render('WallabagImportBundle:Import:check_queue.html.twig', [
+            'nbRedisMessages' => $nbRedisMessages,
+            'nbRabbitMessages' => $nbRabbitMessages,
+        ]);
+    }
+
+    /**
+     * 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
+     *
+     * @return int
+     */
+    private function getTotalMessageInRabbitQueue($importService)
+    {
+        $message = $this
+            ->get('old_sound_rabbit_mq.import_'.$importService.'_consumer')
+            ->getChannel()
+            ->basic_get('wallabag.import.'.$importService);
+
+        if (null === $message) {
+            return 0;
+        }
+
+        return $message->delivery_info['message_count'] + 1;
+    }
 }
diff --git a/src/Wallabag/ImportBundle/Controller/PocketController.php b/src/Wallabag/ImportBundle/Controller/PocketController.php
index 36ee25bf..56be5cbf 100644
--- a/src/Wallabag/ImportBundle/Controller/PocketController.php
+++ b/src/Wallabag/ImportBundle/Controller/PocketController.php
@@ -11,11 +11,30 @@ use Symfony\Component\Form\Extension\Core\Type\CheckboxType;
 class PocketController extends Controller
 {
     /**
+     * Return Pocket Import Service with or without RabbitMQ enabled.
+     *
+     * @return \Wallabag\ImportBundle\Import\PocketImport
+     */
+    private function getPocketImportService()
+    {
+        $pocket = $this->get('wallabag_import.pocket.import');
+        $pocket->setUser($this->getUser());
+
+        if ($this->get('craue_config')->get('import_with_rabbitmq')) {
+            $pocket->setProducer($this->get('old_sound_rabbit_mq.import_pocket_producer'));
+        } elseif ($this->get('craue_config')->get('import_with_redis')) {
+            $pocket->setProducer($this->get('wallabag_import.producer.redis.pocket'));
+        }
+
+        return $pocket;
+    }
+
+    /**
      * @Route("/pocket", name="import_pocket")
      */
     public function indexAction()
     {
-        $pocket = $this->get('wallabag_import.pocket.import');
+        $pocket = $this->getPocketImportService();
         $form = $this->createFormBuilder($pocket)
             ->add('mark_as_read', CheckboxType::class, [
                 'label' => 'import.form.mark_as_read_label',
@@ -24,8 +43,8 @@ class PocketController extends Controller
             ->getForm();
 
         return $this->render('WallabagImportBundle:Pocket:index.html.twig', [
-            'import' => $this->get('wallabag_import.pocket.import'),
-            'has_consumer_key' => '' == trim($this->get('craue_config')->get('pocket_consumer_key')) ? false : true,
+            'import' => $this->getPocketImportService(),
+            'has_consumer_key' => '' === trim($this->getUser()->getConfig()->getPocketConsumerKey()) ? false : true,
             'form' => $form->createView(),
         ]);
     }
@@ -35,7 +54,7 @@ class PocketController extends Controller
      */
     public function authAction(Request $request)
     {
-        $requestToken = $this->get('wallabag_import.pocket.import')
+        $requestToken = $this->getPocketImportService()
             ->getRequestToken($this->generateUrl('import', [], UrlGeneratorInterface::ABSOLUTE_URL));
 
         if (false === $requestToken) {
@@ -62,7 +81,7 @@ class PocketController extends Controller
     public function callbackAction()
     {
         $message = 'flashes.import.notice.failed';
-        $pocket = $this->get('wallabag_import.pocket.import');
+        $pocket = $this->getPocketImportService();
 
         $markAsRead = $this->get('session')->get('mark_as_read');
         $this->get('session')->remove('mark_as_read');
@@ -83,6 +102,12 @@ class PocketController extends Controller
                 '%imported%' => $summary['imported'],
                 '%skipped%' => $summary['skipped'],
             ]);
+
+            if (0 < $summary['queued']) {
+                $message = $this->get('translator')->trans('flashes.import.notice.summary_with_queue', [
+                    '%queued%' => $summary['queued'],
+                ]);
+            }
         }
 
         $this->get('session')->getFlashBag()->add(
diff --git a/src/Wallabag/ImportBundle/Controller/ReadabilityController.php b/src/Wallabag/ImportBundle/Controller/ReadabilityController.php
index b61aa99c..d00e22c2 100644
--- a/src/Wallabag/ImportBundle/Controller/ReadabilityController.php
+++ b/src/Wallabag/ImportBundle/Controller/ReadabilityController.php
@@ -18,15 +18,21 @@ class ReadabilityController extends Controller
         $form->handleRequest($request);
 
         $readability = $this->get('wallabag_import.readability.import');
+        $readability->setUser($this->getUser());
+
+        if ($this->get('craue_config')->get('import_with_rabbitmq')) {
+            $readability->setProducer($this->get('old_sound_rabbit_mq.import_readability_producer'));
+        } elseif ($this->get('craue_config')->get('import_with_redis')) {
+            $readability->setProducer($this->get('wallabag_import.producer.redis.readability'));
+        }
 
         if ($form->isValid()) {
             $file = $form->get('file')->getData();
             $markAsRead = $form->get('mark_as_read')->getData();
             $name = 'readability_'.$this->getUser()->getId().'.json';
 
-            if (in_array($file->getClientMimeType(), $this->getParameter('wallabag_import.allow_mimetypes')) && $file->move($this->getParameter('wallabag_import.resource_dir'), $name)) {
+            if (null !== $file && in_array($file->getClientMimeType(), $this->getParameter('wallabag_import.allow_mimetypes')) && $file->move($this->getParameter('wallabag_import.resource_dir'), $name)) {
                 $res = $readability
-                    ->setUser($this->getUser())
                     ->setFilepath($this->getParameter('wallabag_import.resource_dir').'/'.$name)
                     ->setMarkAsRead($markAsRead)
                     ->import();
@@ -40,6 +46,12 @@ class ReadabilityController extends Controller
                         '%skipped%' => $summary['skipped'],
                     ]);
 
+                    if (0 < $summary['queued']) {
+                        $message = $this->get('translator')->trans('flashes.import.notice.summary_with_queue', [
+                            '%queued%' => $summary['queued'],
+                        ]);
+                    }
+
                     unlink($this->getParameter('wallabag_import.resource_dir').'/'.$name);
                 }
 
diff --git a/src/Wallabag/ImportBundle/Controller/WallabagController.php b/src/Wallabag/ImportBundle/Controller/WallabagController.php
index 76ced0d2..9c0cde80 100644
--- a/src/Wallabag/ImportBundle/Controller/WallabagController.php
+++ b/src/Wallabag/ImportBundle/Controller/WallabagController.php
@@ -38,15 +38,15 @@ abstract class WallabagController extends Controller
         $form->handleRequest($request);
 
         $wallabag = $this->getImportService();
+        $wallabag->setUser($this->getUser());
 
         if ($form->isValid()) {
             $file = $form->get('file')->getData();
             $markAsRead = $form->get('mark_as_read')->getData();
             $name = $this->getUser()->getId().'.json';
 
-            if (in_array($file->getClientMimeType(), $this->getParameter('wallabag_import.allow_mimetypes')) && $file->move($this->getParameter('wallabag_import.resource_dir'), $name)) {
+            if (null !== $file && in_array($file->getClientMimeType(), $this->getParameter('wallabag_import.allow_mimetypes')) && $file->move($this->getParameter('wallabag_import.resource_dir'), $name)) {
                 $res = $wallabag
-                    ->setUser($this->getUser())
                     ->setFilepath($this->getParameter('wallabag_import.resource_dir').'/'.$name)
                     ->setMarkAsRead($markAsRead)
                     ->import();
@@ -60,6 +60,12 @@ abstract class WallabagController extends Controller
                         '%skipped%' => $summary['skipped'],
                     ]);
 
+                    if (0 < $summary['queued']) {
+                        $message = $this->get('translator')->trans('flashes.import.notice.summary_with_queue', [
+                            '%queued%' => $summary['queued'],
+                        ]);
+                    }
+
                     unlink($this->getParameter('wallabag_import.resource_dir').'/'.$name);
                 }
 
diff --git a/src/Wallabag/ImportBundle/Controller/WallabagV1Controller.php b/src/Wallabag/ImportBundle/Controller/WallabagV1Controller.php
index 3e748d57..312c7a35 100644
--- a/src/Wallabag/ImportBundle/Controller/WallabagV1Controller.php
+++ b/src/Wallabag/ImportBundle/Controller/WallabagV1Controller.php
@@ -12,7 +12,15 @@ class WallabagV1Controller extends WallabagController
      */
     protected function getImportService()
     {
-        return $this->get('wallabag_import.wallabag_v1.import');
+        $service = $this->get('wallabag_import.wallabag_v1.import');
+
+        if ($this->get('craue_config')->get('import_with_rabbitmq')) {
+            $service->setProducer($this->get('old_sound_rabbit_mq.import_wallabag_v1_producer'));
+        } elseif ($this->get('craue_config')->get('import_with_redis')) {
+            $service->setProducer($this->get('wallabag_import.producer.redis.wallabag_v1'));
+        }
+
+        return $service;
     }
 
     /**
diff --git a/src/Wallabag/ImportBundle/Controller/WallabagV2Controller.php b/src/Wallabag/ImportBundle/Controller/WallabagV2Controller.php
index c2a42165..45211fe6 100644
--- a/src/Wallabag/ImportBundle/Controller/WallabagV2Controller.php
+++ b/src/Wallabag/ImportBundle/Controller/WallabagV2Controller.php
@@ -12,7 +12,15 @@ class WallabagV2Controller extends WallabagController
      */
     protected function getImportService()
     {
-        return $this->get('wallabag_import.wallabag_v2.import');
+        $service = $this->get('wallabag_import.wallabag_v2.import');
+
+        if ($this->get('craue_config')->get('import_with_rabbitmq')) {
+            $service->setProducer($this->get('old_sound_rabbit_mq.import_wallabag_v2_producer'));
+        } elseif ($this->get('craue_config')->get('import_with_redis')) {
+            $service->setProducer($this->get('wallabag_import.producer.redis.wallabag_v2'));
+        }
+
+        return $service;
     }
 
     /**
diff --git a/src/Wallabag/ImportBundle/Form/Type/UploadImportType.php b/src/Wallabag/ImportBundle/Form/Type/UploadImportType.php
index 92a167d9..f50424c1 100644
--- a/src/Wallabag/ImportBundle/Form/Type/UploadImportType.php
+++ b/src/Wallabag/ImportBundle/Form/Type/UploadImportType.php
@@ -15,6 +15,7 @@ class UploadImportType extends AbstractType
         $builder
             ->add('file', FileType::class, [
                 'label' => 'import.form.file_label',
+                'required' => true,
             ])
             ->add('mark_as_read', CheckboxType::class, [
                 'label' => 'import.form.mark_as_read_label',
diff --git a/src/Wallabag/ImportBundle/Import/AbstractImport.php b/src/Wallabag/ImportBundle/Import/AbstractImport.php
index 14377a35..a1a14576 100644
--- a/src/Wallabag/ImportBundle/Import/AbstractImport.php
+++ b/src/Wallabag/ImportBundle/Import/AbstractImport.php
@@ -7,12 +7,21 @@ use Psr\Log\NullLogger;
 use Doctrine\ORM\EntityManager;
 use Wallabag\CoreBundle\Helper\ContentProxy;
 use Wallabag\CoreBundle\Entity\Entry;
+use Wallabag\CoreBundle\Entity\Tag;
+use Wallabag\UserBundle\Entity\User;
+use OldSound\RabbitMqBundle\RabbitMq\ProducerInterface;
 
 abstract class AbstractImport implements ImportInterface
 {
     protected $em;
     protected $logger;
     protected $contentProxy;
+    protected $producer;
+    protected $user;
+    protected $markAsRead;
+    protected $skippedEntries = 0;
+    protected $importedEntries = 0;
+    protected $queuedEntries = 0;
 
     public function __construct(EntityManager $em, ContentProxy $contentProxy)
     {
@@ -27,21 +36,150 @@ abstract class AbstractImport implements ImportInterface
     }
 
     /**
+     * Set RabbitMQ/Redis Producer to send each entry to a queue.
+     * This method should be called when user has enabled RabbitMQ.
+     *
+     * @param ProducerInterface $producer
+     */
+    public function setProducer(ProducerInterface $producer)
+    {
+        $this->producer = $producer;
+    }
+
+    /**
+     * Set current user.
+     * Could the current *connected* user or one retrieve by the consumer.
+     *
+     * @param User $user
+     */
+    public function setUser(User $user)
+    {
+        $this->user = $user;
+    }
+
+    /**
+     * Set whether articles must be all marked as read.
+     *
+     * @param bool $markAsRead
+     */
+    public function setMarkAsRead($markAsRead)
+    {
+        $this->markAsRead = $markAsRead;
+
+        return $this;
+    }
+
+    /**
+     * Get whether articles must be all marked as read.
+     */
+    public function getMarkAsRead()
+    {
+        return $this->markAsRead;
+    }
+
+    /**
      * Fetch content from the ContentProxy (using graby).
-     * If it fails return false instead of the updated entry.
+     * If it fails return the given entry to be saved in all case (to avoid user to loose the content).
      *
      * @param Entry  $entry   Entry to update
      * @param string $url     Url to grab content for
      * @param array  $content An array with AT LEAST keys title, html, url, language & content_type to skip the fetchContent from the url
      *
-     * @return Entry|false
+     * @return Entry
      */
     protected function fetchContent(Entry $entry, $url, array $content = [])
     {
         try {
             return $this->contentProxy->updateEntry($entry, $url, $content);
         } catch (\Exception $e) {
-            return false;
+            return $entry;
+        }
+    }
+
+    /**
+     * Parse and insert all given entries.
+     *
+     * @param $entries
+     */
+    protected function parseEntries($entries)
+    {
+        $i = 1;
+
+        foreach ($entries as $importedEntry) {
+            $entry = $this->parseEntry($importedEntry);
+
+            if (null === $entry) {
+                continue;
+            }
+
+            // flush every 20 entries
+            if (($i % 20) === 0) {
+                $this->em->flush();
+
+                // clear only affected entities
+                $this->em->clear(Entry::class);
+                $this->em->clear(Tag::class);
+            }
+            ++$i;
+        }
+
+        $this->em->flush();
+    }
+
+    /**
+     * Parse entries and send them to the queue.
+     * It should just be a simple loop on all item, no call to the database should be done
+     * to speedup queuing.
+     *
+     * Faster parse entries for Producer.
+     * We don't care to make check at this time. They'll be done by the consumer.
+     *
+     * @param array $entries
+     */
+    protected function parseEntriesForProducer(array $entries)
+    {
+        foreach ($entries as $importedEntry) {
+            // set userId for the producer (it won't know which user is connected)
+            $importedEntry['userId'] = $this->user->getId();
+
+            if ($this->markAsRead) {
+                $importedEntry = $this->setEntryAsRead($importedEntry);
+            }
+
+            ++$this->queuedEntries;
+
+            $this->producer->publish(json_encode($importedEntry));
         }
     }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getSummary()
+    {
+        return [
+            'skipped' => $this->skippedEntries,
+            'imported' => $this->importedEntries,
+            'queued' => $this->queuedEntries,
+        ];
+    }
+
+    /**
+     * Parse one entry.
+     *
+     * @param array $importedEntry
+     *
+     * @return Entry
+     */
+    abstract public function parseEntry(array $importedEntry);
+
+    /**
+     * Set current imported entry to archived / read.
+     * Implementation is different accross all imports.
+     *
+     * @param array $importedEntry
+     *
+     * @return array
+     */
+    abstract protected function setEntryAsRead(array $importedEntry);
 }
diff --git a/src/Wallabag/ImportBundle/Import/PocketImport.php b/src/Wallabag/ImportBundle/Import/PocketImport.php
index a6f905b1..e00eb44b 100644
--- a/src/Wallabag/ImportBundle/Import/PocketImport.php
+++ b/src/Wallabag/ImportBundle/Import/PocketImport.php
@@ -6,31 +6,34 @@ use Psr\Log\NullLogger;
 use Doctrine\ORM\EntityManager;
 use GuzzleHttp\Client;
 use GuzzleHttp\Exception\RequestException;
-use Symfony\Component\Security\Core\Authentication\Token\Storage\TokenStorageInterface;
 use Wallabag\CoreBundle\Entity\Entry;
 use Wallabag\CoreBundle\Helper\ContentProxy;
-use Craue\ConfigBundle\Util\Config;
 
 class PocketImport extends AbstractImport
 {
-    private $user;
     private $client;
-    private $consumerKey;
-    private $skippedEntries = 0;
-    private $importedEntries = 0;
-    private $markAsRead;
-    protected $accessToken;
+    private $accessToken;
 
-    public function __construct(TokenStorageInterface $tokenStorage, EntityManager $em, ContentProxy $contentProxy, Config $craueConfig)
+    const NB_ELEMENTS = 5000;
+
+    public function __construct(EntityManager $em, ContentProxy $contentProxy)
     {
-        $this->user = $tokenStorage->getToken()->getUser();
         $this->em = $em;
         $this->contentProxy = $contentProxy;
-        $this->consumerKey = $craueConfig->get('pocket_consumer_key');
         $this->logger = new NullLogger();
     }
 
     /**
+     * Only used for test purpose.
+     *
+     * @return string
+     */
+    public function getAccessToken()
+    {
+        return $this->accessToken;
+    }
+
+    /**
      * {@inheritdoc}
      */
     public function getName()
@@ -66,7 +69,7 @@ class PocketImport extends AbstractImport
         $request = $this->client->createRequest('POST', 'https://getpocket.com/v3/oauth/request',
             [
                 'body' => json_encode([
-                    'consumer_key' => $this->consumerKey,
+                    'consumer_key' => $this->user->getConfig()->getPocketConsumerKey(),
                     'redirect_uri' => $redirectUri,
                 ]),
             ]
@@ -96,7 +99,7 @@ class PocketImport extends AbstractImport
         $request = $this->client->createRequest('POST', 'https://getpocket.com/v3/oauth/authorize',
             [
                 'body' => json_encode([
-                    'consumer_key' => $this->consumerKey,
+                    'consumer_key' => $this->user->getConfig()->getPocketConsumerKey(),
                     'code' => $code,
                 ]),
             ]
@@ -116,38 +119,22 @@ class PocketImport extends AbstractImport
     }
 
     /**
-     * Set whether articles must be all marked as read.
-     *
-     * @param bool $markAsRead
-     */
-    public function setMarkAsRead($markAsRead)
-    {
-        $this->markAsRead = $markAsRead;
-
-        return $this;
-    }
-
-    /**
-     * Get whether articles must be all marked as read.
-     */
-    public function getMarkAsRead()
-    {
-        return $this->markAsRead;
-    }
-
-    /**
      * {@inheritdoc}
      */
-    public function import()
+    public function import($offset = 0)
     {
+        static $run = 0;
+
         $request = $this->client->createRequest('POST', 'https://getpocket.com/v3/get',
             [
                 'body' => json_encode([
-                    'consumer_key' => $this->consumerKey,
+                    'consumer_key' => $this->user->getConfig()->getPocketConsumerKey(),
                     'access_token' => $this->accessToken,
                     'detailType' => 'complete',
                     'state' => 'all',
-                    'sort' => 'oldest',
+                    'sort' => 'newest',
+                    'count' => self::NB_ELEMENTS,
+                    'offset' => $offset,
                 ]),
             ]
         );
@@ -162,20 +149,24 @@ class PocketImport extends AbstractImport
 
         $entries = $response->json();
 
-        $this->parseEntries($entries['list']);
+        if ($this->producer) {
+            $this->parseEntriesForProducer($entries['list']);
+        } else {
+            $this->parseEntries($entries['list']);
+        }
 
-        return true;
-    }
+        // if we retrieve exactly the amount of items requested it means we can get more
+        // re-call import and offset item by the amount previous received:
+        //  - first call get 5k offset 0
+        //  - second call get 5k offset 5k
+        //  - and so on
+        if (count($entries['list']) === self::NB_ELEMENTS) {
+            ++$run;
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getSummary()
-    {
-        return [
-            'skipped' => $this->skippedEntries,
-            'imported' => $this->importedEntries,
-        ];
+            return $this->import(self::NB_ELEMENTS * $run);
+        }
+
+        return true;
     }
 
     /**
@@ -189,77 +180,74 @@ class PocketImport extends AbstractImport
     }
 
     /**
-     * @see https://getpocket.com/developer/docs/v3/retrieve
+     * {@inheritdoc}
      *
-     * @param $entries
+     * @see https://getpocket.com/developer/docs/v3/retrieve
      */
-    private function parseEntries($entries)
+    public function parseEntry(array $importedEntry)
     {
-        $i = 1;
-
-        foreach ($entries as $pocketEntry) {
-            $url = isset($pocketEntry['resolved_url']) && $pocketEntry['resolved_url'] != '' ? $pocketEntry['resolved_url'] : $pocketEntry['given_url'];
-
-            $existingEntry = $this->em
-                ->getRepository('WallabagCoreBundle:Entry')
-                ->findByUrlAndUserId($url, $this->user->getId());
-
-            if (false !== $existingEntry) {
-                ++$this->skippedEntries;
-                continue;
-            }
-
-            $entry = new Entry($this->user);
-            $entry = $this->fetchContent($entry, $url);
-
-            // jump to next entry in case of problem while getting content
-            if (false === $entry) {
-                ++$this->skippedEntries;
-                continue;
-            }
-
-            // 0, 1, 2 - 1 if the item is archived - 2 if the item should be deleted
-            if ($pocketEntry['status'] == 1 || $this->markAsRead) {
-                $entry->setArchived(true);
-            }
-
-            // 0 or 1 - 1 If the item is starred
-            if ($pocketEntry['favorite'] == 1) {
-                $entry->setStarred(true);
-            }
-
-            $title = 'Untitled';
-            if (isset($pocketEntry['resolved_title']) && $pocketEntry['resolved_title'] != '') {
-                $title = $pocketEntry['resolved_title'];
-            } elseif (isset($pocketEntry['given_title']) && $pocketEntry['given_title'] != '') {
-                $title = $pocketEntry['given_title'];
-            }
-
-            $entry->setTitle($title);
-
-            // 0, 1, or 2 - 1 if the item has images in it - 2 if the item is an image
-            if (isset($pocketEntry['has_image']) && $pocketEntry['has_image'] > 0 && isset($pocketEntry['images'][1])) {
-                $entry->setPreviewPicture($pocketEntry['images'][1]['src']);
-            }
-
-            if (isset($pocketEntry['tags']) && !empty($pocketEntry['tags'])) {
-                $this->contentProxy->assignTagsToEntry(
-                    $entry,
-                    array_keys($pocketEntry['tags'])
-                );
-            }
-
-            $this->em->persist($entry);
-            ++$this->importedEntries;
-
-            // flush every 20 entries
-            if (($i % 20) === 0) {
-                $this->em->flush();
-            }
-            ++$i;
+        $url = isset($importedEntry['resolved_url']) && $importedEntry['resolved_url'] != '' ? $importedEntry['resolved_url'] : $importedEntry['given_url'];
+
+        $existingEntry = $this->em
+            ->getRepository('WallabagCoreBundle:Entry')
+            ->findByUrlAndUserId($url, $this->user->getId());
+
+        if (false !== $existingEntry) {
+            ++$this->skippedEntries;
+
+            return;
+        }
+
+        $entry = new Entry($this->user);
+        $entry->setUrl($url);
+
+        // update entry with content (in case fetching failed, the given entry will be return)
+        $entry = $this->fetchContent($entry, $url);
+
+        // 0, 1, 2 - 1 if the item is archived - 2 if the item should be deleted
+        $entry->setArchived($importedEntry['status'] == 1 || $this->markAsRead);
+
+        // 0 or 1 - 1 If the item is starred
+        $entry->setStarred($importedEntry['favorite'] == 1);
+
+        $title = 'Untitled';
+        if (isset($importedEntry['resolved_title']) && $importedEntry['resolved_title'] != '') {
+            $title = $importedEntry['resolved_title'];
+        } elseif (isset($importedEntry['given_title']) && $importedEntry['given_title'] != '') {
+            $title = $importedEntry['given_title'];
+        }
+
+        $entry->setTitle($title);
+
+        // 0, 1, or 2 - 1 if the item has images in it - 2 if the item is an image
+        if (isset($importedEntry['has_image']) && $importedEntry['has_image'] > 0 && isset($importedEntry['images'][1])) {
+            $entry->setPreviewPicture($importedEntry['images'][1]['src']);
+        }
+
+        if (isset($importedEntry['tags']) && !empty($importedEntry['tags'])) {
+            $this->contentProxy->assignTagsToEntry(
+                $entry,
+                array_keys($importedEntry['tags'])
+            );
         }
 
-        $this->em->flush();
-        $this->em->clear();
+        if (!empty($importedEntry['time_added'])) {
+            $entry->setCreatedAt((new \DateTime())->setTimestamp($importedEntry['time_added']));
+        }
+
+        $this->em->persist($entry);
+        ++$this->importedEntries;
+
+        return $entry;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function setEntryAsRead(array $importedEntry)
+    {
+        $importedEntry['status'] = '1';
+
+        return $importedEntry;
     }
 }
diff --git a/src/Wallabag/ImportBundle/Import/ReadabilityImport.php b/src/Wallabag/ImportBundle/Import/ReadabilityImport.php
index c7cfe15d..fa2b7053 100644
--- a/src/Wallabag/ImportBundle/Import/ReadabilityImport.php
+++ b/src/Wallabag/ImportBundle/Import/ReadabilityImport.php
@@ -3,28 +3,10 @@
 namespace Wallabag\ImportBundle\Import;
 
 use Wallabag\CoreBundle\Entity\Entry;
-use Wallabag\UserBundle\Entity\User;
 
 class ReadabilityImport extends AbstractImport
 {
-    private $user;
-    private $skippedEntries = 0;
-    private $importedEntries = 0;
     private $filepath;
-    private $markAsRead;
-
-    /**
-     * We define the user in a custom call because on the import command there is no logged in user.
-     * So we can't retrieve user from the `security.token_storage` service.
-     *
-     * @param User $user
-     */
-    public function setUser(User $user)
-    {
-        $this->user = $user;
-
-        return $this;
-    }
 
     /**
      * {@inheritdoc}
@@ -63,37 +45,6 @@ class ReadabilityImport extends AbstractImport
     }
 
     /**
-     * Set whether articles must be all marked as read.
-     *
-     * @param bool $markAsRead
-     */
-    public function setMarkAsRead($markAsRead)
-    {
-        $this->markAsRead = $markAsRead;
-
-        return $this;
-    }
-
-    /**
-     * Get whether articles must be all marked as read.
-     */
-    public function getMarkAsRead()
-    {
-        return $this->markAsRead;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function getSummary()
-    {
-        return [
-            'skipped' => $this->skippedEntries,
-            'imported' => $this->importedEntries,
-        ];
-    }
-
-    /**
      * {@inheritdoc}
      */
     public function import()
@@ -116,64 +67,66 @@ class ReadabilityImport extends AbstractImport
             return false;
         }
 
+        if ($this->producer) {
+            $this->parseEntriesForProducer($data['bookmarks']);
+
+            return true;
+        }
+
         $this->parseEntries($data['bookmarks']);
 
         return true;
     }
 
     /**
-     * Parse and insert all given entries.
-     *
-     * @param $entries
+     * {@inheritdoc}
      */
-    protected function parseEntries($entries)
+    public function parseEntry(array $importedEntry)
     {
-        $i = 1;
-
-        foreach ($entries as $importedEntry) {
-            $existingEntry = $this->em
-                ->getRepository('WallabagCoreBundle:Entry')
-                ->findByUrlAndUserId($importedEntry['article__url'], $this->user->getId());
-
-            if (false !== $existingEntry) {
-                ++$this->skippedEntries;
-                continue;
-            }
-
-            $data = [
-                'title' => $importedEntry['article__title'],
-                'url' => $importedEntry['article__url'],
-                'content_type' => '',
-                'language' => '',
-                'is_archived' => $importedEntry['archive'] || $this->markAsRead,
-                'is_starred' => $importedEntry['favorite'],
-            ];
-
-            $entry = $this->fetchContent(
-                new Entry($this->user),
-                $data['url'],
-                $data
-            );
-
-            // jump to next entry in case of problem while getting content
-            if (false === $entry) {
-                ++$this->skippedEntries;
-                continue;
-            }
-            $entry->setArchived($data['is_archived']);
-            $entry->setStarred($data['is_starred']);
-
-            $this->em->persist($entry);
-            ++$this->importedEntries;
-
-            // flush every 20 entries
-            if (($i % 20) === 0) {
-                $this->em->flush();
-            }
-            ++$i;
+        $existingEntry = $this->em
+            ->getRepository('WallabagCoreBundle:Entry')
+            ->findByUrlAndUserId($importedEntry['article__url'], $this->user->getId());
+
+        if (false !== $existingEntry) {
+            ++$this->skippedEntries;
+
+            return;
         }
 
-        $this->em->flush();
-        $this->em->clear();
+        $data = [
+            'title' => $importedEntry['article__title'],
+            'url' => $importedEntry['article__url'],
+            'content_type' => '',
+            'language' => '',
+            'is_archived' => $importedEntry['archive'] || $this->markAsRead,
+            'is_starred' => $importedEntry['favorite'],
+            'created_at' => $importedEntry['date_added'],
+        ];
+
+        $entry = new Entry($this->user);
+        $entry->setUrl($data['url']);
+        $entry->setTitle($data['title']);
+
+        // update entry with content (in case fetching failed, the given entry will be return)
+        $entry = $this->fetchContent($entry, $data['url'], $data);
+
+        $entry->setArchived($data['is_archived']);
+        $entry->setStarred($data['is_starred']);
+        $entry->setCreatedAt(new \DateTime($data['created_at']));
+
+        $this->em->persist($entry);
+        ++$this->importedEntries;
+
+        return $entry;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function setEntryAsRead(array $importedEntry)
+    {
+        $importedEntry['archive'] = 1;
+
+        return $importedEntry;
     }
 }
diff --git a/src/Wallabag/ImportBundle/Import/WallabagImport.php b/src/Wallabag/ImportBundle/Import/WallabagImport.php
index 581ec178..043bb0a2 100644
--- a/src/Wallabag/ImportBundle/Import/WallabagImport.php
+++ b/src/Wallabag/ImportBundle/Import/WallabagImport.php
@@ -3,15 +3,10 @@
 namespace Wallabag\ImportBundle\Import;
 
 use Wallabag\CoreBundle\Entity\Entry;
-use Wallabag\UserBundle\Entity\User;
 
 abstract class WallabagImport extends AbstractImport
 {
-    protected $user;
-    protected $skippedEntries = 0;
-    protected $importedEntries = 0;
     protected $filepath;
-    protected $markAsRead;
     // untitled in all languages from v1
     protected $untitled = [
         'Untitled',
@@ -29,19 +24,6 @@ abstract class WallabagImport extends AbstractImport
     ];
 
     /**
-     * We define the user in a custom call because on the import command there is no logged in user.
-     * So we can't retrieve user from the `security.token_storage` service.
-     *
-     * @param User $user
-     */
-    public function setUser(User $user)
-    {
-        $this->user = $user;
-
-        return $this;
-    }
-
-    /**
      * {@inheritdoc}
      */
     abstract public function getName();
@@ -79,23 +61,18 @@ abstract class WallabagImport extends AbstractImport
             return false;
         }
 
+        if ($this->producer) {
+            $this->parseEntriesForProducer($data);
+
+            return true;
+        }
+
         $this->parseEntries($data);
 
         return true;
     }
 
     /**
-     * {@inheritdoc}
-     */
-    public function getSummary()
-    {
-        return [
-            'skipped' => $this->skippedEntries,
-            'imported' => $this->importedEntries,
-        ];
-    }
-
-    /**
      * Set file path to the json file.
      *
      * @param string $filepath
@@ -108,85 +85,59 @@ abstract class WallabagImport extends AbstractImport
     }
 
     /**
-     * Set whether articles must be all marked as read.
-     *
-     * @param bool $markAsRead
+     * {@inheritdoc}
      */
-    public function setMarkAsRead($markAsRead)
+    public function parseEntry(array $importedEntry)
     {
-        $this->markAsRead = $markAsRead;
+        $existingEntry = $this->em
+            ->getRepository('WallabagCoreBundle:Entry')
+            ->findByUrlAndUserId($importedEntry['url'], $this->user->getId());
 
-        return $this;
-    }
+        if (false !== $existingEntry) {
+            ++$this->skippedEntries;
 
-    /**
-     * Parse and insert all given entries.
-     *
-     * @param $entries
-     */
-    protected function parseEntries($entries)
-    {
-        $i = 1;
+            return;
+        }
 
-        foreach ($entries as $importedEntry) {
-            $existingEntry = $this->em
-                ->getRepository('WallabagCoreBundle:Entry')
-                ->findByUrlAndUserId($importedEntry['url'], $this->user->getId());
+        $data = $this->prepareEntry($importedEntry);
 
-            if (false !== $existingEntry) {
-                ++$this->skippedEntries;
-                continue;
-            }
+        $entry = new Entry($this->user);
+        $entry->setUrl($data['url']);
+        $entry->setTitle($data['title']);
 
-            $data = $this->prepareEntry($importedEntry, $this->markAsRead);
+        // update entry with content (in case fetching failed, the given entry will be return)
+        $entry = $this->fetchContent($entry, $data['url'], $data);
 
-            $entry = $this->fetchContent(
-                new Entry($this->user),
-                $importedEntry['url'],
-                $data
+        if (array_key_exists('tags', $data)) {
+            $this->contentProxy->assignTagsToEntry(
+                $entry,
+                $data['tags']
             );
+        }
 
-            // jump to next entry in case of problem while getting content
-            if (false === $entry) {
-                ++$this->skippedEntries;
-                continue;
-            }
-
-            if (array_key_exists('tags', $data)) {
-                $this->contentProxy->assignTagsToEntry(
-                    $entry,
-                    $data['tags']
-                );
-            }
-
-            if (isset($importedEntry['preview_picture'])) {
-                $entry->setPreviewPicture($importedEntry['preview_picture']);
-            }
-
-            $entry->setArchived($data['is_archived']);
-            $entry->setStarred($data['is_starred']);
-
-            $this->em->persist($entry);
-            ++$this->importedEntries;
-
-            // flush every 20 entries
-            if (($i % 20) === 0) {
-                $this->em->flush();
-            }
-            ++$i;
+        if (isset($importedEntry['preview_picture'])) {
+            $entry->setPreviewPicture($importedEntry['preview_picture']);
         }
 
-        $this->em->flush();
-        $this->em->clear();
+        $entry->setArchived($data['is_archived']);
+        $entry->setStarred($data['is_starred']);
+
+        if (!empty($data['created_at'])) {
+            $entry->setCreatedAt(new \DateTime($data['created_at']));
+        }
+
+        $this->em->persist($entry);
+        ++$this->importedEntries;
+
+        return $entry;
     }
 
     /**
      * This should return a cleaned array for a given entry to be given to `updateEntry`.
      *
-     * @param array $entry      Data from the imported file
-     * @param bool  $markAsRead Should we mark as read content?
+     * @param array $entry Data from the imported file
      *
      * @return array
      */
-    abstract protected function prepareEntry($entry = [], $markAsRead = false);
+    abstract protected function prepareEntry($entry = []);
 }
diff --git a/src/Wallabag/ImportBundle/Import/WallabagV1Import.php b/src/Wallabag/ImportBundle/Import/WallabagV1Import.php
index 6cf3467a..4f001062 100644
--- a/src/Wallabag/ImportBundle/Import/WallabagV1Import.php
+++ b/src/Wallabag/ImportBundle/Import/WallabagV1Import.php
@@ -31,7 +31,7 @@ class WallabagV1Import extends WallabagImport
     /**
      * {@inheritdoc}
      */
-    protected function prepareEntry($entry = [], $markAsRead = false)
+    protected function prepareEntry($entry = [])
     {
         $data = [
             'title' => $entry['title'],
@@ -39,9 +39,10 @@ class WallabagV1Import extends WallabagImport
             'url' => $entry['url'],
             'content_type' => '',
             'language' => '',
-            'is_archived' => $entry['is_read'] || $markAsRead,
+            'is_archived' => $entry['is_read'] || $this->markAsRead,
             'is_starred' => $entry['is_fav'],
             'tags' => '',
+            'created_at' => '',
         ];
 
         // force content to be refreshed in case on bad fetch in the v1 installation
@@ -56,4 +57,14 @@ class WallabagV1Import extends WallabagImport
 
         return $data;
     }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function setEntryAsRead(array $importedEntry)
+    {
+        $importedEntry['is_read'] = 1;
+
+        return $importedEntry;
+    }
 }
diff --git a/src/Wallabag/ImportBundle/Import/WallabagV2Import.php b/src/Wallabag/ImportBundle/Import/WallabagV2Import.php
index d0035b63..37c8ca14 100644
--- a/src/Wallabag/ImportBundle/Import/WallabagV2Import.php
+++ b/src/Wallabag/ImportBundle/Import/WallabagV2Import.php
@@ -31,12 +31,22 @@ class WallabagV2Import extends WallabagImport
     /**
      * {@inheritdoc}
      */
-    protected function prepareEntry($entry = [], $markAsRead = false)
+    protected function prepareEntry($entry = [])
     {
         return [
             'html' => $entry['content'],
             'content_type' => $entry['mimetype'],
-            'is_archived' => ($entry['is_archived'] || $markAsRead),
+            'is_archived' => ($entry['is_archived'] || $this->markAsRead),
         ] + $entry;
     }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function setEntryAsRead(array $importedEntry)
+    {
+        $importedEntry['is_archived'] = 1;
+
+        return $importedEntry;
+    }
 }
diff --git a/src/Wallabag/ImportBundle/Redis/Producer.php b/src/Wallabag/ImportBundle/Redis/Producer.php
new file mode 100644
index 00000000..fedc3e57
--- /dev/null
+++ b/src/Wallabag/ImportBundle/Redis/Producer.php
@@ -0,0 +1,36 @@
+<?php
+
+namespace Wallabag\ImportBundle\Redis;
+
+use OldSound\RabbitMqBundle\RabbitMq\ProducerInterface;
+use Simpleue\Queue\RedisQueue;
+
+/**
+ * This is a proxy class for "Simpleue\Queue\RedisQueue".
+ * It allow us to use the same way to publish a message between RabbitMQ & Redis: publish().
+ *
+ * It implements the ProducerInterface of RabbitMQ (yes it's ugly) so we can have the same
+ * kind of class which implements the same interface.
+ * So we can inject either a RabbitMQ producer or a Redis producer with the same signature
+ */
+class Producer implements ProducerInterface
+{
+    private $queue;
+
+    public function __construct(RedisQueue $queue)
+    {
+        $this->queue = $queue;
+    }
+
+    /**
+     * Publish a message in the Redis queue.
+     *
+     * @param string $msgBody
+     * @param string $routingKey           NOT USED
+     * @param array  $additionalProperties NOT USED
+     */
+    public function publish($msgBody, $routingKey = '', $additionalProperties = array())
+    {
+        $this->queue->sendJob($msgBody);
+    }
+}
diff --git a/src/Wallabag/ImportBundle/Resources/config/rabbit.yml b/src/Wallabag/ImportBundle/Resources/config/rabbit.yml
new file mode 100644
index 00000000..aa049749
--- /dev/null
+++ b/src/Wallabag/ImportBundle/Resources/config/rabbit.yml
@@ -0,0 +1,30 @@
+# RabbitMQ stuff
+services:
+    wallabag_import.consumer.amqp.pocket:
+        class: Wallabag\ImportBundle\Consumer\AMQPEntryConsumer
+        arguments:
+            - "@doctrine.orm.entity_manager"
+            - "@wallabag_user.user_repository"
+            - "@wallabag_import.pocket.import"
+            - "@logger"
+    wallabag_import.consumer.amqp.readability:
+        class: Wallabag\ImportBundle\Consumer\AMQPEntryConsumer
+        arguments:
+            - "@doctrine.orm.entity_manager"
+            - "@wallabag_user.user_repository"
+            - "@wallabag_import.readability.import"
+            - "@logger"
+    wallabag_import.consumer.amqp.wallabag_v1:
+        class: Wallabag\ImportBundle\Consumer\AMQPEntryConsumer
+        arguments:
+            - "@doctrine.orm.entity_manager"
+            - "@wallabag_user.user_repository"
+            - "@wallabag_import.wallabag_v1.import"
+            - "@logger"
+    wallabag_import.consumer.amqp.wallabag_v2:
+        class: Wallabag\ImportBundle\Consumer\AMQPEntryConsumer
+        arguments:
+            - "@doctrine.orm.entity_manager"
+            - "@wallabag_user.user_repository"
+            - "@wallabag_import.wallabag_v2.import"
+            - "@logger"
diff --git a/src/Wallabag/ImportBundle/Resources/config/redis.yml b/src/Wallabag/ImportBundle/Resources/config/redis.yml
new file mode 100644
index 00000000..7d3248e5
--- /dev/null
+++ b/src/Wallabag/ImportBundle/Resources/config/redis.yml
@@ -0,0 +1,81 @@
+# Redis stuff
+services:
+    # readability
+    wallabag_import.queue.redis.readability:
+        class: Simpleue\Queue\RedisQueue
+        arguments:
+            - "@wallabag_core.redis.client"
+            - "wallabag.import.readability"
+
+    wallabag_import.producer.redis.readability:
+        class: Wallabag\ImportBundle\Redis\Producer
+        arguments:
+            - "@wallabag_import.queue.redis.readability"
+
+    wallabag_import.consumer.redis.readability:
+        class: Wallabag\ImportBundle\Consumer\RedisEntryConsumer
+        arguments:
+            - "@doctrine.orm.entity_manager"
+            - "@wallabag_user.user_repository"
+            - "@wallabag_import.readability.import"
+            - "@logger"
+
+    # pocket
+    wallabag_import.queue.redis.pocket:
+        class: Simpleue\Queue\RedisQueue
+        arguments:
+            - "@wallabag_core.redis.client"
+            - "wallabag.import.pocket"
+
+    wallabag_import.producer.redis.pocket:
+        class: Wallabag\ImportBundle\Redis\Producer
+        arguments:
+            - "@wallabag_import.queue.redis.pocket"
+
+    wallabag_import.consumer.redis.pocket:
+        class: Wallabag\ImportBundle\Consumer\RedisEntryConsumer
+        arguments:
+            - "@doctrine.orm.entity_manager"
+            - "@wallabag_user.user_repository"
+            - "@wallabag_import.pocket.import"
+            - "@logger"
+
+    # wallabag v1
+    wallabag_import.queue.redis.wallabag_v1:
+        class: Simpleue\Queue\RedisQueue
+        arguments:
+            - "@wallabag_core.redis.client"
+            - "wallabag.import.wallabag_v1"
+
+    wallabag_import.producer.redis.wallabag_v1:
+        class: Wallabag\ImportBundle\Redis\Producer
+        arguments:
+            - "@wallabag_import.queue.redis.wallabag_v1"
+
+    wallabag_import.consumer.redis.wallabag_v1:
+        class: Wallabag\ImportBundle\Consumer\RedisEntryConsumer
+        arguments:
+            - "@doctrine.orm.entity_manager"
+            - "@wallabag_user.user_repository"
+            - "@wallabag_import.wallabag_v1.import"
+            - "@logger"
+
+    # wallabag v2
+    wallabag_import.queue.redis.wallabag_v2:
+        class: Simpleue\Queue\RedisQueue
+        arguments:
+            - "@wallabag_core.redis.client"
+            - "wallabag.import.wallabag_v2"
+
+    wallabag_import.producer.redis.wallabag_v2:
+        class: Wallabag\ImportBundle\Redis\Producer
+        arguments:
+            - "@wallabag_import.queue.redis.wallabag_v2"
+
+    wallabag_import.consumer.redis.wallabag_v2:
+        class: Wallabag\ImportBundle\Consumer\RedisEntryConsumer
+        arguments:
+            - "@doctrine.orm.entity_manager"
+            - "@wallabag_user.user_repository"
+            - "@wallabag_import.wallabag_v2.import"
+            - "@logger"
diff --git a/src/Wallabag/ImportBundle/Resources/config/services.yml b/src/Wallabag/ImportBundle/Resources/config/services.yml
index 520d43af..f03404ae 100644
--- a/src/Wallabag/ImportBundle/Resources/config/services.yml
+++ b/src/Wallabag/ImportBundle/Resources/config/services.yml
@@ -1,3 +1,7 @@
+imports:
+    - { resource: rabbit.yml }
+    - { resource: redis.yml }
+
 services:
     wallabag_import.chain:
         class: Wallabag\ImportBundle\Import\ImportChain
@@ -14,7 +18,6 @@ services:
     wallabag_import.pocket.import:
         class: Wallabag\ImportBundle\Import\PocketImport
         arguments:
-            - "@security.token_storage"
             - "@doctrine.orm.entity_manager"
             - "@wallabag_core.content_proxy"
             - "@craue_config"
diff --git a/src/Wallabag/ImportBundle/Resources/views/Import/_workerEnabled.html.twig b/src/Wallabag/ImportBundle/Resources/views/Import/_workerEnabled.html.twig
new file mode 100644
index 00000000..2390a41f
--- /dev/null
+++ b/src/Wallabag/ImportBundle/Resources/views/Import/_workerEnabled.html.twig
@@ -0,0 +1,8 @@
+{% set redis = craue_setting('import_with_redis') %}
+{% set rabbit = craue_setting('import_with_rabbitmq') %}
+
+{% if redis or rabbit %}
+    <div class="card-panel yellow darken-1 black-text">
+        {{ 'import.worker.enabled'|trans }} <strong>{% if rabbit %}RabbitMQ{% elseif redis %}Redis{% endif %}</strong>
+    </div>
+{% endif %}
diff --git a/src/Wallabag/ImportBundle/Resources/views/Import/check_queue.html.twig b/src/Wallabag/ImportBundle/Resources/views/Import/check_queue.html.twig
new file mode 100644
index 00000000..7168ea35
--- /dev/null
+++ b/src/Wallabag/ImportBundle/Resources/views/Import/check_queue.html.twig
@@ -0,0 +1,11 @@
+{% if nbRedisMessages > 0 %}
+    <script>
+        Materialize.toast('Messages in queue: {{ nbRedisMessages }}', 4000);
+    </script>
+{% endif %}
+
+{% if nbRabbitMessages > 0 %}
+    <script>
+        Materialize.toast('Messages in queue: {{ nbRabbitMessages }}', 4000);
+    </script>
+{% endif %}
diff --git a/src/Wallabag/ImportBundle/Resources/views/Pocket/index.html.twig b/src/Wallabag/ImportBundle/Resources/views/Pocket/index.html.twig
index 401dbd33..6195fa07 100644
--- a/src/Wallabag/ImportBundle/Resources/views/Pocket/index.html.twig
+++ b/src/Wallabag/ImportBundle/Resources/views/Pocket/index.html.twig
@@ -6,15 +6,13 @@
 <div class="row">
     <div class="col s12">
         <div class="card-panel settings">
+            {% include 'WallabagImportBundle:Import:_workerEnabled.html.twig' %}
+
             {% if not has_consumer_key %}
-                <div class="card-panel red darken-1">
+                <div class="card-panel red white-text">
                     {{ 'import.pocket.config_missing.description'|trans }}
 
-                    {% if is_granted('ROLE_SUPER_ADMIN') %}
-                        {{ 'import.pocket.config_missing.admin_message'|trans({'%keyurls%': '<a href="' ~ path('craue_config_settings_modify') ~ '#set-import">', '%keyurle%':'</a>'})|raw }}
-                    {% else %}
-                        {{ 'import.pocket.config_missing.user_message'|trans }}
-                    {% endif %}
+                    {{ 'import.pocket.config_missing.admin_message'|trans({'%keyurls%': '<a href="' ~ path('config') ~ '">', '%keyurle%':'</a>'})|raw }}
                 </div>
             {% endif %}
 
@@ -29,7 +27,7 @@
                         {{ form_label(form.mark_as_read) }}
                     </div>
                 </div>
-                <button class="btn waves-effect waves-light" type="submit" name="action">
+                <button class="btn waves-effect waves-light" type="submit" name="action" {% if not has_consumer_key %}disabled="disabled"{% endif %}>
                     {{ 'import.pocket.connect_to_pocket'|trans }}
                 </button>
             </form>
diff --git a/src/Wallabag/ImportBundle/Resources/views/Readability/index.html.twig b/src/Wallabag/ImportBundle/Resources/views/Readability/index.html.twig
index f527d309..74653b0f 100644
--- a/src/Wallabag/ImportBundle/Resources/views/Readability/index.html.twig
+++ b/src/Wallabag/ImportBundle/Resources/views/Readability/index.html.twig
@@ -6,6 +6,8 @@
 <div class="row">
     <div class="col s12">
         <div class="card-panel settings">
+            {% include 'WallabagImportBundle:Import:_workerEnabled.html.twig' %}
+
             <div class="row">
                 <blockquote>{{ import.description|trans }}</blockquote>
                 <p>{{ 'import.readability.how_to'|trans }}</p>
diff --git a/src/Wallabag/ImportBundle/Resources/views/WallabagV1/index.html.twig b/src/Wallabag/ImportBundle/Resources/views/WallabagV1/index.html.twig
index 13e24c8c..0b19bc34 100644
--- a/src/Wallabag/ImportBundle/Resources/views/WallabagV1/index.html.twig
+++ b/src/Wallabag/ImportBundle/Resources/views/WallabagV1/index.html.twig
@@ -6,6 +6,8 @@
 <div class="row">
     <div class="col s12">
         <div class="card-panel settings">
+            {% include 'WallabagImportBundle:Import:_workerEnabled.html.twig' %}
+
             <div class="row">
                 <blockquote>{{ import.description|trans }}</blockquote>
                 <p>{{ 'import.wallabag_v1.how_to'|trans }}</p>