Merge branch 'Yproximite-feature/iron-mq-4'

k-k · k-k · commit f1adda6cb790 · 2015-12-18T14:01:54.000-07:00
diff --git a/composer.json b/composer.json
@@ -28,7 +28,7 @@
     "require-dev": {
         "phpunit/phpunit":        "~3.7",
         "aws/aws-sdk-php":        "~2.5",
-        "iron-io/iron_mq":        "~2.0",
+        "iron-io/iron_mq":        "^4.0",
         "symfony/finder":         "~2.3",
         "symfony/filesystem":     "~2.3"
     },
diff --git a/docs/iron-mq-provider.rst b/docs/iron-mq-provider.rst
@@ -18,7 +18,7 @@ and needs to have the library included in your ``composer.json`` file.
 
     {
         require: {
-            "iron-io/iron_mq": "~1.5"
+            "iron-io/iron_mq": "^4.0"
         }
     }
 
diff --git a/src/DependencyInjection/Configuration.php b/src/DependencyInjection/Configuration.php
@@ -78,19 +78,17 @@ private function getProvidersNode()
                     ->scalarNode('project_id')->end()
                     ->scalarNode('service')->end()
                     ->enumNode('host')
-                        ->defaultValue('mq-aws-us-east-1')
+                        ->defaultValue('mq-aws-eu-west-1-1')
                         ->values([
-                            'mq-aws-us-east-1',
-                            'mq-aws-eu-west-1',
-                            'mq-rackspace-ord',
-                            'mq-rackspace-lon',
+                            'mq-aws-eu-west-1-1',
+                            'mq-aws-us-east-1-1',
                         ])
                     ->end()
                     ->scalarNode('port')
                         ->defaultValue('443')
                     ->end()
                     ->scalarNode('api_version')
-                        ->defaultValue(1)
+                        ->defaultValue(3)
                     ->end()
                     // AWS
                     ->scalarNode('key')->end()
@@ -145,12 +143,17 @@ private function getQueuesNode()
                                 ->defaultFalse()
                                 ->info('Whether notifications are sent to the subscribers')
                             ->end()
+                            ->scalarNode('push_type')
+                                ->defaultValue('multicast')
+                                ->info('Whether the push queue is multicast or unicast')
+                                ->example('unicast')
+                            ->end()
                             ->scalarNode('notification_retries')
                                 ->defaultValue(3)
                                 ->info('How many attempts the Push Notifications are retried if the Subscriber returns an error')
                                 ->example(3)
                             ->end()
-                            ->scalarNode('notification_retry_delay')
+                            ->scalarNode('notification_retries_delay')
                                 ->defaultValue(60)
                                 ->info('Delay between each Push Notification retry in seconds')
                                 ->example(3)
@@ -180,6 +183,11 @@ private function getQueuesNode()
                                 ->info('How many seconds to Long Poll when requesting messages - if supported')
                                 ->example(3)
                             ->end()
+                            ->scalarNode('rate_limit')
+                                ->defaultValue(-1)
+                                ->info('How many push requests per second will be triggered. -1 for unlimited, 0 disables push')
+                                ->example(1)
+                            ->end()
                             ->append($this->getSubscribersNode())
                         ->end()
                     ->end()
diff --git a/src/Provider/IronMqProvider.php b/src/Provider/IronMqProvider.php
@@ -22,7 +22,7 @@
 
 namespace Uecode\Bundle\QPushBundle\Provider;
 
-use IronMQ;
+use IronMQ\IronMQ;
 use Doctrine\Common\Cache\Cache;
 use Symfony\Bridge\Monolog\Logger;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
@@ -71,10 +71,13 @@ public function create()
     {
         if ($this->options['push_notifications']) {
             $params = [
-                'push_type'     => 'multicast',
-                'retries'       => $this->options['notification_retries'],
-                'retry_delay'   => $this->options['notification_retry_delay'],
-                'subscribers'   => []
+                'type' => $this->options['push_type'],
+                'push'      => [
+                    'rate_limit'    => $this->options['rate_limit'],
+                    'retries'       => $this->options['notification_retries'],
+                    'retries_delay' => $this->options['notification_retries_delay'],
+                    'subscribers'   => []
+                ]
             ];
 
             foreach ($this->options['subscribers'] as $subscriber) {
@@ -84,14 +87,14 @@ public function create()
                     );
                 }
 
-                $params['subscribers'][] = ['url' => $subscriber['endpoint']];
+                $params['push']['subscribers'][] = ['url' => $subscriber['endpoint']];
             }
 
         } else {
             $params = ['push_type' => 'pull'];
         }
 
-        $result = $this->ironmq->updateQueue($this->getNameWithPrefix(), $params);
+        $result = $this->ironmq->createQueue($this->getNameWithPrefix(), $params);
         $this->queue = $result;
 
         $key = $this->getNameWithPrefix();
@@ -301,4 +304,64 @@ public function onMessageReceived(MessageEvent $event)
 
         $event->stopPropagation();
     }
+
+    /**
+     * Get queue info
+     *
+     * This allows to get queue size. Allowing to know if processing is finished or not
+     *
+     * @return stdObject|null
+     */
+    public function queueInfo()
+    {
+        if ($this->queueExists()) {
+            $key = $this->getNameWithPrefix();
+            $this->queue = $this->ironmq->getQueue($key);
+
+            return $this->queue;
+        }
+
+        return null;
+    }
+
+    /**
+     * Publishes multiple message at once
+     *
+     * @param array $messages
+     * @param array $options
+     *
+     * @return array
+     */
+    public function publishMessages(array $messages, array $options = [])
+    {
+        $options      = $this->mergeOptions($options);
+        $publishStart = microtime(true);
+
+        if (!$this->queueExists()) {
+            $this->create();
+        }
+
+        $encodedMessages = [];
+        foreach ($messages as $message) {
+            $encodedMessages[] = json_encode($message + ['_qpush_queue' => $this->name]);
+        }
+
+        $result = $this->ironmq->postMessages(
+            $this->getNameWithPrefix(),
+            $encodedMessages,
+            [
+                'timeout'       => $options['message_timeout'],
+                'delay'         => $options['message_delay'],
+                'expires_in'    => $options['message_expiration']
+            ]
+        );
+
+        $context = [
+            'message_ids'    => $result->ids,
+            'publish_time'  => microtime(true) - $publishStart
+        ];
+        $this->log(200, "Messages have been published.", $context);
+
+        return $result->ids;
+    }
 }
diff --git a/tests/MockClient/IronMqMockClient.php b/tests/MockClient/IronMqMockClient.php
@@ -31,7 +31,7 @@ class IronMqMockClient
 {
     private $deleteCount = 0;
 
-    public function updateQueue($queue, array $params = [])
+    public function createQueue($queue, array $params = [])
     {
         $response                   = new \stdClass;
         $response->id               = '530295fe3c94fbcf0c79cffe';
@@ -97,4 +97,16 @@ public function deleteMessage($queue, $id)
 
         return $response;
     }
+
+    public function getQueue($queue)
+    {
+        $response                   = new \stdClass;
+        $response->id               = '530295fe3c94fbcf0c79cffe';
+        $response->name             = 'test';
+        $response->size             = 0;
+        $response->total_messages   = 0;
+        $response->project_id       = '52f67d032001c00005000057';
+
+        return $response;
+    }
 }
diff --git a/tests/Provider/IronMqProviderTest.php b/tests/Provider/IronMqProviderTest.php
@@ -58,16 +58,18 @@ private function getIronMqProvider(array $options = [])
     {
         $options = array_merge(
             [
-                'logging_enabled'           => false,
-                'push_notifications'        => true,
-                'notification_retries'      => 3,
-                'notification_retry_delay'  => 60,
-                'message_delay'             => 0,
-                'message_timeout'           => 30,
-                'message_expiration'        => 604800,
-                'messages_to_receive'       => 1,
-                'receive_wait_time'         => 3,
-                'subscribers'               => [
+                'logging_enabled'            => false,
+                'push_notifications'         => true,
+                'push_type'                  => 'multicast',
+                'notification_retries'       => 3,
+                'notification_retries_delay' => 60,
+                'message_delay'              => 0,
+                'message_timeout'            => 30,
+                'message_expiration'         => 604800,
+                'messages_to_receive'        => 1,
+                'rate_limit'                 => -1,
+                'receive_wait_time'          => 3,
+                'subscribers'                => [
                     [ 'protocol' => 'http', 'endpoint' => 'http://fake.com' ]
                 ]
             ],
@@ -195,4 +197,15 @@ public function testOnMessageReceived()
             new Message(123, ['foo' => 'bar'], [])
         ));
     }
+
+    public function testQueueInfo()
+    {
+        $this->assertNull($this->provider->queueInfo());
+
+        $this->provider->create();
+        $queue = $this->provider->queueInfo();
+        $this->assertEquals('530295fe3c94fbcf0c79cffe', $queue->id);
+        $this->assertEquals('test', $queue->name);
+        $this->assertEquals('52f67d032001c00005000057', $queue->project_id);
+    }
 }

Original file line number	Diff line number	Diff line change
@@ -18,7 +18,7 @@ and needs to have the library included in your ``composer.json`` file.
`18`	`18`
`19`	`19`	`{`
`20`	`20`	`require: {`
`21`		`- "iron-io/iron_mq": "~1.5"`
	`21`	`+ "iron-io/iron_mq": "^4.0"`
`22`	`22`	`}`
`23`	`23`	`}`
`24`	`24`
Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,7 @@ class IronMqMockClient`
`31`	`31`	`{`
`32`	`32`	`private $deleteCount = 0;`
`33`	`33`
`34`		`- public function updateQueue($queue, array $params = [])`
	`34`	`+ public function createQueue($queue, array $params = [])`
`35`	`35`	`{`
`36`	`36`	`$response = new \stdClass;`
`37`	`37`	`$response->id = '530295fe3c94fbcf0c79cffe';`
`@@ -97,4 +97,16 @@ public function deleteMessage($queue, $id)`
`97`	`97`
`98`	`98`	`return $response;`
`99`	`99`	`}`
	`100`	`+`
	`101`	`+ public function getQueue($queue)`
	`102`	`+ {`
	`103`	`+ $response = new \stdClass;`
	`104`	`+ $response->id = '530295fe3c94fbcf0c79cffe';`
	`105`	`+ $response->name = 'test';`
	`106`	`+ $response->size = 0;`
	`107`	`+ $response->total_messages = 0;`
	`108`	`+ $response->project_id = '52f67d032001c00005000057';`
	`109`	`+`
	`110`	`+ return $response;`
	`111`	`+ }`
`100`	`112`	`}`