Merge pull request #16 from graze/feature/purge-queue

Add purge to ProducerInterface

Author: sjparkinson

src/Adapter/AdapterInterface.php

@@ -16,6 +16,7 @@
 
 use Graze\Queue\Adapter\Exception\FailedAcknowledgementException;
 use Graze\Queue\Adapter\Exception\FailedEnqueueException;
+use Graze\Queue\Adapter\Exception\UnsupportedOperationException;
 use Graze\Queue\Message\MessageFactoryInterface;
 use Graze\Queue\Message\MessageInterface;
 use Iterator;
@@ -43,4 +44,6 @@ public function dequeue(MessageFactoryInterface $factory, $limit);
      * @throws FailedEnqueueException
      */
     public function enqueue(array $messages);
+
+    public function purge();
 }

src/Adapter/ArrayAdapter.php

@@ -22,7 +22,7 @@
 final class ArrayAdapter implements AdapterInterface
 {
     /**
-     * @param MessageInterface[]
+     * @param array
      */
     protected $queue = [];
 
@@ -70,6 +70,14 @@ public function enqueue(array $messages)
         }
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function purge()
+    {
+        $this->queue = [];
+    }
+
     /**
      * @param MessageInterface $message
      */

src/Adapter/SqsAdapter.php

@@ -190,6 +190,14 @@ public function enqueue(array $messages)
         }
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function purge()
+    {
+        $this->client->purgeQueue(['QueueUrl' => $this->getQueueUrl()]);
+    }
+
     /**
      * @param MessageInterface[] $messages
      * @return array
@@ -254,7 +262,7 @@ protected function getOption($name, $default = null)
      */
     protected function getQueueUrl()
     {
-        if (!$this->url) {
+        if (! $this->url) {
             $result = $this->client->createQueue([
                 'QueueName' => $this->name,
                 'Attributes' => $this->options,

src/Client.php

@@ -84,6 +84,14 @@ public function send(array $messages)
         return $this->adapter->enqueue($messages);
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function purge()
+    {
+        return $this->adapter->purge();
+    }
+
     /**
      * @return callable
      */

src/ConsumerInterface.php

@@ -19,8 +19,8 @@
 interface ConsumerInterface
 {
     /**
-     * @param callable     $worker
-     * @param integer|null $limit  Integer limit or Null no limit
+     * @param callable $worker
+     * @param integer  $limit
      */
     public function receive(callable $worker, $limit = 1);
 }

src/ProducerInterface.php

@@ -14,17 +14,21 @@
 
 namespace Graze\Queue;
 
+use Graze\Queue\Adapter\Exception\UnsupportedOperationException;
 use Graze\Queue\Message\MessageInterface;
 
 interface ProducerInterface
 {
     /**
-     * @return MessageInterface
+     * @param string $body
+     * @param array  $options
      */
     public function create($body, array $options = []);
 
     /**
-     * @param MessageInterface[] $message
+     * @param array $message
      */
     public function send(array $messages);
+
+    public function purge();
 }

tests/integration/ArrayIntegrationTest.php

@@ -82,4 +82,16 @@ public function testSend()
     {
         $this->client->send([$this->client->create('foo')]);
     }
+
+    public function testPurge()
+    {
+        $this->client->purge();
+
+        $msgs = [];
+        $this->client->receive(function ($msg) use (&$msgs) {
+            $msgs[] = $msg;
+        }, null);
+
+        assertThat($msgs, is(emptyArray()));
+    }
 }

tests/integration/SqsIntegrationTest.php

@@ -202,4 +202,33 @@ public function testSend()
 
         $this->client->send([$this->client->create('foo')]);
     }
+
+    public function testPurge()
+    {
+        $url = $this->stubCreateQueue();
+        $timeout = $this->stubQueueVisibilityTimeout($url);
+
+        $receiveModel = m::mock('Aws\ResultInterface');
+        $receiveModel->shouldReceive('get')->once()->with('Messages')->andReturn([]);
+        $this->sqsClient->shouldReceive('receiveMessage')->once()->with([
+            'QueueUrl' => $url,
+            'AttributeNames' => ['All'],
+            'MaxNumberOfMessages' => 1,
+            'VisibilityTimeout' => $timeout
+        ])->andReturn($receiveModel);
+
+        $purgeModel = m::mock('Aws\ResultInterface');
+        $this->sqsClient->shouldReceive('purgeQueue')->once()->with([
+            'QueueUrl' => $url,
+        ])->andReturn($purgeModel);
+
+        $this->client->purge();
+
+        $msgs = [];
+        $this->client->receive(function ($msg) use (&$msgs) {
+            $msgs[] = $msg;
+        });
+
+        assertThat($msgs, is(emptyArray()));
+    }
 }

tests/unit/Adapter/ArrayAdapterTest.php

@@ -107,4 +107,17 @@ public function testEnqueue()
 
         assertThat(iterator_to_array($iterator), is(identicalTo($merged)));
     }
+
+    public function testPurge()
+    {
+        $iterator = $this->adapter->dequeue($this->factory, 10);
+
+        assertThat(iterator_to_array($iterator), is(nonEmptyArray()));
+
+        $this->adapter->purge();
+
+        $iterator = $this->adapter->dequeue($this->factory, 10);
+
+        assertThat(iterator_to_array($iterator), is(emptyArray()));
+    }
 }

tests/unit/Adapter/SqsAdapterTest.php

@@ -220,4 +220,16 @@ public function testReceiveMessageWaitTimeSecondsOption()
         assertThat($iterator, is(anInstanceOf('Generator')));
         assertThat(iterator_to_array($iterator), is(equalTo($this->messages)));
     }
+
+    public function testPurge()
+    {
+        $adapter = new SqsAdapter($this->client, 'foo');
+        $url = $this->stubCreateQueue('foo');
+
+        $this->client->shouldReceive('purgeQueue')->once()->with([
+            'QueueUrl' => $url,
+        ])->andReturn($this->model);
+
+        assertThat($adapter->purge(), is(nullValue()));
+    }
 }