From 0bf44ffd2851d7c31f69574eeccedf6b7e7a1bbe Mon Sep 17 00:00:00 2001
From: ahfeel <ahfeel@gmail.com>
Date: Tue, 7 Nov 2017 15:59:20 +0100
Subject: [PATCH] * Better concurrency system and memory management by
 unstacking notifications. Based on work of
 ECOMDevelopment:bugfix/connection-limit

---
 src/Client.php | 78 ++++++++++++++++++++++++++++++++------------------
 1 file changed, 50 insertions(+), 28 deletions(-)

diff --git a/src/Client.php b/src/Client.php
index fd236ab..0b8d24b 100644
--- a/src/Client.php
+++ b/src/Client.php
@@ -38,16 +38,24 @@ class Client
      */
     private $isProductionEnv;
 
+    /**
+     * Number of concurrent requests to multiplex in the same connection
+     *
+     * @var bool
+     */
+    private $nbConcurrentRequests;
+
     /**
      * Client constructor.
      *
      * @param AuthProviderInterface $authProvider
      * @param bool $isProductionEnv
      */
-    public function __construct(AuthProviderInterface $authProvider, bool $isProductionEnv = false)
+    public function __construct(AuthProviderInterface $authProvider, bool $isProductionEnv = false, int $nbConcurrentRequests = 10)
     {
         $this->authProvider = $authProvider;
         $this->isProductionEnv = $isProductionEnv;
+        $this->nbConcurrentRequests = $nbConcurrentRequests;
     }
 
     /**
@@ -65,46 +73,60 @@ public function push(): array
 
         curl_multi_setopt($mh, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX);
 
-        $handles = [];
-        foreach ($this->notifications as $k => $notification) {
-            $request = new Request($notification, $this->isProductionEnv);
-            $handles[] = $ch = curl_init();
-
-            $this->authProvider->authenticateClient($request);
-
-            curl_setopt_array($ch, $request->getOptions());
-            curl_setopt($ch, CURLOPT_HTTPHEADER, $request->getDecoratedHeaders());
+        $i = 0;
+        while (!empty($this->notifications) && $i++ < $this->nbConcurrentRequests) {
+            $notification = array_pop($this->notifications);
+            curl_multi_add_handle($mh, $this->prepareHandle($notification));
         }
 
-        $handleChunks = array_chunk($handles, 10);
-        foreach ($handleChunks as $handleChunk) {
-            foreach ($handleChunk as $handle) {
-                curl_multi_add_handle($mh, $handle);   
+        do {
+            while (($execrun = curl_multi_exec($mh, $running)) == CURLM_CALL_MULTI_PERFORM);
+
+            if ($execrun != CURLM_OK) {
+                break;
             }
 
-            $running = null;
-            do {
-                while(($execrun = curl_multi_exec($mh, $running)) == CURLM_CALL_MULTI_PERFORM);
+            while ($done = curl_multi_info_read($mh)) {
+                $handle = $done['handle'];
+
+                $result = curl_multi_getcontent($handle);
 
-                while($done = curl_multi_info_read($mh)) {
-                    $handle = $done['handle'];
+                // find out which token the response is about
+                $token = curl_getinfo($handle, CURLINFO_PRIVATE);
 
-                    $result = curl_multi_getcontent($handle);
-                    $token = curl_getinfo($handle, CURLINFO_PRIVATE);
+                list($headers, $body) = explode("\r\n\r\n", $result, 2);
+                $statusCode = curl_getinfo($handle, CURLINFO_HTTP_CODE);
+                $responseCollection[] = new Response($statusCode, $headers, $body, $token);
+                curl_multi_remove_handle($mh, $handle);
 
-                    list($headers, $body) = explode("\r\n\r\n", $result, 2);
-                    $statusCode = curl_getinfo($handle, CURLINFO_HTTP_CODE);
-                    $responseCollection[] = new Response($statusCode, $headers, $body, $token);
-                    curl_multi_remove_handle($mh, $handle);
+                if (!empty($this->notifications)) {
+                    $notification = array_pop($this->notifications);
+                    curl_multi_add_handle($mh, $this->prepareHandle($notification));
                 }
-            } while ($running);
-        }
+            }
+        } while ($running);
 
         curl_multi_close($mh);
 
         return $responseCollection;
     }
 
+    private function prepareHandle(Notification $notification)
+    {
+        $request = new Request($notification, $this->isProductionEnv);
+        $ch = curl_init();
+
+        $this->authProvider->authenticateClient($request);
+
+        curl_setopt_array($ch, $request->getOptions());
+        curl_setopt($ch, CURLOPT_HTTPHEADER, $request->getDecoratedHeaders());
+
+        // store device token to identify response
+        curl_setopt($ch, CURLOPT_PRIVATE, $notification->getDeviceToken());
+
+        return $ch;
+    }
+
     /**
      * Add notification in queue for sending.
      *
@@ -136,7 +158,7 @@ public function addNotifications(array $notifications)
      *
      * @return Notification[]
      */
-    public function getNotifications()
+    public function getNotifications(): array
     {
         return $this->notifications;
     }