Scaling and sending in batches (#92)

* Send notifications in batches

* Update README for scaling

* Fix formatting

Author: Minishlink

README.md

@@ -258,6 +258,14 @@ local public and private keys and compute the shared secret.
 Then, if you have a PHP >= 7.1, WebPush uses `openssl` in order to encrypt the payload with the encryption key.
 Otherwise, if you have PHP < 7.1, it uses [Spomky-Labs/php-aes-gcm](https://github.com/Spomky-Labs/php-aes-gcm), which is slower.
 
+### How do I scale?
+Here are some ideas:
+
+1. Upgrade to PHP 7.1
+2. Make sure MultiCurl is available on your server
+3. Find the right balance for your needs between security and performance (see above)
+4. Find the right batch size (set it in `defaultOptions` or as parameter to `flush()`)
+
 ### How to solve "SSL certificate problem: unable to get local issuer certificate"?
 Your installation lacks some certificates.
 

src/WebPush.php

@@ -29,7 +29,7 @@ class WebPush
     /** @var array Array of array of Notifications */
     private $notifications;
 
-    /** @var array Default options : TTL, urgency, topic */
+    /** @var array Default options : TTL, urgency, topic, batchSize */
     private $defaultOptions;
 
     /** @var int Automatic padding of payloads, if disabled, trade security for bandwidth */
@@ -39,7 +39,7 @@ class WebPush
      * WebPush constructor.
      *
      * @param array               $auth           Some servers needs authentication
-     * @param array               $defaultOptions TTL, urgency, topic
+     * @param array               $defaultOptions TTL, urgency, topic, batchSize
      * @param int|null            $timeout        Timeout of POST request
      * @param array               $clientOptions
      */
@@ -113,54 +113,60 @@ public function sendNotification($endpoint, $payload = null, $userPublicKey = nu
     /**
      * Flush notifications. Triggers the requests.
      *
+     * @param int $batchSize Defaults the value defined in defaultOptions during instanciation (which defaults to 1000).
      * @return array|bool If there are no errors, return true.
      *                    If there were no notifications in the queue, return false.
-     *                    Else return an array of information for each notification sent (success, statusCode, headers, content)
-     *
-     * @throws \ErrorException
+     * Else return an array of information for each notification sent (success, statusCode, headers, content)
      */
-    public function flush()
+    public function flush($batchSize = null)
     {
         if (empty($this->notifications)) {
             return false;
         }
 
-        // for each endpoint server type
-        $requests = $this->prepare($this->notifications);
-        $promises = [];
-        foreach ($requests as $request) {
-            $promises[] = $this->client->sendAsync($request);
+        if (!isset($batchSize)) {
+            $batchSize = $this->defaultOptions['batchSize'];
         }
-        $results = Promise\settle($promises)->wait();
 
+        $batches = array_chunk($this->notifications, $batchSize);
         $return = array();
         $completeSuccess = true;
-        foreach ($results as $result) {
-            if ($result['state'] === "rejected") {
-                /** @var RequestException $reason **/
-                $reason = $result['reason'];
-
-                $error = array(
-                    'success' => false,
-                    'endpoint' => "".$reason->getRequest()->getUri(),
-                    'message' => $reason->getMessage(),
-                );
-
-                $response = $reason->getResponse();
-                if ($response !== null) {
-                    $statusCode = $response->getStatusCode();
-                    $error['statusCode'] = $statusCode;
-                    $error['expired'] = in_array($statusCode, array(400, 404, 410));
-                    $error['content'] = $response->getBody();
-                    $error['headers'] = $response->getHeaders();
+        foreach ($batches as $batch) {
+            // for each endpoint server type
+            $requests = $this->prepare($batch);
+            $promises = [];
+            foreach ($requests as $request) {
+                $promises[] = $this->client->sendAsync($request);
+            }
+            $results = Promise\settle($promises)->wait();
+
+            foreach ($results as $result) {
+                if ($result['state'] === "rejected") {
+                    /** @var RequestException $reason **/
+                    $reason = $result['reason'];
+
+                    $error = array(
+                        'success' => false,
+                        'endpoint' => "".$reason->getRequest()->getUri(),
+                        'message' => $reason->getMessage(),
+                    );
+
+                    $response = $reason->getResponse();
+                    if ($response !== null) {
+                        $statusCode = $response->getStatusCode();
+                        $error['statusCode'] = $statusCode;
+                        $error['expired'] = in_array($statusCode, array(400, 404, 410));
+                        $error['content'] = $response->getBody();
+                        $error['headers'] = $response->getHeaders();
+                    }
+
+                    $return[] = $error;
+                    $completeSuccess = false;
+                } else {
+                    $return[] = array(
+                        'success' => true,
+                    );
                 }
-
-                $return[] = $error;
-                $completeSuccess = false;
-            } else {
-                $return[] = array(
-                    'success' => true,
-                );
             }
         }
 
@@ -293,12 +299,13 @@ public function getDefaultOptions()
     }
 
     /**
-     * @param array $defaultOptions Keys 'TTL' (Time To Live, defaults 4 weeks), 'urgency', and 'topic'
+     * @param array $defaultOptions Keys 'TTL' (Time To Live, defaults 4 weeks), 'urgency', 'topic', 'batchSize'
      */
     public function setDefaultOptions(array $defaultOptions)
     {
         $this->defaultOptions['TTL'] = array_key_exists('TTL', $defaultOptions) ? $defaultOptions['TTL'] : 2419200;
         $this->defaultOptions['urgency'] = array_key_exists('urgency', $defaultOptions) ? $defaultOptions['urgency'] : null;
         $this->defaultOptions['topic'] = array_key_exists('topic', $defaultOptions) ? $defaultOptions['topic'] : null;
+        $this->defaultOptions['batchSize'] = array_key_exists('batchSize', $defaultOptions) ? $defaultOptions['batchSize'] : 1000;
     }
 }

tests/WebPushTest.php

@@ -118,6 +118,23 @@ public function testSendNotification($endpoint, $payload, $userPublicKey, $userA
         $this->assertTrue($res);
     }
 
+    public function testSendNotificationBatch()
+    {
+        $batchSize = 10;
+        $total = 50;
+
+        $notifications = $this->notificationProvider();
+        $notifications = array_fill(0, $total, $notifications[0]);
+
+        foreach ($notifications as $notification) {
+            $this->webPush->sendNotification($notification[0], $notification[1], $notification[2], $notification[3]);
+        }
+
+        $res = $this->webPush->flush($batchSize);
+
+        $this->assertTrue($res);
+    }
+
     public function testSendNotificationWithTooBigPayload()
     {
         $this->setExpectedException('ErrorException', 'Size of payload must not be greater than 4078 octets.');