Allow usage of an in-memory message puller

Author: sroze

src/ContinuousPipe/Message/InMemory/ArrayMessagePuller.php

@@ -0,0 +1,32 @@
+<?php
+
+namespace ContinuousPipe\Message\InMemory;
+
+use ContinuousPipe\Message\MessagePuller;
+use ContinuousPipe\Message\PulledMessage;
+
+class ArrayMessagePuller implements MessagePuller
+{
+    /**
+     * @var PulledMessage[]
+     */
+    private $messages;
+
+    /**
+     * @param PulledMessage[] $messages
+     */
+    public function __construct(array $messages = [])
+    {
+        $this->messages = $messages;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function pull(): \Generator
+    {
+        foreach ($this->messages as $message) {
+            yield $message;
+        }
+    }
+}

src/ContinuousPipe/MessageBundle/Resources/config/command.xml

@@ -8,18 +8,12 @@
         <service id="continuouspipe.message.command.pull_and_consumer.throwable_catcher.doctrine" class="ContinuousPipe\Message\AutoRetry\CatchDoctrineDriverException" />
         <service id="continuouspipe.message.command.pull_and_consumer.throwable_catcher" alias="continuouspipe.message.command.pull_and_consumer.throwable_catcher.doctrine" />
         <service id="continuouspipe.message.command.pull_and_consumer" class="ContinuousPipe\MessageBundle\Command\PullAndConsumeMessageCommand">
-            <argument type="service" id="continuouspipe.message.message_poller" />
+            <argument type="service" id="continuouspipe.message.message_puller" />
             <argument type="service" id="continuouspipe.message.message_consumer" />
             <argument type="service" id="continuouspipe.message.command.pull_and_consumer.throwable_catcher" />
             <argument type="service" id="logger" />
 
             <tag name="console.command" />
         </service>
-
-        <service id="continuouspipe.message.command.extend_message_deadline" class="ContinuousPipe\MessageBundle\Command\ExtendMessageDeadlineCommand">
-            <argument type="service" id="continuouspipe.message.google_pub_sub.message_poller"/>
-
-            <tag name="console.command" />
-        </service>
     </services>
 </container>

src/ContinuousPipe/MessageBundle/Resources/config/drivers/direct.xml

@@ -9,6 +9,7 @@
             <argument type="service" id="continuouspipe.message.message_consumer" />
         </service>
 
+        <service id="continuouspipe.message.message_puller" class="ContinuousPipe\Message\InMemory\ArrayMessagePuller" />
         <service id="continuouspipe.message.message_producer" alias="continuouspipe.message.direct.message_producer" />
     </services>
 </container>

src/ContinuousPipe/MessageBundle/Resources/config/drivers/google-pub-sub.xml

@@ -12,7 +12,7 @@
             <argument>%google_pub_sub_topic%</argument>
         </service>
 
-        <service id="continuouspipe.message.google_pub_sub.message_poller" class="ContinuousPipe\Message\GooglePubSub\PubSubMessagePuller">
+        <service id="continuouspipe.message.google_pub_sub.message_puller" class="ContinuousPipe\Message\GooglePubSub\PubSubMessagePuller">
             <argument type="service" id="jms_serializer" />
             <argument type="service" id="logger" />
             <argument>%google_pub_sub_project_id%</argument>
@@ -22,6 +22,12 @@
         </service>
 
         <service id="continuouspipe.message.message_producer" alias="continuouspipe.message.google_pub_sub.message_producer" />
-        <service id="continuouspipe.message.message_poller" alias="continuouspipe.message.google_pub_sub.message_poller" />
+        <service id="continuouspipe.message.message_puller" alias="continuouspipe.message.google_pub_sub.message_puller" />
+
+        <service id="continuouspipe.message.command.extend_message_deadline" class="ContinuousPipe\MessageBundle\Command\ExtendMessageDeadlineCommand">
+            <argument type="service" id="continuouspipe.message.google_pub_sub.message_puller"/>
+
+            <tag name="console.command" />
+        </service>
     </services>
 </container>