1st draft for rabbitMQ

app/AppKernel.php

@@ -38,6 +38,7 @@ class AppKernel extends Kernel
             new Wallabag\UserBundle\WallabagUserBundle(),
             new Wallabag\ImportBundle\WallabagImportBundle(),
             new Wallabag\AnnotationBundle\WallabagAnnotationBundle(),
+            new OldSound\RabbitMqBundle\OldSoundRabbitMqBundle(),
         ];
 
         if (in_array($this->getEnvironment(), ['dev', 'test'], true)) {

app/config/config.yml

@@ -215,3 +215,28 @@ lexik_maintenance:
     response:
         code: 503
         status: "wallabag Service Temporarily Unavailable"
+
+old_sound_rabbit_mq:
+    connections:
+        default:
+            host:     %rabbitmq_host%
+            port:     %rabbitmq_port%
+            user:     %rabbitmq_user%
+            password: %rabbitmq_password%
+            vhost:    /
+            lazy:     false
+    producers:
+        wallabag:
+            connection: default
+            exchange_options:
+                name: 'wallabag_exchange'
+                type: topic
+    consumers:
+        entries:
+            connection: default
+            exchange_options:
+                name: 'wallabag_exchange'
+                type: topic
+            queue_options:
+                name: 'wallabag_queue'
+            callback: wallabag_import.consumer.entry

app/config/parameters.yml.dist

@@ -38,3 +38,15 @@ parameters:
     fosuser_confirmation: true
 
     from_email: no-reply@wallabag.org
+
+    rss_limit: 50
+
+    # pocket import
+    pocket_consumer_key: xxxxxxxx
+
+    # RabbitMQ processing
+    rabbitmq: false
+    rabbitmq_host: localhost
+    rabbitmq_port: 5672
+    rabbitmq_user: guest
+    rabbitmq_password: guest

composer.json

@@ -81,7 +81,8 @@
         "lexik/maintenance-bundle": "~2.1",
         "ocramius/proxy-manager": "1.*",
         "white-october/pagerfanta-bundle": "^1.0",
-        "mouf/nodejs-installer": "~1.0"
+        "mouf/nodejs-installer": "~1.0",
+        "php-amqplib/rabbitmq-bundle": "^1.8"
     },
     "require-dev": {
         "doctrine/doctrine-fixtures-bundle": "~2.2",

docs/en/developer/rabbitmq.rst

@@ -0,0 +1,49 @@
+Install RabbitMQ for asynchronous tasks
+=======================================
+
+In order to launch asynchronous tasks (useful for huge imports for example), we use RabbitMQ.
+
+Requirements
+------------
+
+You need to have RabbitMQ installed on your server.
+
+Installation
+~~~~~~~~~~~~
+
+.. code:: bash
+
+  wget https://www.rabbitmq.com/rabbitmq-signing-key-public.asc
+  apt-key add rabbitmq-signing-key-public.asc
+  apt-get update
+  apt-get install rabbitmq-server
+
+Configuration and launch
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. code:: bash
+
+  rabbitmq-plugins enable rabbitmq_management # (useful to have a web interface, available at http://localhost:15672/ (guest/guest)
+  rabbitmq-server -detached
+
+Stop RabbitMQ
+~~~~~~~~~~~~~
+
+.. code:: bash
+
+  rabbitmqctl stop
+
+
+Configure RabbitMQ in wallabag
+------------------------------
+
+Edit your ``parameters.yml`` file to edit RabbitMQ configuration.
+
+Launch RabbitMQ consumer
+------------------------
+
+Put this command in a cron job:
+
+.. code:: bash
+
+  bin/console rabbitmq:consumer entries -w

src/Wallabag/ImportBundle/Component/AMPQ/EntryConsumer.php

@@ -0,0 +1,39 @@
+<?php
+
+namespace Wallabag\ImportBundle\Component\AMPQ;
+
+use Doctrine\ORM\EntityManager;
+use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface;
+use PhpAmqpLib\Message\AMQPMessage;
+use Wallabag\CoreBundle\Helper\ContentProxy;
+use Wallabag\CoreBundle\Repository\EntryRepository;
+
+class EntryConsumer implements ConsumerInterface
+{
+    private $em;
+    private $contentProxy;
+    private $entryRepository;
+
+    public function __construct(EntityManager $em, EntryRepository $entryRepository, ContentProxy $contentProxy)
+    {
+        $this->em = $em;
+        $this->entryRepository = $entryRepository;
+        $this->contentProxy = $contentProxy;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function execute(AMQPMessage $msg)
+    {
+        $storedEntry = unserialize($msg->body);
+        $entry = $this->entryRepository->findByUrlAndUserId($storedEntry['url'], $storedEntry['userId']);
+        if ($entry) {
+            $entry = $this->contentProxy->updateEntry($entry, $entry->getUrl());
+            if ($entry) {
+                $this->em->persist($entry);
+                $this->em->flush();
+            }
+        }
+    }
+}

src/Wallabag/ImportBundle/Import/PocketImport.php

@@ -2,6 +2,8 @@
 
 namespace Wallabag\ImportBundle\Import;
 
+use OldSound\RabbitMqBundle\RabbitMq\Producer;
+use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use Doctrine\ORM\EntityManager;
 use GuzzleHttp\Client;
@@ -20,14 +22,18 @@ class PocketImport extends AbstractImport
     private $importedEntries = 0;
     private $markAsRead;
     protected $accessToken;
+    private $producer;
+    private $rabbitMQ;
 
-    public function __construct(TokenStorageInterface $tokenStorage, EntityManager $em, ContentProxy $contentProxy, Config $craueConfig)
+    public function __construct(TokenStorageInterface $tokenStorage, EntityManager $em, ContentProxy $contentProxy, Config $craueConfig, $rabbitMQ, Producer $producer)
     {
         $this->user = $tokenStorage->getToken()->getUser();
         $this->em = $em;
         $this->contentProxy = $contentProxy;
         $this->consumerKey = $craueConfig->get('pocket_consumer_key');
         $this->logger = new NullLogger();
+        $this->rabbitMQ = $rabbitMQ;
+        $this->producer = $producer;
     }
 
     /**
@@ -197,7 +203,7 @@ class PocketImport extends AbstractImport
     {
         $i = 1;
 
-        foreach ($entries as $pocketEntry) {
+        foreach ($entries as &$pocketEntry) {
             $url = isset($pocketEntry['resolved_url']) && $pocketEntry['resolved_url'] != '' ? $pocketEntry['resolved_url'] : $pocketEntry['given_url'];
 
             $existingEntry = $this->em
@@ -210,6 +216,8 @@ class PocketImport extends AbstractImport
             }
 
             $entry = new Entry($this->user);
+
+            if (!$this->rabbitMQ) {
                 $entry = $this->fetchContent($entry, $url);
 
                 // jump to next entry in case of problem while getting content
@@ -217,6 +225,7 @@ class PocketImport extends AbstractImport
                     ++$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) {
@@ -236,6 +245,7 @@ class PocketImport extends AbstractImport
             }
 
             $entry->setTitle($title);
+            $entry->setUrl($url);
 
             // 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])) {
@@ -249,6 +259,9 @@ class PocketImport extends AbstractImport
                 );
             }
 
+            $pocketEntry['url'] = $url;
+            $pocketEntry['userId'] = $this->user->getId();
+
             $this->em->persist($entry);
             ++$this->importedEntries;
 
@@ -256,10 +269,16 @@ class PocketImport extends AbstractImport
             if (($i % 20) === 0) {
                 $this->em->flush();
             }
+
             ++$i;
         }
 
         $this->em->flush();
-        $this->em->clear();
+
+        if ($this->rabbitMQ) {
+            foreach ($entries as $entry) {
+                $this->producer->publish(serialize($entry));
+            }
+        }
     }
 }

src/Wallabag/ImportBundle/Resources/config/services.yml

@@ -1,4 +1,11 @@
 services:
+    wallabag_import.consumer.entry:
+        class: Wallabag\ImportBundle\Component\AMPQ\EntryConsumer
+        arguments:
+            - "@doctrine.orm.entity_manager"
+            - "@wallabag_core.entry_repository"
+            - "@wallabag_core.content_proxy"
+
     wallabag_import.chain:
         class: Wallabag\ImportBundle\Import\ImportChain
 
@@ -18,6 +25,8 @@ services:
             - "@doctrine.orm.entity_manager"
             - "@wallabag_core.content_proxy"
             - "@craue_config"
+            - %rabbitmq%
+            - "@old_sound_rabbit_mq.wallabag_producer"
         calls:
             - [ setClient, [ "@wallabag_import.pocket.client" ] ]
             - [ setLogger, [ "@logger" ]]

tests/Wallabag/ImportBundle/Import/PocketImportTest.php

@@ -27,8 +27,9 @@ class PocketImportTest extends \PHPUnit_Framework_TestCase
     protected $em;
     protected $contentProxy;
     protected $logHandler;
+    protected $producer;
 
-    private function getPocketImport($consumerKey = 'ConsumerKey')
+    private function getPocketImport($consumerKey = 'ConsumerKey', $rabbitMQ = false)
     {
         $this->user = new User();
 
@@ -65,11 +66,17 @@ class PocketImportTest extends \PHPUnit_Framework_TestCase
             ->with('pocket_consumer_key')
             ->willReturn($consumerKey);
 
+        $this->producer = $this->getMockBuilder('OldSound\RabbitMqBundle\RabbitMq\Producer')
+            ->disableOriginalConstructor()
+            ->getMock();
+
         $pocket = new PocketImportMock(
             $this->tokenStorage,
             $this->em,
             $this->contentProxy,
-            $config
+            $config,
+            $rabbitMQ,
+            $this->producer
         );
 
         $this->logHandler = new TestHandler();