Merge pull request #3 from php-enqueue/use-enqueue05-features

Upgrade to enqueue 0.5.x. Use its features. Simplify code.

Author: makasim

Async/Commands.php

@@ -0,0 +1,12 @@
+<?php
+namespace Enqueue\ElasticaBundle\Async;
+
+
+class Commands
+{
+    const POPULATE = 'fos_elastica_populate';
+
+    private function __construct()
+    {
+    }
+}

Async/ElasticaPopulateProcessor.php

@@ -1,14 +1,15 @@
 <?php
 namespace Enqueue\ElasticaBundle\Async;
 
+use Enqueue\Client\CommandSubscriberInterface;
 use Enqueue\Consumption\QueueSubscriberInterface;
 use Enqueue\Psr\PsrContext;
 use Enqueue\Psr\PsrMessage;
 use Enqueue\Psr\PsrProcessor;
 use Enqueue\Util\JSON;
 use FOS\ElasticaBundle\Provider\ProviderRegistry;
 
-class ElasticaPopulateProcessor implements PsrProcessor, QueueSubscriberInterface
+class ElasticaPopulateProcessor implements PsrProcessor, CommandSubscriberInterface, QueueSubscriberInterface
 {
     /**
      * @var ProviderRegistry
@@ -69,11 +70,24 @@ private function sendReply(PsrContext $context, PsrMessage $message, $successful
         $context->createProducer()->send($replyQueue, $replyMessage);
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedCommand()
+    {
+        return [
+            'processorName' => Commands::POPULATE,
+            'queueName' => Commands::POPULATE,
+            'queueNameHardcoded' => true,
+            'exclusive' => true,
+        ];
+    }
+
     /**
      * {@inheritdoc}
      */
     public static function getSubscribedQueues()
     {
-        return ['fos_elastica_populate'];
+        return [Commands::POPULATE];
     }
 }

Elastica/AsyncDoctrineOrmProvider.php

@@ -1,25 +1,29 @@
 <?php
 namespace Enqueue\ElasticaBundle\Elastica;
 
-use Enqueue\Psr\PsrContext;
-use Enqueue\Util\JSON;
+use Enqueue\Client\ProducerInterface;
+use Enqueue\ElasticaBundle\Async\Commands;
+use Enqueue\Rpc\Promise;
 use FOS\ElasticaBundle\Doctrine\ORM\Provider;
 
 class AsyncDoctrineOrmProvider extends Provider
 {
+    /**
+     * @var int
+     */
     private $batchSize;
 
     /**
-     * @var PsrContext
+     * @var ProducerInterface
      */
-    private $context;
+    private $producer;
 
     /**
-     * @param PsrContext $context
+     * @param ProducerInterface $producer
      */
-    public function setContext(PsrContext $context)
+    public function setContext(ProducerInterface $producer)
     {
-        $this->context = $context;
+        $this->producer = $producer;
     }
 
     /**
@@ -42,49 +46,41 @@ protected function doPopulate($options, \Closure $loggerClosure = null)
         $nbObjects = $this->countObjects($queryBuilder);
         $offset = $options['offset'];
 
-        $queue = $this->context->createQueue('fos_elastica_populate');
-        $resultQueue = $this->context->createTemporaryQueue();
-        $consumer = $this->context->createConsumer($resultQueue);
-
-        $producer = $this->context->createProducer();
-
-        $nbMessages = 0;
+        /** @var Promise[] $promises */
+        $promises = [];
         for (; $offset < $nbObjects; $offset += $options['batch_size']) {
             $options['offset'] = $offset;
             $options['real_populate'] = true;
-            $message = $this->context->createMessage(JSON::encode($options));
-            $message->setReplyTo($resultQueue->getQueueName());
-            $producer->send($queue, $message);
 
-            $nbMessages++;
+            $promises[] = $this->producer->sendCommand(Commands::POPULATE, $options, true);
         }
 
         $limitTime = time() + 180;
-        while ($nbMessages) {
-            if ($message = $consumer->receive(20000)) {
-                $errorMessage = null;
-
-                $errorMessage = null;
-                if (false == $message->getProperty('fos-populate-successful', false)) {
-                    $errorMessage = sprintf(
-                        '<error>Batch failed: </error> <comment>Failed to process message %s</comment>',
-                        $message->getBody()
-                    );
-                }
-
-                if ($loggerClosure) {
-                    $loggerClosure($options['batch_size'], $nbObjects, $errorMessage);
+        while ($promises) {
+            foreach ($promises as $index => $promise) {
+                if ($message = $promise->receiveNoWait()) {
+                    unset($promises[$index]);
+
+                    $errorMessage = null;
+                    if (false == $message->getProperty('fos-populate-successful', false)) {
+                        $errorMessage = sprintf(
+                            '<error>Batch failed: </error> <comment>Failed to process message %s</comment>',
+                            $message->getBody()
+                        );
+                    }
+
+                    if ($loggerClosure) {
+                        $loggerClosure($options['batch_size'], $nbObjects, $errorMessage);
+                    }
+
+                    $limitTime = time() + 180;
                 }
 
-                $consumer->acknowledge($message);
-
-                $nbMessages--;
+                sleep(1);
 
-                $limitTime = time() + 180;
-            }
-
-            if (time() > $limitTime) {
-                throw new \LogicException(sprintf('No response in %d seconds', 180));
+                if (time() > $limitTime) {
+                    throw new \LogicException(sprintf('No response in %d seconds', 180));
+                }
             }
         }
     }

README.md

@@ -61,9 +61,7 @@ Here's an example of what your EnqueueBundle configuration may look like:
 ```yaml
 enqueue:
     transport:
-        default: 'fs'
-        fs:
-            store_dir: %kernel.root_dir%/../var/messages
+        default: 'file://%kernel.root_dir%/../var/messages'
 ```
 
 Sure you can configure other transports like: [rabbitmq, amqp, stomp and so on](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/bundle/config_reference.md)
@@ -84,10 +82,16 @@ If you want to disable this behavior you can un register the bundle or use env v
 $ ENQUEUE_ELASTICA_DISABLE_ASYNC=1 ./bin/console fos:elastica:populate 
 ```
 
-and have pull of consumer commands run somewhere, run them as many as you'd like
+Run some consumers either using client (you might have to enable it) consume command:
+
+```bash
+$ ./bin/console enqueue:consume --setup-broker -vvv 
+```
+
+or a transport one: 
 
 ```bash
-$ ./bin/console enqueue:transport:consume enqueue_elastica.populate_processor -vv 
+$ ./bin/console enqueue:transport:consume enqueue_elastica.populate_processor -vvv 
 ```
 
 We suggest to use [supervisor](http://supervisord.org/) on production to control numbers of consumers and restart them.   

Resources/config/services.yml

@@ -7,6 +7,6 @@ services:
   enqueue_elastica.purge_fos_elastic_populate_queue_listener:
       class: 'Enqueue\ElasticaBundle\Listener\PurgeFosElasticPopulateQueueListener'
       arguments:
-          - '@enqueue.transport.context'
+          - '@enqueue.client.producer'
       tags:
           -  { name: 'kernel.event_subscriber' }