Merge pull request #93 from php-enqueue/client-spool-producer

[client] SpoolProducer

Author: makasim

composer.json

@@ -24,6 +24,8 @@
     "symfony/monolog-bundle": "^2.8|^3",
     "symfony/browser-kit": "^2.8|^3",
     "symfony/expression-language": "^2.8|^3",
+    "symfony/event-dispatcher": "^2.8|^3",
+    "symfony/console": "^2.8|^3",
     "friendsofphp/php-cs-fixer": "^2",
     "empi89/php-amqp-stubs": "*@dev",
     "phpstan/phpstan": "^0.7.0"

docs/bundle/message_producer.md

@@ -0,0 +1,51 @@
+# Message producer
+
+You can choose how to send messages either using a transport directly or with the client. 
+Transport gives you the access to all transport specific features so you can tune things where the client provides you with easy to use abstraction.
+ 
+## Transport
+ 
+```php
+<?php
+
+/** @var Symfony\Component\DependencyInjection\ContainerInterface $container */
+
+/** @var Enqueue\Psr\PsrContext $context */
+$context = $container->get('enqueue.transport.context');
+
+$context->createProducer()->send(
+    $context->createQueue('a_queue'),
+    $context->createMessage('Hello there!')
+);
+```
+
+## Client
+
+The client is shipped with two types of producers. The first one sends messages immediately 
+where another one (it is called spool producer) collects them in memory and sends them `onTerminate` event (the response is already sent).
+
+
+  
+```php
+<?php
+
+/** @var Symfony\Component\DependencyInjection\ContainerInterface $container */
+
+/** @var \Enqueue\Client\ProducerInterface $producer */
+$producer = $container->get('enqueue.producer');
+
+// message is being sent right now
+$producer->send('a_topic', 'Hello there!');
+
+
+/** @var \Enqueue\Client\SpoolProducer $spoolProducer */
+$spoolProducer = $container->get('enqueue.spool_producer');
+
+// message is being sent on console.terminate or kernel.terminate event
+$spoolProducer->send('a_topic', 'Hello there!');
+
+// you could send queued messages manually by calling flush method 
+$spoolProducer->flush();
+```
+
+[back to index](../index.md)

docs/index.md

@@ -25,6 +25,7 @@
     - [Quick tour](bundle/quick_tour.md)
     - [Config reference](bundle/config_reference.md)
     - [Cli commands](bundle/cli_commands.md)
+    - [Message producer](bundle/message_producer.md)
     - [Message processor](bundle/message_processor.md)
     - [Job queue](bundle/job_queue.md)
     - [Consumption extension](bundle/consumption_extension.md)

pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php

@@ -64,6 +64,7 @@ public function load(array $configs, ContainerBuilder $container)
 
         if (isset($config['client'])) {
             $loader->load('client.yml');
+            $loader->load('extensions/flush_spool_producer_extension.yml');
 
             foreach ($config['transport'] as $name => $transportConfig) {
                 $this->factories[$name]->createDriver($container, $transportConfig);
@@ -88,10 +89,10 @@ public function load(array $configs, ContainerBuilder $container)
             $container->setParameter('enqueue.client.default_queue_name', $config['client']['default_processor_queue']);
 
             if (false == empty($config['client']['traceable_producer'])) {
-                $producerId = 'enqueue.client.traceable_message_producer';
+                $producerId = 'enqueue.client.traceable_producer';
                 $container->register($producerId, TraceableProducer::class)
                     ->setDecoratedService('enqueue.client.producer')
-                    ->addArgument(new Reference('enqueue.client.traceable_message_producer.inner'))
+                    ->addArgument(new Reference('enqueue.client.traceable_producer.inner'))
                 ;
             }
 

pkg/enqueue-bundle/Resources/config/client.yml

@@ -9,6 +9,11 @@ services:
             - '@enqueue.client.driver'
             - '@enqueue.client.extensions'
 
+    enqueue.client.spool_producer:
+        class: 'Enqueue\Client\SpoolProducer'
+        arguments:
+            - '@enqueue.client.producer'
+
     enqueue.client.extensions:
         class: 'Enqueue\Client\ChainExtension'
         public: false
@@ -18,6 +23,9 @@ services:
     enqueue.producer:
         alias: 'enqueue.client.producer'
 
+    enqueue.spool_producer:
+        alias: 'enqueue.client.spool_producer'
+
     enqueue.client.rpc_client:
         class: 'Enqueue\Client\RpcClient'
         arguments:
@@ -123,3 +131,10 @@ services:
               name: 'data_collector'
               template: 'EnqueueBundle:Profiler:panel.html.twig'
               id: 'enqueue.message_queue'
+
+    enqueue.flush_spool_producer_listener:
+        class: 'Enqueue\Symfony\Client\FlushSpoolProducerListener'
+        arguments:
+            - '@enqueue.client.spool_producer'
+        tags:
+            - { name: 'kernel.event_subscriber' }

pkg/enqueue-bundle/Resources/config/extensions/flush_spool_producer_extension.yml

@@ -0,0 +1,8 @@
+services:
+    enqueue.client.flush_spool_producer_extension:
+        class: 'Enqueue\Client\ConsumptionExtension\FlushSpoolProducerExtension'
+        public: false
+        arguments:
+            - '@enqueue.client.spool_producer'
+        tags:
+            - { name: 'enqueue.consumption.extension', priority: -100 }

pkg/enqueue-bundle/Tests/Functional/Client/SpoolProducerTest.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace Enqueue\Bundle\Tests\Functional\Client;
+
+use Enqueue\Bundle\Tests\Functional\WebTestCase;
+use Enqueue\Client\SpoolProducer;
+
+/**
+ * @group functional
+ */
+class SpoolProducerTest extends WebTestCase
+{
+    public function testCouldBeGetFromContainerAsService()
+    {
+        $producer = $this->container->get('enqueue.client.spool_producer');
+
+        $this->assertInstanceOf(SpoolProducer::class, $producer);
+    }
+
+    public function testCouldBeGetFromContainerAsShortenAlias()
+    {
+        $producer = $this->container->get('enqueue.client.spool_producer');
+        $aliasProducer = $this->container->get('enqueue.spool_producer');
+
+        $this->assertSame($producer, $aliasProducer);
+    }
+}

pkg/enqueue-bundle/Tests/Unit/DependencyInjection/EnqueueExtensionTest.php

@@ -246,7 +246,7 @@ public function testShouldUseTraceableMessageProducerIfTraceableProducerOptionSe
             ],
         ]], $container);
 
-        $producer = $container->getDefinition('enqueue.client.traceable_message_producer');
+        $producer = $container->getDefinition('enqueue.client.traceable_producer');
         self::assertEquals(TraceableProducer::class, $producer->getClass());
         self::assertEquals(
             ['enqueue.client.producer', null, 0],
@@ -255,7 +255,7 @@ public function testShouldUseTraceableMessageProducerIfTraceableProducerOptionSe
 
         self::assertInstanceOf(Reference::class, $producer->getArgument(0));
         self::assertEquals(
-            'enqueue.client.traceable_message_producer.inner',
+            'enqueue.client.traceable_producer.inner',
             (string) $producer->getArgument(0)
         );
     }

pkg/enqueue/Client/ConsumptionExtension/FlushSpoolProducerExtension.php

@@ -0,0 +1,39 @@
+<?php
+
+namespace Enqueue\Client\ConsumptionExtension;
+
+use Enqueue\Client\SpoolProducer;
+use Enqueue\Consumption\Context;
+use Enqueue\Consumption\EmptyExtensionTrait;
+use Enqueue\Consumption\ExtensionInterface;
+
+class FlushSpoolProducerExtension implements ExtensionInterface
+{
+    use EmptyExtensionTrait;
+
+    /**
+     * @var SpoolProducer
+     */
+    private $producer;
+
+    /**
+     * @param SpoolProducer $producer
+     */
+    public function __construct(SpoolProducer $producer)
+    {
+        $this->producer = $producer;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function onPostReceived(Context $context)
+    {
+        $this->producer->flush();
+    }
+
+    public function onInterrupted(Context $context)
+    {
+        $this->producer->flush();
+    }
+}

pkg/enqueue/Client/SpoolProducer.php

@@ -0,0 +1,46 @@
+<?php
+
+namespace Enqueue\Client;
+
+class SpoolProducer implements ProducerInterface
+{
+    /**
+     * @var ProducerInterface
+     */
+    private $realProducer;
+
+    /**
+     * @var array
+     */
+    private $queue;
+
+    /**
+     * @param ProducerInterface $realProducer
+     */
+    public function __construct(ProducerInterface $realProducer)
+    {
+        $this->realProducer = $realProducer;
+
+        $this->queue = new \SplQueue();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function send($topic, $message)
+    {
+        $this->queue->enqueue([$topic, $message]);
+    }
+
+    /**
+     * When it is called it sends all previously queued messages.
+     */
+    public function flush()
+    {
+        while (false == $this->queue->isEmpty()) {
+            list($topic, $message) = $this->queue->dequeue();
+
+            $this->realProducer->send($topic, $message);
+        }
+    }
+}

pkg/enqueue/Symfony/Client/FlushSpoolProducerListener.php

@@ -0,0 +1,47 @@
+<?php
+
+namespace Enqueue\Symfony\Client;
+
+use Enqueue\Client\SpoolProducer;
+use Symfony\Component\Console\ConsoleEvents;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\HttpKernel\KernelEvents;
+
+class FlushSpoolProducerListener implements EventSubscriberInterface
+{
+    /**
+     * @var SpoolProducer
+     */
+    private $producer;
+
+    /**
+     * @param SpoolProducer $producer
+     */
+    public function __construct(SpoolProducer $producer)
+    {
+        $this->producer = $producer;
+    }
+
+    public function flushMessages()
+    {
+        $this->producer->flush();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedEvents()
+    {
+        $events = [];
+
+        if (class_exists(KernelEvents::class)) {
+            $events[KernelEvents::TERMINATE] = 'flushMessages';
+        }
+
+        if (class_exists(ConsoleEvents::class)) {
+            $events[ConsoleEvents::TERMINATE] = 'flushMessages';
+        }
+
+        return $events;
+    }
+}