Merge remote-tracking branch 'origin/master' into async-events

makasim · makasim · commit b18ce0439bba · 2017-05-18T18:02:25.000+03:00
diff --git a/README.md b/README.md
@@ -8,7 +8,7 @@ This is where all development happens. The repository provides a friendly enviro
 Features:
 
 * [JMS](https://docs.oracle.com/javaee/7/api/javax/jms/package-summary.html) like transport [abstraction](https://github.com/php-enqueue/psr-queue).
-* Feature rich.
+* [Feature rich](docs/quick_tour.md).
 * Supports  transports:
     - [AMQP](docs/transport/amqp.md) (RabbitMQ, ActiveMQ and others), 
     - [STOMP](docs/transport/stomp.md)
@@ -18,7 +18,56 @@ Features:
     - [Filesystem](docs/transport/filesystem.md)
     - [Null](docs/transport/null.md).
 * Generic purpose abstraction level (the transport level).
+
+    ```php
+    <?php
+    use function Enqueue\dsn_to_context;
+    use function Enqueue\send_queue;
+    use function Enqueue\consume;
+    use Enqueue\Psr\PsrMessage;
+    use Enqueue\Consumption\Result;
+
+    // composer require enqueue/enqueue enqueue/amqp-ext
+        
+    $c = dsn_to_context('amqp:://');
+  
+    send_queue($c, 'a_queue', 'Hello there');
+       
+    consume($c, 'a_queue', function(PsrMessage $message) {
+       $body = $message->getBody();
+      
+       // to stop consumption: throw new \Enqueue\Consumption\Exception\ConsumptionInterruptedException;  
+      
+       return Result::ACK;
+    });
+    ```
+
 * Easy to use abstraction level (the client level).
+
+    ```php
+    <?php
+    use Enqueue\SimpleClient\SimpleClient;
+    use Enqueue\Psr\PsrMessage;
+    use Enqueue\Consumption\Result;
+    
+    // composer require enqueue/simple-client enqueue/fs
+    
+    $client = new SimpleClient('file://');
+    $client->bind('a_topic', 'a_processor', function(PsrMessage $message) {
+        $body = $message->getBody();
+        
+        // to stop consumption: throw new \Enqueue\Consumption\Exception\ConsumptionInterruptedException;
+        
+        return Result::ACK;
+    });
+    
+    $client->setupBroker();
+    
+    $client->send('a_topic', 'Hello there');
+    
+    $client->consume();
+    ```
+
 * [Symfony bundle](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/bundle/quick_tour.md)
 * [Magento1 extension](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/magento/quick_tour.md)
 * [Message bus](http://www.enterpriseintegrationpatterns.com/patterns/messaging/MessageBus.html) support.
@@ -30,6 +79,7 @@ Features:
 
 ## Resources
 
+* [Quick tour](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/quick_tour.md)
 * [Documentation](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/index.md)
 * [Questions](https://gitter.im/php-enqueue/Lobby)
 * [Issue Tracker](https://github.com/php-enqueue/enqueue-dev/issues)
diff --git a/composer.json b/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"
diff --git a/docs/bundle/message_producer.md b/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)
diff --git a/docs/index.md b/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)
diff --git a/pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php b/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'))
                 ;
             }
 
diff --git a/pkg/enqueue-bundle/README.md b/pkg/enqueue-bundle/README.md
@@ -9,6 +9,7 @@ Integrates message queue components to Symfony application.
 
 ## Resources
 
+* [Quick tour](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/bundle/quick_tour.md)
 * [Documentation](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/index.md)
 * [Questions](https://gitter.im/php-enqueue/Lobby)
 * [Issue Tracker](https://github.com/php-enqueue/enqueue-dev/issues)
diff --git a/pkg/enqueue-bundle/Resources/config/client.yml b/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' }
diff --git a/pkg/enqueue-bundle/Resources/config/extensions/flush_spool_producer_extension.yml b/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 }
diff --git a/pkg/enqueue-bundle/Tests/Functional/Client/SpoolProducerTest.php b/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);
+    }
+}
diff --git a/pkg/enqueue-bundle/Tests/Unit/DependencyInjection/EnqueueExtensionTest.php b/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)
         );
     }
diff --git a/pkg/enqueue/Client/ConsumptionExtension/FlushSpoolProducerExtension.php b/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();
+    }
+}
diff --git a/pkg/enqueue/Client/SpoolProducer.php b/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);
+        }
+    }
+}
diff --git a/pkg/enqueue/Symfony/Client/FlushSpoolProducerListener.php b/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;
+    }
+}
diff --git a/pkg/enqueue/Tests/Client/ConsumptionExtension/FlushSpoolProducerExtensionTest.php b/pkg/enqueue/Tests/Client/ConsumptionExtension/FlushSpoolProducerExtensionTest.php
diff --git a/pkg/enqueue/Tests/Client/SpoolProducerTest.php b/pkg/enqueue/Tests/Client/SpoolProducerTest.php
diff --git a/pkg/enqueue/Tests/Symfony/Client/FlushSpoolProducerListenerTest.php b/pkg/enqueue/Tests/Symfony/Client/FlushSpoolProducerListenerTest.php
diff --git a/pkg/enqueue/composer.json b/pkg/enqueue/composer.json
diff --git a/pkg/enqueue/functions.php b/pkg/enqueue/functions.php
