Merge pull request #220 from tw2066/keepAlive

Support KeepAlive to avoid duplicate connections

Author: isublimity

src/Transport/CurlerRequest.php

@@ -106,7 +106,7 @@ class CurlerRequest
     /**
      * @param bool $id
      */
-    public function __construct($id = false)
+    public function __construct($id = false, $handle = null)
     {
         $this->id = $id;
 
@@ -127,23 +127,7 @@ public function __construct($id = false)
             CURLOPT_RETURNTRANSFER => TRUE,
             CURLOPT_USERAGENT => 'smi2/PHPClickHouse/client',
         );
-    }
-
-    /**
-     *
-     */
-    public function __destruct()
-    {
-        $this->close();
-    }
-
-
-    public function close()
-    {
-        if ($this->handle) {
-            curl_close($this->handle);
-        }
-        $this->handle = null;
+        $this->handle = $handle;
     }
 
     /**
@@ -374,10 +358,7 @@ public function isPersistent()
      */
     public function keepAlive(int $sec = 60)
     {
-        $this->options[CURLOPT_FORBID_REUSE] = TRUE;
-        $this->headers['Connection'] = 'Keep-Alive';
-        $this->headers['Keep-Alive'] = $sec;
-
+        $this->headers['Connection'] = 'keep-alive';
         return $this;
     }
 

src/Transport/CurlerRolling.php

@@ -239,21 +239,16 @@ public function getRunningRequests()
 
     /**
      * @param CurlerRequest $request
-     * @param bool $auto_close
      * @return mixed
      * @throws TransportException
      */
-    public function execOne(CurlerRequest $request, $auto_close = false)
+    public function execOne(CurlerRequest $request)
     {
         $h = $request->handle();
         curl_exec($h);
 
         $request->setResponse($this->makeResponse($h));
 
-        if ($auto_close) {
-            $request->close();
-        }
-
         return $request->response()->http_code();
     }
 

src/Transport/Http.php

@@ -95,6 +95,12 @@ class Http
      * @var null|resource
      */
     private $stdErrOut = null;
+
+    /**
+     * @var null|resource
+     */
+    private $handle = null;
+
     /**
      * Http constructor.
      * @param string $host
@@ -116,6 +122,7 @@ public function __construct($host, $port, $username, $password, $authMethod = nu
         $this->_settings = new Settings();
 
         $this->setCurler();
+        $this->setHandle();
     }
 
 
@@ -235,7 +242,7 @@ private function getUrl($params = []): string
      */
     private function newRequest($extendinfo): CurlerRequest
     {
-        $new = new CurlerRequest();
+        $new = new CurlerRequest(false, $this->handle);
 
         switch ($this->_authMethod) {
             case self::AUTH_METHOD_QUERY_STRING:
@@ -266,7 +273,8 @@ private function newRequest($extendinfo): CurlerRequest
         }
 
         $new->timeOut($this->settings()->getTimeOut());
-        $new->connectTimeOut($this->_connectTimeOut);//->keepAlive(); // one sec
+        $new->connectTimeOut($this->_connectTimeOut);
+        $new->keepAlive();
         $new->verbose(boolval($this->_verbose));
 
         return $new;
@@ -569,7 +577,7 @@ public function getRequestWrite(Query $query): CurlerRequest
      */
     public function ping(): bool
     {
-        $request = new CurlerRequest();
+        $request = new CurlerRequest(false, $this->handle);
         $request->url($this->getUri())->verbose(false)->GET()->connectTimeOut($this->getConnectTimeOut());
         $this->_curler->execOne($request);
 
@@ -748,18 +756,16 @@ private function streaming(Stream $streamRW, CurlerRequest $request): Statement
 
             $request->header('Transfer-Encoding', 'chunked');
 
-
             if ($streamRW->isWrite()) {
                 $request->setReadFunction($callable);
             } else {
                 $request->setWriteFunction($callable);
 
-
 //                $request->setHeaderFunction($callableHead);
             }
 
 
-            $this->_curler->execOne($request, true);
+            $this->_curler->execOne($request);
             $response = new Statement($request);
             if ($response->isError()) {
                 $response->error();
@@ -802,4 +808,17 @@ public function streamWrite(Stream $streamWrite, $sql, $bindings = []): Statemen
         $request = $this->writeStreamData($sql);
         return $this->streaming($streamWrite, $request);
     }
+
+    public function __destruct()
+    {
+        if ($this->handle) {
+            curl_close($this->handle);
+        }
+    }
+
+    public function setHandle()
+    {
+        $this->handle = curl_init();
+    }
+
 }

src/Transport/StreamInsert.php

@@ -61,7 +61,7 @@ public function insert($callback)
             //
             $this->request->header('Transfer-Encoding', 'chunked');
             $this->request->setReadFunction($callback);
-            $this->curlerRolling->execOne($this->request, true);
+            $this->curlerRolling->execOne($this->request);
             $statement = new Statement($this->request);
             $statement->error();
             return $statement;