bind processors

Author: ASKozienko

shell/enqueue.php

@@ -20,15 +20,14 @@
 // init
 Mage::app('admin', 'store');
 
-/** @var \Enqueue\Client\SimpleClient $client */
-$client = Mage::helper('enqueue')->getClient();
+/** @var \Enqueue_Enqueue_Helper_Data $enqueue */
+$enqueue = Mage::helper('enqueue');
+$enqueue->bindProcessors();
 
-//bind message processor
-//$helper = Mage::helper('helper-name');
-//$client->bind('topic', 'consumer-name', [$helper, 'method-name']);
+/** @var \Enqueue\Client\SimpleClient $client */
+$client = $enqueue->getClient();
 
 $application = new Application();
-
 $application->add(new SetupBrokerCommand($client->getDriver()));
 $application->add(new ProduceMessageCommand($client->getProducer()));
 $application->add(new QueuesCommand($client->getQueueMetaRegistry()));

src/Enqueue/Enqueue/Helper/Data.php

@@ -1,12 +1,41 @@
 <?php
 
+use Enqueue\Psr\PsrProcessor;
+
 class Enqueue_Enqueue_Helper_Data extends Mage_Core_Helper_Data
 {
     /**
      * @var \Enqueue\Client\SimpleClient
      */
     private $client;
 
+    public function bindProcessors()
+    {
+        if (false == $processors = Mage::getStoreConfig('enqueue/processors')) {
+            return;
+        }
+
+        foreach ($processors as $name => $config) {
+            if (empty($config['topic'])) {
+                throw new \LogicException('Topic name is not set for processor: "%s"', $name);
+            }
+
+            if (empty($config['helper'])) {
+                throw new \LogicException('Helper name is not set for processor: "%s"', $name);
+            }
+
+            $this->getClient()->bind($config['topic'], $name, function () use ($config) {
+                $processor = Mage::helper($config['helper']);
+
+                if (false == $processor instanceof PsrProcessor) {
+                    throw new \LogicException();
+                }
+
+                call_user_func_array([$processor, 'process'], func_get_args());
+            });
+        }
+    }
+
     public function getClient()
     {
         if (null === $this->client) {