Merge pull request #2 from php-enqueue/upd-to-0-3-enqueue

Sync with change in Enqueue 0.3.x version.

Author: makasim

Async/ElasticaPopulateProcessor.php

@@ -1,13 +1,14 @@
 <?php
 namespace Enqueue\ElasticaBundle\Async;
 
-use Enqueue\Psr\Context;
-use Enqueue\Psr\Message;
-use Enqueue\Psr\Processor;
+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 Processor
+class ElasticaPopulateProcessor implements PsrProcessor, QueueSubscriberInterface
 {
     /**
      * @var ProviderRegistry
@@ -25,7 +26,7 @@ public function __construct(ProviderRegistry $providerRegistry)
     /**
      * {@inheritdoc}
      */
-    public function process(Message $message, Context $context)
+    public function process(PsrMessage $message, PsrContext $context)
     {
         if (false == $message->getReplyTo()) {
             return self::REJECT;
@@ -54,11 +55,11 @@ public function process(Message $message, Context $context)
     }
 
     /**
-     * @param Context $context
-     * @param Message $message
+     * @param PsrContext $context
+     * @param PsrMessage $message
      * @param bool $successful
      */
-    private function sendReply(Context $context, Message $message, $successful)
+    private function sendReply(PsrContext $context, PsrMessage $message, $successful)
     {
         $replyMessage = $context->createMessage($message->getBody(), $message->getProperties(), $message->getHeaders());
         $replyMessage->setProperty('fos-populate-successful', (int) $successful);
@@ -67,4 +68,12 @@ private function sendReply(Context $context, Message $message, $successful)
 
         $context->createProducer()->send($replyQueue, $replyMessage);
     }
+
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedQueues()
+    {
+        return ['fos_elastica_populate'];
+    }
 }

Elastica/AsyncDoctrineOrmProvider.php

@@ -1,7 +1,7 @@
 <?php
 namespace Enqueue\ElasticaBundle\Elastica;
 
-use Enqueue\Psr\Context;
+use Enqueue\Psr\PsrContext;
 use Enqueue\Util\JSON;
 use FOS\ElasticaBundle\Doctrine\ORM\Provider;
 
@@ -10,14 +10,14 @@ class AsyncDoctrineOrmProvider extends Provider
     private $batchSize;
 
     /**
-     * @var Context
+     * @var PsrContext
      */
     private $context;
 
     /**
-     * @param Context $context
+     * @param PsrContext $context
      */
-    public function setContext(Context $context)
+    public function setContext(PsrContext $context)
     {
         $this->context = $context;
     }
@@ -42,7 +42,7 @@ protected function doPopulate($options, \Closure $loggerClosure = null)
         $nbObjects = $this->countObjects($queryBuilder);
         $offset = $options['offset'];
 
-        $queue = $this->context->createQueue('fos_elastica.populate');
+        $queue = $this->context->createQueue('fos_elastica_populate');
         $resultQueue = $this->context->createTemporaryQueue();
         $consumer = $this->context->createConsumer($resultQueue);
 

Listener/PurgeFosElasticPopulateQueueListener.php

@@ -1,29 +1,29 @@
 <?php
 namespace Enqueue\ElasticaBundle\Listener;
 
-use Enqueue\Psr\Context;
+use Enqueue\Psr\PsrContext;
 use FOS\ElasticaBundle\Event\IndexPopulateEvent;
 use Symfony\Component\EventDispatcher\EventSubscriberInterface;
 
 class PurgeFosElasticPopulateQueueListener implements EventSubscriberInterface
 {
     /**
-     * @var Context
+     * @var PsrContext
      */
     private $context;
 
     /**
-     * @param Context $context
+     * @param PsrContext $context
      */
-    public function __construct(Context $context)
+    public function __construct(PsrContext $context)
     {
         $this->context = $context;
     }
 
     public function onPreIndexPopulate(IndexPopulateEvent $event)
     {
         if (method_exists($this->context, 'purge')) {
-            $queue = $this->context->createQueue('fos_elastica.populate');
+            $queue = $this->context->createQueue('fos_elastica_populate');
 
             $this->context->purge($queue);
 

README.md

@@ -67,7 +67,7 @@ enqueue:
 ```
 
 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)
-Create a `fos_elastica.populate` queue on broker side, if needed.
+Create a `fos_elastica_populate` queue on broker side, if needed.
 
 ## Usage 
 
@@ -87,7 +87,7 @@ $ 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
 
 ```bash
-$ ./bin/console enqueue:transport:consume fos_elastica.populate enqueue_elastica.populate_processor 
+$ ./bin/console enqueue:transport:consume enqueue_elastica.populate_processor -vv 
 ```
 
 We suggest to use [supervisor](http://supervisord.org/) on production to control numbers of consumers and restart them.   
@@ -97,7 +97,7 @@ Here's config example
 ```
 # cat /etc/supervisor/conf.d/fos_elastica_populate.conf 
 [program:fos_elastica_populate]
-command=/mqs/symfony/bin/console enqueue:transport:consume fos_elastica.populate enqueue_elastica.populate_processor
+command=/mqs/symfony/bin/console enqueue:transport:consume fos_elastica_populate enqueue_elastica.populate_processor
 process_name=%(program_name)s_%(process_num)02d
 numprocs=10
 autostart=true

composer.json

@@ -7,7 +7,8 @@
     "require": {
         "php": ">=5.6",
         "friendsofsymfony/elastica-bundle": "^3",
-        "enqueue/enqueue-bundle": "^0.2"
+        "enqueue/enqueue-bundle": "^0.3.5",
+        "enqueue/enqueue": "^0.3.5"
     },
     "autoload": {
         "psr-4": { "Enqueue\\ElasticaBundle\\": "" }